celery-celery-10204

__all__ = ('asynloop', 'synloop')

# pylint: disable=redefined-outer-name
# We cache globals and attribute lookups, so disable this warning.

logger = get_logger(__name__)


def _quick_drain(connection, timeout=0.1):
    try:
        connection.drain_events(timeout=timeout)
    except Exception as exc:  # pylint: disable=broad-except
        exc_errno = getattr(exc, 'errno', None)
        if exc_errno is not None and exc_errno != errno.EAGAIN:
            raise


def _enable_amqheartbeats(timer, connection, rate=2.0):
    heartbeat_error = [None]

    if not connection:
 
…[3270 chars omitted]…

            hub.reset()
        except Exception as exc:  # pylint: disable=broad-except
            logger.exception(
                'Error cleaning up after event loop: %r', exc)
        # Clear stale timer entries accumulated across reconnects (e.g.
        # maybe_restore_messages registered via call_repeatedly). Without
        # this, each reconnect appends a new entry; all of them fire during

def synloop(obj, connection, consumer, blueprint, hub, qos,
            heartbeat, clock, hbrate=2.0, **kwargs):
    """Fallback blocking event loop for transports that doesn't support AIO."""
    RUN = bootsteps.RUN
    on_task_received = obj.create_task_handler()
    perform_pending_operations = obj.perform
…[485 chars omitted]…
           qos.update()
        try:
            perform_pending_operations()
            connection.drain_events(timeout=2.0)
        except socket.timeout:
            pass
        except OSError:
            if blueprint.state == RUN:
                raise

    while blueprint.state == RUN and obj.connection:
        try:
            state.maybe_shutdown()
        finally:
            _loop_cycle()

76-            Mock(), timer=Mock(), controller=Mock(), app=app)
--
130:    asynloop(*x.args)
131-    return x, x.consumer.on_message
132-
133-
134:class test_asynloop:
135-
136-    def setup_method(self):
137-        @self.app.task(shared=False)
138-        def add(x, y):
139-            return x + y
--
151:            asynloop(*x.args)
152-
153-    def test_setup_heartbeat(self):
154-        x = X(self.app, heartbeat=10)
155-        x.hub.timer.call_repeatedly = Mock(name='x.hub.call_repea
…[2853 chars omitted]…
_reraises_original(self):
596:    def test_timeout_ignored(self):
603:    def test_updates_qos_when_changed(self):
618:    def test_ignores_socket_errors_when_closed(self):
623:    def test_no_connection(self):
634:    def test_heartbeat_error(self):
651:    def test_no_heartbeat_support(self):
672:    def test_drain(self):
676:    def test_drain_error(self):
683:    def test_drain_error_EAGAIN(self):

    def test_updates_qos_when_changed(self):
        x = X(self.app)
        x.qos.prev = 2
        x.qos.value = 2
        x.timeout_then_error(x.connection.drain_events)
        with pytest.raises(socket.error):
            synloop(*x.args)
        x.qos.update.assert_not_called()

        x.qos.value = 4
        x.timeout_then_error(x.connection.drain_events)
        with pytest.raises(socket.error):
            synloop(*x.args)
        x.qos.update.assert_called_with()

    def test_ignores_socket_errors_when_closed(self):
        x = X(self.app)
 
…[1663 chars omitted]…
:
        exc = KeyError()
        exc.errno = 313
        self.connection.drain_events.side_effect = exc
        with pytest.raises(KeyError):
            _quick_drain(self.connection, timeout=33.3)

    def test_drain_error_EAGAIN(self):
        exc = KeyError()
        exc.errno = errno.EAGAIN
        self.connection.drain_events.side_effect = exc
        _quick_drain(self.connection, timeout=33.3)

class X:

    def __init__(self, app, heartbeat=None, on_task_message=None,
                 transport_driver_type=None):
        hub = Hub()
        (
            self.obj,
            self.connection,
            self.consumer,
            self.blueprint,
            self.hub,
            self.qos,
            self.heartbeat,
            self.clock,
        ) = self.args = [Mock(name='obj'),
                         Mock(name='connection'),
                         Mock(name='consumer'),
                
…[2662 chars omitted]…
):
        mock = Mock() if mock is None else mock

        def closing(*args, **kwargs):
            if not mod or mock.call_count >= mod:
                self.close()
        mock.side_effect = closing
        return mock


def get_task_callback(*args, **kwargs):
    x = X(*args, **kwargs)
    x.blueprint.state = CLOSE
    asynloop(*x.args)
    return x, x.consumer.on_message


class test_asynloop:

            connection.drain_events(timeout=2.0)
        except socket.timeout:
            pass
        except OSError:
            if blueprint.state == RUN:
                raise

    def _reset_hub():
        # Reset the hub on connection errors (e.g. when using the gevent or
        # eventlet pool) so that stale file descriptors, callbacks and timer
        # entries from the old connection are not reused after reconnect.
        # All hub state is re-registered when the consumer restarts and
        # blueprint.start() is called again.  We intentionally do NOT reset
…[562 chars omitted]…
lean up hub state when exiting due to a connection
            # error.  Graceful shutdowns (WorkerShutdown/WorkerTerminate)
            # extend SystemExit and are not caught here, and a normal exit
            # does not raise at all.
            errors = getattr(connection, 'connection_errors', None) or ()
            if isinstance(exc, tuple(errors)):
                _reset_hub()
            raise