# Path: snakemq/exceptions.py
# class SendNotFinished(SnakeMQException):
#     pass
# 
# Path: snakemq/poll.py
# def poll(self, timeout):
#     """
#     :param timeout: seconds
#     """
#     if len(self.fds) == 0:
#         time.sleep(timeout)
#         return []
# 
#     rlist = []
#     wlist = []
#     xlist = []
#     for fd, mask in self.fds.items():
#         fd = self._socket_to_fd(fd)
#         if mask & select.EPOLLIN:
#             rlist.append(fd)
#         if mask & select.EPOLLOUT:
#             wlist.append(fd)
#         xlist.append(fd)
# 
#     rlist, wlist, xlist = select.select(rlist, wlist, xlist, timeout)
# 
#     res = {}
#     for fd in rlist:
#         res[fd] = res.get(fd, 0) | select.EPOLLIN
#     for fd in wlist:
#         res[fd] = res.get(fd, 0) | select.EPOLLOUT
#     for fd in xlist:
#         res[fd] = res.get(fd, 0) | select.EPOLLERR
# 
#     return res.items()
# 
# 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 select
import socket
import errno
import time
import bisect
import logging
    import ssl
    import dummyssl as ssl
from snakemq.exceptions import SendNotFinished
from snakemq.poll import poll
from snakemq.pollbell import Bell
from snakemq.callbacks import Callback
            if fd in self._listen_socks_filenos:
                self.handle_accept(sock)
            else:
                self.handle_recv(sock)

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

    def handle_fd_mask(self, fd, mask):
        if fd == self._poll_bell.r:
            assert mask & select.EPOLLIN
            self._poll_bell.read(BELL_READ)  # flush the pipe
        else:
            # socket might have been already discarded by the Link
            # so this pass might be skipped
            if fd not in self._sock_by_fd:
                return
            sock = self._sock_by_fd[fd]

            if mask & (select.EPOLLERR | select.EPOLLHUP):
                self.handle_sock_err(sock)
            else:
                if sock in self._in_ssl_handshake:
                    if self.ssl_handshake(sock) == SSL_HANDSHAKE_DONE:
                        # connection is ready for user IO
                        self.on_connect(sock.conn_id)
                else:
                    self.handle_sock_io(fd, sock, mask)

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


# Path: snakemq/exceptions.py
# class SendNotFinished(SnakeMQException):
#     pass
# 
# Path: snakemq/poll.py
# def poll(self, timeout):
#     """
#     :param timeout: seconds
#     """
#     if len(self.fds) == 0:
#         time.sleep(timeout)
#         return []
# 
#     rlist = []
#     wlist = []
#     xlist = []
#     for fd, mask in self.fds.items():
#         fd = self._socket_to_fd(fd)
#         if mask & select.EPOLLIN:
#             rlist.append(fd)
#         if mask & select.EPOLLOUT:
#             wlist.append(fd)
#         xlist.append(fd)
# 
#     rlist, wlist, xlist = select.select(rlist, wlist, xlist, timeout)
# 
#     res = {}
#     for fd in rlist:
#         res[fd] = res.get(fd, 0) | select.EPOLLIN
#     for fd in wlist:
#         res[fd] = res.get(fd, 0) | select.EPOLLOUT
#     for fd in xlist:
#         res[fd] = res.get(fd, 0) | select.EPOLLERR
# 
#     return res.items()
# 
# 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/link.py
import select
import socket
import errno
import time
import bisect
import logging
    import ssl
    import dummyssl as ssl
from snakemq.exceptions import SendNotFinished
from snakemq.poll import poll
from snakemq.pollbell import Bell
from snakemq.callbacks import Callback

            if fd in self._listen_socks_filenos:
                self.handle_accept(sock)
            else:
                self.handle_recv(sock)

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

    def handle_fd_mask(self, fd, mask):
        if fd == self._poll_bell.r:
            assert mask & select.EPOLLIN
            self._poll_bell.read(BELL_READ)  # flush the pipe
        else:
            # socket might have been already discarded by the Link
            # so this pass might be skipped
            if fd not in self._sock_by_fd:
                return
            sock = self._sock_by_fd[fd]

            if mask & (select.EPOLLERR | select.EPOLLHUP):
                self.handle_sock_err(sock)
            else:
                if sock in self._in_ssl_handshake:
                    if self.ssl_handshake(sock) == SSL_HANDSHAKE_DONE:
                        # connection is ready for user IO
                        self.on_connect(sock.conn_id)
                else:
                    self.handle_sock_io(fd, sock, mask)

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


    def poll(self, poll_timeout):