def doKEvent(self, timeout):
        """
        Poll the kqueue for new events.
        """
        if timeout is None:
            timeout = 1

        try:
            l = self._kq.control([], len(self._selectables), timeout)
        except OSError as e:
            if e[0] == errno.EINTR:
                return
            else:
                raise

        _drdw = self._doWriteOrRead
        for event in l:
            fd = event.ident
            try:
                selectable = self._selectables[fd]
            except KeyError:
                # Handles the infrequent case where one selectable's
                # handler disconnects another.
                continue
            else:
                log.callWithLogger(selectable, _drdw, selectable, fd, event)

    def read(self, fd):
        if not self.watcher:
            return
        w = self.watcher
        # doRead can cause self.shutdown to be called so keep
        # a reference to self.watcher

        def _read():
            #Don't call me again, until the data has been read
            self.notifier.setEnabled(False)
            why = None
            try:
                why = w.doRead()
                inRead = True
            except:
                inRead = False
                log.err()
                why = sys.exc_info()[1]
            if why:
                self.reactor._disconnectSelectable(w, why, inRead)
            elif self.watcher:
                self.notifier.setEnabled(True)
                # Re enable notification following sucessfull read
            self.reactor._iterate(fromqt=True)
        log.callWithLogger(w, _read)

    def ssh_CHANNEL_DATA(self, packet):
        """
        The other side is sending us data.  Payload::
            uint32 local channel number
            string data

        Check to make sure the other side hasn't sent too much data (more
        than what's in the window, or more than the maximum packet size).  If
        they have, close the channel.  Otherwise, decrease the available
        window and pass the data to the channel's dataReceived().
        """
        localChannel, dataLength = struct.unpack('>2L', packet[:8])
        channel = self.channels[localChannel]
        # XXX should this move to dataReceived to put client in charge?
        if (dataLength > channel.localWindowLeft or
           dataLength > channel.localMaxPacket): # more data than we want
            log.callWithLogger(channel, log.msg, 'too much data')
            self.sendClose(channel)
            return
            #packet = packet[:channel.localWindowLeft+4]
        data = common.getNS(packet[4:])[0]
        channel.localWindowLeft -= dataLength
        if channel.localWindowLeft < channel.localWindowSize / 2:
            self.adjustWindow(channel, channel.localWindowSize - \
                                       channel.localWindowLeft)
            #log.msg('local window left: %s/%s' % (channel.localWindowLeft,
            #                                    channel.localWindowSize))
        log.callWithLogger(channel, channel.dataReceived, data)

 def _ioEventCallback(self, source, condition):
     """
     Called by event loop when an I/O event occurs.
     """
     log.callWithLogger(
         source, self._doReadOrWrite, source, source, condition)
     return True  # True = don't auto-remove the source

 def disconnectAll(self):
     """Disconnect every reader, and writer in the system.
     """
     selectables = self.removeAll()
     for reader in selectables:
         log.callWithLogger(reader, reader.connectionLost,
                            failure.Failure(main.CONNECTION_LOST))

 def dataReceived(self, data):
     self.buf = self.buf+data
     if not self.gotVersion:
         parts = self.buf.split('\n')
         for p in parts:
             if p[: 4] == 'SSH-':
                 self.gotVersion = 1
                 self.otherVersionString = p.strip()
                 if p.split('-')[1]not in('1.99', '2.0'): # bad version
                     self.sendDisconnect(DISCONNECT_PROTOCOL_VERSION_NOT_SUPPORTED, 'bad version %s'%p.split('-')[1])
                     return
                 i = parts.index(p)
                 self.buf = '\n'.join(parts[i+1:])
     packet = self.getPacket()
     while packet:
         messageNum = ord(packet[0])
         if messageNum < 50:
             messageType = messages[messageNum][4:]
             f = getattr(self, 'ssh_%s'%messageType, None)
             if f:
                 f(packet[1:])
             else:
                 log.msg("couldn't handle %s"%messageType)
                 log.msg(repr(packet[1:]))
                 self.sendUnimplemented()
         elif self.service:
             log.callWithLogger(self.service, self.service.packetReceived,
                                              ord(packet[0]), packet[1:])
         else:
             log.msg("couldn't handle %s"%messageNum)
             log.msg(repr(packet[1:]))
             self.sendUnimplemented()
         packet = self.getPacket()

    def doRead(self):
        """
        Some data is available for reading on ZeroMQ descriptor.

        ZeroMQ is signalling that we should process some events,
        we're starting to receive incoming messages.

        Implementation of :tm:`IReadDescriptor
        <internet.interfaces.IReadDescriptor>`.
        """
        if self.read_scheduled is not None:
            if not self.read_scheduled.called:
                self.read_scheduled.cancel()
            self.read_scheduled = None

        while True:
            if self.factory is None:  # disconnected
                return

            events = self.socket.get(constants.EVENTS)

            if (events & constants.POLLIN) != constants.POLLIN:
                return

            try:
                message = self._readMultipart()
            except error.ZMQError as e:
                if e.errno == constants.EAGAIN:
                    continue

                raise e

            log.callWithLogger(self, self.messageReceived, message)

    def doKEvent(self, timeout):
        """
        Poll the kqueue for new events.
        """
        if timeout is None:
            timeout = 1

        try:
            events = self._kq.control([], len(self._selectables), timeout)
        except OSError as e:
            # Since this command blocks for potentially a while, it's possible
            # EINTR can be raised for various reasons (for example, if the user
            # hits ^C).
            if e.errno == errno.EINTR:
                return
            else:
                raise

        _drdw = self._doWriteOrRead
        for event in events:
            fd = event.ident
            try:
                selectable = self._selectables[fd]
            except KeyError:
                # Handles the infrequent case where one selectable's
                # handler disconnects another.
                continue
            else:
                log.callWithLogger(selectable, _drdw, selectable, fd, event)

    def doPoll(self, timeout):
        """
        Poll the poller for new events.
        """
        if timeout is None:
            timeout = -1  # Wait indefinitely.

        try:
            # Limit the number of events to the number of io objects we're
            # currently tracking (because that's maybe a good heuristic) and
            # the amount of time we block to the value specified by our
            # caller.
            l = self._poller.poll(timeout, len(self._selectables))
        except IOError as err:
            if err.errno == errno.EINTR:
                return
            # See epoll_wait(2) for documentation on the other conditions
            # under which this can fail.  They can only be due to a serious
            # programming error on our part, so let's just announce them
            # loudly.
            raise

        _drdw = self._doReadOrWrite
        for fd, event in l:
            try:
                selectable = self._selectables[fd]
            except KeyError:
                pass
            else:
                log.callWithLogger(selectable, _drdw, selectable, fd, event)

 def ssh_CHANNEL_OPEN(self, packet):
     channelType, rest = common.getNS(packet)
     senderChannel, windowSize, maxPacket = struct.unpack('>3L', rest[: 12])
     packet = rest[12:]
     try:
         channel = self.getChannel(channelType, windowSize, maxPacket, packet)
         localChannel = self.localChannelID
         self.localChannelID+=1
         channel.id = localChannel
         self.channels[localChannel] = channel
         self.channelsToRemoteChannel[channel] = senderChannel
         self.localToRemoteChannel[localChannel] = senderChannel
         self.transport.sendPacket(MSG_CHANNEL_OPEN_CONFIRMATION, 
             struct.pack('>4L', senderChannel, localChannel, 
                 channel.localWindowSize, 
                 channel.localMaxPacket)+channel.specificData)
         log.callWithLogger(channel, channel.channelOpen, '')
     except Exception, e:
         log.msg('channel open failed')
         log.err(e)
         if isinstance(e, error.ConchError):
             reason, textualInfo = e.args[0], e.data
         else:
             reason = OPEN_CONNECT_FAILED
             textualInfo = "unknown failure"
         self.transport.sendPacket(MSG_CHANNEL_OPEN_FAILURE, 
                             struct.pack('>2L', senderChannel, reason)+ \
                            common.NS(textualInfo)+common.NS(''))

    def ssh_CHANNEL_EXTENDED_DATA(self, packet):
        """
        The other side is sending us exteneded data.  Payload::
            uint32  local channel number
            uint32  type code
            string  data

        Check to make sure the other side hasn't sent too much data (more
        than what's in the window, or or than the maximum packet size).  If
        they have, close the channel.  Otherwise, decrease the available
        window and pass the data and type code to the channel's
        extReceived().
        """
        localChannel, typeCode, dataLength = struct.unpack('>3L', packet[:12])
        channel = self.channels[localChannel]
        if (dataLength > channel.localWindowLeft or
                dataLength > channel.localMaxPacket):
            log.callWithLogger(channel, log.msg, 'too much extdata')
            self.sendClose(channel)
            return
        data = common.getNS(packet[8:])[0]
        channel.localWindowLeft -= dataLength
        if channel.localWindowLeft < channel.localWindowSize / 2:
            self.adjustWindow(channel, channel.localWindowSize -
                                       channel.localWindowLeft)
        log.callWithLogger(channel, channel.extReceived, typeCode, data)

 def _invoke_callback(self, fd, events):
     if fd not in self._fds:
         return
     (reader, writer) = self._fds[fd]
     if reader:
         err = None
         if reader.fileno() == -1:
             err = error.ConnectionLost()
         elif events & IOLoop.READ:
             err = log.callWithLogger(reader, reader.doRead)
         if err is None and events & IOLoop.ERROR:
             err = error.ConnectionLost()
         if err is not None:
             self.removeReader(reader)
             reader.readConnectionLost(failure.Failure(err))
     if writer:
         err = None
         if writer.fileno() == -1:
             err = error.ConnectionLost()
         elif events & IOLoop.WRITE:
             err = log.callWithLogger(writer, writer.doWrite)
         if err is None and events & IOLoop.ERROR:
             err = error.ConnectionLost()
         if err is not None:
             self.removeWriter(writer)
             writer.writeConnectionLost(failure.Failure(err))

 def ssh_CHANNEL_FAILURE(self, packet):
     localChannel = struct.unpack('>L', packet[: 4])[0]
     if self.deferreds.get(localChannel):
         d = self.deferreds[localChannel].pop(0)
         log.callWithLogger(self.channels[localChannel],
                            d.errback, 
                            error.ConchError('channel request failed'))

 def _runPendingEvents(self, pending_events):
     # pending_events is a list of (fd, mode) pairs.
     while pending_events:
         fd, mode = pending_events.pop()
         if fd in self._selectables:
             selectable = self._selectables[fd]
             log.callWithLogger(selectable,
                     self._doReadOrWrite, fd, mode, selectable)

 def ssh_CHANNEL_OPEN_FAILURE(self, packet):
     localChannel, reasonCode = struct.unpack('>2L', packet[: 8])
     reasonDesc = common.getNS(packet[8:])[0]
     channel = self.channels[localChannel]
     del self.channels[localChannel]
     channel.conn = self
     reason = error.ConchError(reasonDesc, reasonCode)
     log.callWithLogger(channel, channel.openFailed, reason)

 def ssh_CHANNEL_CLOSE(self, packet):
     localChannel = struct.unpack('>L', packet[: 4])[0]
     channel = self.channels[localChannel]
     if channel.remoteClosed:
         return
     log.callWithLogger(channel, channel.closeReceived)
     channel.remoteClosed = 1
     if channel.localClosed and channel.remoteClosed:
         self.channelClosed(channel)

 def ssh_CHANNEL_OPEN_CONFIRMATION(self, packet):
     localChannel, remoteChannel, windowSize, maxPacket = struct.unpack('>4L', packet[: 16])
     specificData = packet[16:]
     channel = self.channels[localChannel]
     channel.conn = self
     self.localToRemoteChannel[localChannel] = remoteChannel
     self.channelsToRemoteChannel[channel] = remoteChannel
     channel.remoteWindowLeft = windowSize
     channel.remoteMaxPacket = maxPacket
     log.callWithLogger(channel, channel.channelOpen, specificData)

    def ssh_CHANNEL_EOF(self, packet):
        """
        The other side is not sending any more data.  Payload::
            uint32  local channel number

        Notify the channel by calling its eofReceived() method.
        """
        localChannel = struct.unpack('>L', packet[:4])[0]
        channel = self.channels[localChannel]
        log.callWithLogger(channel, channel.eofReceived)

        def resolve_cb(result):
            if add_service:
                method = protocol.addService

            else:
                # remove the service from the list before calling back the proto
                _registered_services.remove(info)
                method = protocol.removeService

            # call it with the reader instance to make things cleaner
            log.callWithLogger(reader, method, info)

    def ssh_CHANNEL_SUCCESS(self, packet):
        """
        Our channel request to the other other side succeeded.  Payload::
            uint32  local channel number

        Get the C{Deferred} out of self.deferreds and call it back.
        """
        localChannel = struct.unpack('>L', packet[:4])[0]
        if self.deferreds.get(localChannel):
            d = self.deferreds[localChannel].pop(0)
            log.callWithLogger(self.channels[localChannel],
                               d.callback, '')