def twisted_consume(callback, bindings=None, queues=None):
    """
    Start a consumer using the provided callback and run it using the Twisted
    event loop (reactor).

    .. note:: Callbacks run in a Twisted-managed thread pool using the
        :func:`twisted.internet.threads.deferToThread` API to avoid them blocking
        the event loop. If you wish to use Twisted APIs in your callback you must
        use the :func:`twisted.internet.threads.blockingCallFromThread` or
        :class:`twisted.internet.interfaces.IReactorFromThreads` APIs.

    This API expects the caller to start the reactor.

    Args:
        callback (callable): A callable object that accepts one positional argument,
            a :class:`.Message` or a class object that implements the ``__call__``
            method. The class will be instantiated before use.
        bindings (dict or list of dict): Bindings to declare before consuming. This
            should be the same format as the :ref:`conf-bindings` configuration.
        queues (dict): The queue to declare and consume from. Each key in this
            dictionary should be a queue name to declare, and each value should
            be a dictionary with the "durable", "auto_delete", "exclusive", and
            "arguments" keys.
    Returns:
        twisted.internet.defer.Deferred:
            A deferred that fires with the list of one or more
            :class:`.Consumer` objects. Each consumer object has a
            :attr:`.Consumer.result` instance variable that is a Deferred that
            fires or errors when the consumer halts. Note that this API is
            meant to survive network problems, so consuming will continue until
            :meth:`.Consumer.cancel` is called or a fatal server error occurs.
            The deferred returned by this function may error back with a
            :class:`fedora_messaging.exceptions.BadDeclaration` if queues or
            bindings cannot be declared on the broker, a
            :class:`fedora_messaging.exceptions.PermissionException` if the user
            doesn't have access to the queue, or
            :class:`fedora_messaging.exceptions.ConnectionException` if the TLS
            or AMQP handshake fails.

    """
    if isinstance(bindings, dict):
        bindings = [bindings]
    callback = _check_callback(callback)

    global _twisted_service
    if _twisted_service is None:
        _twisted_service = service.FedoraMessagingServiceV2(config.conf["amqp_url"])
        reactor.callWhenRunning(_twisted_service.startService)
        # Twisted is killing the underlying connection before stopService gets
        # called, so we need to add it as a pre-shutdown event to gracefully
        # finish up messages in progress.
        reactor.addSystemEventTrigger(
            "before", "shutdown", _twisted_service.stopService
        )

    return _twisted_service._service.factory.consume(callback, bindings, queues)
def twisted_consume ( callback , bindings = None , queues = None ) : if isinstance ( bindings , dict ) : bindings = [ bindings ] callback = _check_callback ( callback ) global _twisted_service if _twisted_service is None : _twisted_service = service . FedoraMessagingServiceV2 ( config . conf [ "amqp_url" ] ) reactor . callWhenRunning ( _twisted_service . startService ) reactor . addSystemEventTrigger ( "before" , "shutdown" , _twisted_service . stopService ) return _twisted_service . _service . factory . consume ( callback , bindings , queues )
Start a consumer using the provided callback and run it using the Twisted
    event loop (reactor).

    .. note:: Callbacks run in a Twisted-managed thread pool using the
        :func:`twisted.internet.threads.deferToThread` API to avoid them blocking
        the event loop. If you wish to use Twisted APIs in your callback you must
        use the :func:`twisted.internet.threads.blockingCallFromThread` or
        :class:`twisted.internet.interfaces.IReactorFromThreads` APIs.

    This API expects the caller to start the reactor.

    Args:
        callback (callable): A callable object that accepts one positional argument,
            a :class:`.Message` or a class object that implements the ``__call__``
            method. The class will be instantiated before use.
        bindings (dict or list of dict): Bindings to declare before consuming. This
            should be the same format as the :ref:`conf-bindings` configuration.
        queues (dict): The queue to declare and consume from. Each key in this
            dictionary should be a queue name to declare, and each value should
            be a dictionary with the "durable", "auto_delete", "exclusive", and
            "arguments" keys.
    Returns:
        twisted.internet.defer.Deferred:
            A deferred that fires with the list of one or more
            :class:`.Consumer` objects. Each consumer object has a
            :attr:`.Consumer.result` instance variable that is a Deferred that
            fires or errors when the consumer halts. Note that this API is
            meant to survive network problems, so consuming will continue until
            :meth:`.Consumer.cancel` is called or a fatal server error occurs.
            The deferred returned by this function may error back with a
            :class:`fedora_messaging.exceptions.BadDeclaration` if queues or
            bindings cannot be declared on the broker, a
            :class:`fedora_messaging.exceptions.PermissionException` if the user
            doesn't have access to the queue, or
            :class:`fedora_messaging.exceptions.ConnectionException` if the TLS
            or AMQP handshake fails.