# Path: snakemq/exceptions.py
# class SnakeMQBrokenMessage(SnakeMQBrokenFormat):
#     """
#     Received message has wrong structure.
#     """
#     pass
# 
# class SnakeMQException(Exception):
#     pass
# 
# class SnakeMQIncompatibleProtocol(SnakeMQException):
#     """
#     Remote side has incompatible protocol version.
#     """
#     pass
# 
# class SnakeMQNoIdent(SnakeMQException):
#     """
#     Remote side did not identified itself.
#     """
#     pass
# 
# class NoConnection(SnakeMQException):
#     pass
# 
# Path: snakemq/queues.py
# class QueuesManager(object):
#     def __init__(self, storage):
#         """
#         :param storage: None or persistent storage
#         """
#         assert (storage is None) or isinstance(storage, QueuesStorageBase)
#         self.storage = storage
#         self.queues = {}  #: name:Queue
#         self.log = logging.getLogger("snakemq.queuesmanager")
#         if storage:
#             self.load_from_storage()
#             self.log.debug("queues in storage: %i" % len(self.queues))
# 
#         self.on_item_drop = Callback()  #: ``func(queue_name, item_uuid)``
# 
#     ####################################################
# 
#     def load_from_storage(self):
#         for queue_name in self.storage.get_queues():
#             self.get_queue(queue_name)
# 
#     ####################################################
# 
#     def get_queue(self, queue_name):
#         """
#         :return: Queue
#         """
#         if queue_name in self.queues:
#             queue = self.queues[queue_name]
#         else:
#             queue = Queue(queue_name, self)
#             self.queues[queue_name] = queue
#         return queue
# 
#     ####################################################
# 
#     def cleanup(self):
#         """
#         remove empty queues
#         """
#         for queue_name, queue in self.queues.items():
#             if not queue:
#                 del self.queues[queue_name]
# 
#     ####################################################
# 
#     def close(self):
#         """
#         Delete queues and close persistent storage.
#         """
#         self.queues.clear()
#         if self.storage:
#             self.storage.close()
#             self.storage = None
# 
#     ####################################################
# 
#     def collect_garbage(self):
#         """
#         Call this periodically to remove obsolete items and empty queues.
#         """
#         # TODO
# 
#     ####################################################
# 
#     def __len__(self):
#         return len(self.queues)
# 
# Path: snakemq/callbacks.py
# class Callback(object):
#     def __init__(self):
#         self.callbacks = []
# 
#     ####################################################
# 
#     def add(self, func):
#         self.callbacks.append(func)
# 
#     ####################################################
# 
#     def remove(self, func):
#         self.callbacks.remove(func)
# 
#     ####################################################
# 
#     def __call__(self, *args, **kwargs):
#         for callback in self.callbacks:
#             callback(*args, **kwargs)


import struct
import logging
import threading
import re
import time
import snakemq.version
from snakemq.exceptions import (SnakeMQBrokenMessage, SnakeMQException,
                                SnakeMQIncompatibleProtocol, SnakeMQNoIdent,
                                NoConnection)
from snakemq.queues import QueuesManager
from snakemq.message import Message
from snakemq.callbacks import Callback
    def parse_incompatible_protocol(self, conn_id):
        self.log.debug("conn=%s remote side rejected protocol version" % conn_id)
        # TODO

    ###########################################################

    def parse_identification(self, payload, conn_id):
        remote_ident = memstr(payload).decode(ENCODING, "replace")
        self.log.debug("conn=%s remote ident '%s'" % (conn_id, remote_ident))

        if conn_id in self._ident_by_conn:
            # avoid multiple identifications from the same peer
            return

        if remote_ident in self._conn_by_ident:
            # two peers with the same identifications can't be allowed
            self.log.debug("duplicate ident '%s'" % (remote_ident))
            self.packeter.link.close(conn_id)
            return

        with self._lock:
            self.queues_manager.get_queue(remote_ident).connect()
        self._ident_by_conn[conn_id] = remote_ident
        self._conn_by_ident[remote_ident] = conn_id
        self.on_connect(conn_id, remote_ident)

    ###########################################################

    def parse_message(self, payload, conn_id):
        if len(payload) < FRAME_FORMAT_MESSAGE_SIZE:

# Path: snakemq/exceptions.py
# class SnakeMQBrokenMessage(SnakeMQBrokenFormat):
#     """
#     Received message has wrong structure.
#     """
#     pass
# 
# class SnakeMQException(Exception):
#     pass
# 
# class SnakeMQIncompatibleProtocol(SnakeMQException):
#     """
#     Remote side has incompatible protocol version.
#     """
#     pass
# 
# class SnakeMQNoIdent(SnakeMQException):
#     """
#     Remote side did not identified itself.
#     """
#     pass
# 
# class NoConnection(SnakeMQException):
#     pass
# 
# Path: snakemq/queues.py
# class QueuesManager(object):
#     def __init__(self, storage):
#         """
#         :param storage: None or persistent storage
#         """
#         assert (storage is None) or isinstance(storage, QueuesStorageBase)
#         self.storage = storage
#         self.queues = {}  #: name:Queue
#         self.log = logging.getLogger("snakemq.queuesmanager")
#         if storage:
#             self.load_from_storage()
#             self.log.debug("queues in storage: %i" % len(self.queues))
# 
#         self.on_item_drop = Callback()  #: ``func(queue_name, item_uuid)``
# 
#     ####################################################
# 
#     def load_from_storage(self):
#         for queue_name in self.storage.get_queues():
#             self.get_queue(queue_name)
# 
#     ####################################################
# 
#     def get_queue(self, queue_name):
#         """
#         :return: Queue
#         """
#         if queue_name in self.queues:
#             queue = self.queues[queue_name]
#         else:
#             queue = Queue(queue_name, self)
#             self.queues[queue_name] = queue
#         return queue
# 
#     ####################################################
# 
#     def cleanup(self):
#         """
#         remove empty queues
#         """
#         for queue_name, queue in self.queues.items():
#             if not queue:
#                 del self.queues[queue_name]
# 
#     ####################################################
# 
#     def close(self):
#         """
#         Delete queues and close persistent storage.
#         """
#         self.queues.clear()
#         if self.storage:
#             self.storage.close()
#             self.storage = None
# 
#     ####################################################
# 
#     def collect_garbage(self):
#         """
#         Call this periodically to remove obsolete items and empty queues.
#         """
#         # TODO
# 
#     ####################################################
# 
#     def __len__(self):
#         return len(self.queues)
# 
# Path: snakemq/callbacks.py
# class Callback(object):
#     def __init__(self):
#         self.callbacks = []
# 
#     ####################################################
# 
#     def add(self, func):
#         self.callbacks.append(func)
# 
#     ####################################################
# 
#     def remove(self, func):
#         self.callbacks.remove(func)
# 
#     ####################################################
# 
#     def __call__(self, *args, **kwargs):
#         for callback in self.callbacks:
#             callback(*args, **kwargs)


# Path: snakemq/messaging.py
import struct
import logging
import threading
import re
import time
import snakemq.version
from snakemq.exceptions import (SnakeMQBrokenMessage, SnakeMQException,
                                SnakeMQIncompatibleProtocol, SnakeMQNoIdent,
                                NoConnection)
from snakemq.queues import QueuesManager
from snakemq.message import Message
from snakemq.callbacks import Callback

    def parse_incompatible_protocol(self, conn_id):
        self.log.debug("conn=%s remote side rejected protocol version" % conn_id)
        # TODO

    ###########################################################

    def parse_identification(self, payload, conn_id):
        remote_ident = memstr(payload).decode(ENCODING, "replace")
        self.log.debug("conn=%s remote ident '%s'" % (conn_id, remote_ident))

        if conn_id in self._ident_by_conn:
            # avoid multiple identifications from the same peer
            return

        if remote_ident in self._conn_by_ident:
            # two peers with the same identifications can't be allowed
            self.log.debug("duplicate ident '%s'" % (remote_ident))
            self.packeter.link.close(conn_id)
            return

        with self._lock:
            self.queues_manager.get_queue(remote_ident).connect()
        self._ident_by_conn[conn_id] = remote_ident
        self._conn_by_ident[remote_ident] = conn_id
        self.on_connect(conn_id, remote_ident)

    ###########################################################

    def parse_message(self, payload, conn_id):
        if len(payload) < FRAME_FORMAT_MESSAGE_SIZE:

            raise SnakeMQBrokenMessage("message")