def __init__(self, results, app=None, ready_barrier=None, **kwargs):
     self._app = app
     self.results = results
     self.on_ready = promise(args=(self,))
     self._on_full = ready_barrier or barrier(results)
     if self._on_full:
         self._on_full.then(promise(self._on_ready, weak=True))

 def on_task_received(message):
     # payload will only be set for v1 protocol, since v2
     # will defer deserializing the message body to the pool.
     payload = None
     try:
         type_ = message.headers['task']                # protocol v2
     except TypeError:
         return on_unknown_message(None, message)
     except KeyError:
         try:
             payload = message.decode()
         except Exception as exc:  # pylint: disable=broad-except
             return self.on_decode_error(message, exc)
         try:
             type_, payload = payload['task'], payload  # protocol v1
         except (TypeError, KeyError):
             return on_unknown_message(payload, message)
     try:
         strategy = strategies[type_]
     except KeyError as exc:
         return on_unknown_task(None, message, exc)
     else:
         try:
             strategy(
                 message, payload,
                 promise(call_soon, (message.ack_log_error,)),
                 promise(call_soon, (message.reject_log_error,)),
                 callbacks,
             )
         except InvalidTaskError as exc:
             return on_invalid_task(payload, message, exc)

    def get(self, timeout=None, propagate=True, interval=0.5,
            no_ack=True, follow_parents=True, callback=None, on_message=None,
            on_interval=None, disable_sync_subtasks=True,
            EXCEPTION_STATES=states.EXCEPTION_STATES,
            PROPAGATE_STATES=states.PROPAGATE_STATES):
        """Wait until task is ready, and return its result.

        Warning:
           Waiting for tasks within a task may lead to deadlocks.
           Please read :ref:`task-synchronous-subtasks`.

        Arguments:
            timeout (float): How long to wait, in seconds, before the
                operation times out.
            propagate (bool): Re-raise exception if the task failed.
            interval (float): Time to wait (in seconds) before retrying to
                retrieve the result.  Note that this does not have any effect
                when using the RPC/redis result store backends, as they don't
                use polling.
            no_ack (bool): Enable amqp no ack (automatically acknowledge
                message).  If this is :const:`False` then the message will
                **not be acked**.
            follow_parents (bool): Re-raise any exception raised by
                parent tasks.
            disable_sync_subtasks (bool): Disable tasks to wait for sub tasks
                this is the default configuration. CAUTION do not enable this
                unless you must.

        Raises:
            celery.exceptions.TimeoutError: if `timeout` isn't
                :const:`None` and the result does not arrive within
                `timeout` seconds.
            Exception: If the remote call raised an exception then that
                exception will be re-raised in the caller process.
        """
        if disable_sync_subtasks:
            assert_will_not_block()
        _on_interval = promise()
        if follow_parents and propagate and self.parent:
            on_interval = promise(self._maybe_reraise_parent_error, weak=True)
            self._maybe_reraise_parent_error()
        if on_interval:
            _on_interval.then(on_interval)

        if self._cache:
            if propagate:
                self.maybe_throw(callback=callback)
            return self.result

        self.backend.add_pending_result(self)
        return self.backend.wait_for_pending(
            self, timeout=timeout,
            interval=interval,
            on_interval=_on_interval,
            no_ack=no_ack,
            propagate=propagate,
            callback=callback,
            on_message=on_message,
        )

 def __init__(self, results, app=None, ready_barrier=None, **kwargs):
     self._app = app
     self._cache = None
     self.results = results
     self.on_ready = promise(args=(self,))
     self._on_full = ready_barrier
     if self._on_full:
         self._on_full.then(promise(self.on_ready))

    def get(self, timeout=None, propagate=True, interval=0.5,
            no_ack=True, follow_parents=True, callback=None, on_interval=None,
            EXCEPTION_STATES=states.EXCEPTION_STATES,
            PROPAGATE_STATES=states.PROPAGATE_STATES):
        """Wait until task is ready, and return its result.

        .. warning::

           Waiting for tasks within a task may lead to deadlocks.
           Please read :ref:`task-synchronous-subtasks`.

        :keyword timeout: How long to wait, in seconds, before the
                          operation times out.
        :keyword propagate: Re-raise exception if the task failed.
        :keyword interval: Time to wait (in seconds) before retrying to
           retrieve the result.  Note that this does not have any effect
           when using the amqp result store backend, as it does not
           use polling.
        :keyword no_ack: Enable amqp no ack (automatically acknowledge
            message).  If this is :const:`False` then the message will
            **not be acked**.
        :keyword follow_parents: Re-raise any exception raised by parent task.

        :raises celery.exceptions.TimeoutError: if `timeout` is not
            :const:`None` and the result does not arrive within `timeout`
            seconds.

        If the remote call raised an exception then that exception will
        be re-raised.

        """
        assert_will_not_block()
        _on_interval = promise()
        if follow_parents and propagate and self.parent:
            on_interval = promise(self._maybe_reraise_parent_error)
            self._maybe_reraise_parent_error()
        if on_interval:
            _on_interval.then(on_interval)

        if self._cache:
            if propagate:
                self.maybe_throw(callback=callback)
            return self.result

        self.backend.add_pending_result(self)
        return self.backend.wait_for_pending(
            self, timeout=timeout,
            interval=interval,
            on_interval=_on_interval,
            no_ack=no_ack,
            propagate=propagate,
            callback=callback,
        )

 def __init__(self, event, data, sender, subscriber,
              id=None, on_success=None, on_error=None,
              timeout=None, on_timeout=None,
              retry=None, retry_max=None, retry_delay=None,
              headers=None, user_agent=None, app=None,
              recipient_validators=None):
     self.app = app_or_default(app or self.app)
     self.id = id or uuid()
     self.event = event
     self.data = data
     self.sender = sender
     self.subscriber = subscriber
     self.timeout = timeout
     self.on_success = on_success
     self.on_timeout = maybe_promise(on_timeout)
     self.on_error = on_error
     self.retry = self.app.settings.THORN_RETRY if retry is None else retry
     self.retry_max = (
         self.app.settings.THORN_RETRY_MAX
         if retry_max is None else retry_max)
     self.retry_delay = (
         self.app.settings.THORN_RETRY_DELAY
         if retry_delay is None else retry_delay)
     if recipient_validators is None:
         recipient_validators = self.app.settings.THORN_RECIPIENT_VALIDATORS
     self._recipient_validators = recipient_validators
     self.response = None
     self._headers = headers
     self._set_promise_target(promise(
         args=(self,), callback=self.on_success, on_error=self.on_error,
     ))
     if user_agent:
         self.user_agent = user_agent

 def __init__(
     self,
     url,
     method="GET",
     on_ready=None,
     on_timeout=None,
     on_stream=None,
     on_prepare=None,
     on_header=None,
     headers=None,
     **kwargs
 ):
     self.url = url
     self.method = method or self.method
     self.on_ready = maybe_promise(on_ready) or promise()
     self.on_timeout = maybe_promise(on_timeout)
     self.on_stream = maybe_promise(on_stream)
     self.on_prepare = maybe_promise(on_prepare)
     self.on_header = maybe_promise(on_header)
     if kwargs:
         for k, v in items(kwargs):
             setattr(self, k, v)
     if not isinstance(headers, Headers):
         headers = Headers(headers or {})
     self.headers = headers

 def __init__(self, id, ret_value, state, traceback=None):
     self.id = id
     self._result = ret_value
     self._state = state
     self._traceback = traceback
     self.on_ready = promise(args=(self,))
     self.on_ready()

 def _schedule_queue(self, queue):
     if queue in self._active_queues:
         if self.qos.can_consume():
             self._get_bulk_async(
                 queue, callback=promise(self._loop1, (queue,)),
             )
         else:
             self._loop1(queue)

 def __init__(self, id, ret_value, state, traceback=None):
     # pylint: disable=super-init-not-called
     # XXX should really not be inheriting from AsyncResult
     self.id = id
     self._result = ret_value
     self._state = state
     self._traceback = traceback
     self.on_ready = promise()
     self.on_ready(self)

 def send_ack(response, pid, job, fd):
     # Only used when synack is enabled.
     # Schedule writing ack response for when the fd is writable.
     msg = Ack(job, fd, precalc[response])
     callback = promise(write_generator_done)
     cor = _write_ack(fd, msg, callback=callback)
     mark_write_gen_as_active(cor)
     mark_write_fd_as_active(fd)
     callback.args = (cor,)
     add_writer(fd, cor)

    def _mexe(self, request, sender=None, callback=None):
        callback = callback or promise()
        conn = self.get_http_connection()

        if callable(sender):
            sender(conn, request.method, request.path, request.body,
                   request.headers, callback)
        else:
            conn.request(request.method, request.url,
                         request.body, request.headers)
            conn.getresponse(callback=callback)
        return callback

 def basic_cancel(self, consumer_tag):
     # If we are busy reading messages we may experience
     # a race condition where a message is consumed after
     # canceling, so we must delay this operation until reading
     # is complete (Issue celery/celery#1773).
     connection = self.connection
     if connection:
         if connection.cycle._in_protected_read:
             return connection.cycle.after_read.add(
                 promise(self._basic_cancel, (consumer_tag,)),
             )
         return self._basic_cancel(consumer_tag)

 def test_basic_consume_no_wait(self):
     callback = Mock()
     ret_promise = promise()
     self.c.send_method.return_value = ret_promise
     ret = self.c.basic_consume(
         'q', 123, arguments={'x': 1},
         callback=callback, nowait=True
     )
     self.c.send_method.assert_called_with(
         spec.Basic.Consume, 'BssbbbbF',
         (0, 'q', 123, False, False, False, True, {'x': 1}),
         wait=None,
         returns_tuple=True
     )
     assert self.c.callbacks[123] is callback
     assert ret == ret_promise

 def __init__(self, event, data, sender, subscriber,
              id=None, on_success=None, on_error=None,
              timeout=None, on_timeout=None,
              retry=None, retry_max=None, retry_delay=None,
              headers=None, user_agent=None, app=None,
              recipient_validators=None,
              allow_keepalive=True,
              allow_redirects=None):
     # type: (str, Dict, Any, Subscriber, str, Callable,
     #        Callable, float, Callable, bool, int,
     #        float, Mapping, str, App, Sequence[Callable],
     #        bool, bool) -> None
     self.app = app_or_default(app or self.app)
     self.id = id or uuid()
     self.event = event
     self.data = data
     self.sender = sender
     self.subscriber = subscriber
     self.timeout = timeout
     self.on_success = on_success
     self.on_timeout = maybe_promise(on_timeout)
     self.on_error = on_error
     self.retry = self.app.settings.THORN_RETRY if retry is None else retry
     self.retry_max = (
         self.app.settings.THORN_RETRY_MAX
         if retry_max is None else retry_max)
     self.retry_delay = (
         self.app.settings.THORN_RETRY_DELAY
         if retry_delay is None else retry_delay)
     if recipient_validators is None:
         recipient_validators = self.app.settings.THORN_RECIPIENT_VALIDATORS
     self.allow_keepalive = allow_keepalive
     if allow_redirects is None:
         allow_redirects = self.app.settings.THORN_ALLOW_REDIRECTS
     self.allow_redirects = allow_redirects
     self._recipient_validators = recipient_validators
     self.response = None
     self._headers = headers
     self._set_promise_target(promise(
         args=(self,), callback=self.on_success, on_error=self.on_error,
     ))
     if user_agent:
         self.user_agent = user_agent

    def send_method(self, sig,
                    format=None, args=None, content=None,
                    wait=None, callback=None, returns_tuple=False):
        p = promise()
        conn = self.connection
        if conn is None:
            raise RecoverableConnectionError('connection already closed')
        args = dumps(format, args) if format else ''
        try:
            conn.frame_writer(1, self.channel_id, sig, args, content)
        except StopIteration:
            raise RecoverableConnectionError('connection already closed')

        # TODO temp: callback should be after write_method ... ;)
        if callback:
            p.then(callback)
        p()
        if wait:
            return self.wait(wait, returns_tuple=returns_tuple)
        return p

    def _mexe(self, request, sender=None, callback=None):
        callback = callback or promise()
        boto.log.debug(
            'HTTP %s/%s headers=%s body=%s',
            request.host, request.path,
            request.headers, request.body,
        )

        conn = self.get_http_connection(
            request.host, request.port, self.is_secure,
        )
        request.authorize(connection=self)

        if callable(sender):
            sender(conn, request.method, request.path, request.body,
                   request.headers, callback)
        else:
            conn.request(request.method, request.path,
                         request.body, request.headers)
            conn.getresponse(callback=callback)
        return callback

 def call_soon(self, callback, *args):
     if not isinstance(callback, Thenable):
         callback = promise(callback, args)
     self._ready.add(callback)
     return callback

 def test_send_method__callback(self):
     callback = Mock(name='callback')
     p = promise(callback)
     self.c.send_method((50, 60), 'iB', (30, 0), callback=p)
     callback.assert_called_with()

 def test_call_soon(self, promise):
     callback = Mock(name='callback')
     ret = self.hub.call_soon(callback, 1, 2, 3)
     promise.assert_called_with(callback, (1, 2, 3))
     assert promise() in self.hub._ready
     assert ret is promise()