# Path: pq/utils/serializers.py
# class MessageDict(Message):
# 
#     def __init__(self, *args, **kwargs):
#         self.__dict__.update(*args, **kwargs)
# 
# Path: pq/utils/concurrency.py
# ASYNC_IO = 1        # run in the worker event loop
# THREAD_IO = 3       # run in the event loop executor
# CPUBOUND = 4        # run in a subprocess
# 
# Path: pq/mq.py
# class Manager(BaseComponent):
# 
#     def green_pool(self):
#         return GreenPool(loop=self._loop)
# 
#     def http(self):
#         return HttpClient(loop=self._loop)
# 
#     def queues(self):
#         """List of queue names for Message consumers
#         """
#         queues = [self.backend.node_name]
#         queues.extend(self.cfg.task_queues)
#         return queues
# 
#     async def store_message(self, message):
#         """Dummy function to store a message into a persistent database
#         """
#         pass
# 
#     def start(self):
#         """Optional start method, called by the backend when it starts
#         """
#         pass
# 
#     def close(self):
#         pass
# 
# def register_broker(name, factory=None):
#     if factory is None:
#         dotted_path = brokers.get(name)
#         if not dotted_path:
#             raise ImproperlyConfigured('No such message broker: %s' % name)
#         factory = module_attribute(dotted_path, safe=True)
#         if not factory:
#             raise ImproperlyConfigured(
#                 '"%s" store not available' % dotted_path)
#     else:
#         brokers[name] = factory
#     return factory


import platform
from asyncio import gather, new_event_loop
from pulsar.api import ensure_future, EventHandler, as_coroutine
from pulsar.apps.data import create_store
from pulsar.apps.greenio import GreenHttp
from pulsar.apps.http import HttpClient
from pulsar.utils.importer import module_attribute
from ..utils.serializers import MessageDict
from ..utils import concurrency
from ..mq import Manager, register_broker
        ]
        if message.id:
            coro.append(
                self.channels.publish(message.type, message.id, message)
            )
        await gather(*coro)

    def tick(self, monitor):
        pass

    def info(self):
        for consumer in self.consumers:
            try:
                info = consumer.info()
            except Exception:
                self.logger.exception('Unhandled information exception')
            else:
                if info:
                    yield consumer.name, info

    def lock(self, name, **kwargs):
        """aquire a distributed global lock for ``name``
        """
        return self.channels.lock('lock-%s' % name, **kwargs)

    def http_sessions(self, model=None):
        """Return an HTTP session handler for a given concurrency model
        """
        if model == concurrency.THREAD_IO:
            return HttpClient(loop=new_event_loop())

# Path: pq/utils/serializers.py
# class MessageDict(Message):
# 
#     def __init__(self, *args, **kwargs):
#         self.__dict__.update(*args, **kwargs)
# 
# Path: pq/utils/concurrency.py
# ASYNC_IO = 1        # run in the worker event loop
# THREAD_IO = 3       # run in the event loop executor
# CPUBOUND = 4        # run in a subprocess
# 
# Path: pq/mq.py
# class Manager(BaseComponent):
# 
#     def green_pool(self):
#         return GreenPool(loop=self._loop)
# 
#     def http(self):
#         return HttpClient(loop=self._loop)
# 
#     def queues(self):
#         """List of queue names for Message consumers
#         """
#         queues = [self.backend.node_name]
#         queues.extend(self.cfg.task_queues)
#         return queues
# 
#     async def store_message(self, message):
#         """Dummy function to store a message into a persistent database
#         """
#         pass
# 
#     def start(self):
#         """Optional start method, called by the backend when it starts
#         """
#         pass
# 
#     def close(self):
#         pass
# 
# def register_broker(name, factory=None):
#     if factory is None:
#         dotted_path = brokers.get(name)
#         if not dotted_path:
#             raise ImproperlyConfigured('No such message broker: %s' % name)
#         factory = module_attribute(dotted_path, safe=True)
#         if not factory:
#             raise ImproperlyConfigured(
#                 '"%s" store not available' % dotted_path)
#     else:
#         brokers[name] = factory
#     return factory


# Path: pq/server/producer.py
import platform
from asyncio import gather, new_event_loop
from pulsar.api import ensure_future, EventHandler, as_coroutine
from pulsar.apps.data import create_store
from pulsar.apps.greenio import GreenHttp
from pulsar.apps.http import HttpClient
from pulsar.utils.importer import module_attribute
from ..utils.serializers import MessageDict
from ..utils import concurrency
from ..mq import Manager, register_broker

        ]
        if message.id:
            coro.append(
                self.channels.publish(message.type, message.id, message)
            )
        await gather(*coro)

    def tick(self, monitor):
        pass

    def info(self):
        for consumer in self.consumers:
            try:
                info = consumer.info()
            except Exception:
                self.logger.exception('Unhandled information exception')
            else:
                if info:
                    yield consumer.name, info

    def lock(self, name, **kwargs):
        """aquire a distributed global lock for ``name``
        """
        return self.channels.lock('lock-%s' % name, **kwargs)

    def http_sessions(self, model=None):
        """Return an HTTP session handler for a given concurrency model
        """
        if model == concurrency.THREAD_IO:
            return HttpClient(loop=new_event_loop())

        elif model == concurrency.ASYNC_IO: