def addEventListener(self, service, listener):
     assert protocol.isDispatchThread()
     svc_name = str(service)
     listener.svc_name = svc_name
     list = self.event_listeners.get(svc_name) or []
     list.append(listener)
     self.event_listeners[svc_name] = list

 def _processQueue(self):
     assert protocol.isDispatchThread()
     with self._lock:
         for cmd in self._queue:
             service, command, args, kwargs = cmd
             self._invoke(service, command, *args, **kwargs)
         del self._queue[:]

 def sendCommand(self, service, name, args, listener):
     assert protocol.isDispatchThread()
     if self.state == STATE_OPENING: raise Exception("Channel is waiting for Hello message")
     if self.state == STATE_CLOSED: raise Exception("Channel is closed")
     msg = Message('C')
     msg.service = str(service)
     msg.name = name
     msg.data = args
     channel = self
     class CancelableToken(Token):
         def __init__(self, listener):
             super(CancelableToken, self).__init__(listener=listener)
         def cancel(self):
             assert protocol.isDispatchThread()
             if channel.state != STATE_OPEN: return False
             with channel.out_lock:
                 if msg.is_sent: return False
                 msg.is_canceled = True
             del channel.out_tokens[msg.token.getID()]
             return True
     token = CancelableToken(listener)
     msg.token = token
     self.out_tokens[token.getID()] = msg
     self.addToOutQueue(msg)
     return token

 def wait(self, timeout=None):
     assert not protocol.isDispatchThread()
     with self._lock:
         while self._pending or self._queue:
             self._lock.wait(timeout)
             if timeout:
                 break

 def sendResult(self, token, results):
     assert protocol.isDispatchThread()
     if self.state != STATE_OPEN: raise Exception("Channel is closed")
     msg = Message('R')
     msg.data = results
     msg.token = token
     self.addToOutQueue(msg)

 def __sendCongestionLevel(self):
     self.local_congestion_cnt += 1
     if self.local_congestion_cnt < 8: return
     self.local_congestion_cnt = 0
     if self.state != STATE_OPEN: return
     timeVal = int(time.time() * 1000)
     if timeVal - self.local_congestion_time < 500: return
     assert protocol.isDispatchThread()
     level = protocol.getCongestionLevel()
     if level == self.local_congestion_level: return
     i = (level - self.local_congestion_level) / 8
     if i != 0: level = self.local_congestion_level + i
     self.local_congestion_time = timeVal
     with self.out_lock:
         msg = None
         if self.out_queue:
             msg = self.out_queue[0]
         if msg is None or msg.type != 'F':
             msg = Message('F')
             self.out_queue.insert(0, msg)
             self.out_lock.notify()
         data = "%i\0" % self.local_congestion_level
         msg.data = data
         msg.trace = self.trace_listeners
         self.local_congestion_level = level

 def getRemoteService(self, cls_or_name):
     assert protocol.isDispatchThread()
     assert self.state != STATE_OPENING
     if type(cls_or_name) == types.StringType:
         return self.remote_service_by_name.get(cls_or_name)
     else:
         return self.remote_service_by_class.get(cls_or_name)

 def getData(self):
     """
     @return cached data object.
     Note: It is prohibited to call this method when cache is not valid.
     """
     assert protocol.isDispatchThread()
     assert self.__valid
     return self.__data

 def cancel(self):
     if not protocol.isDispatchThread():
         protocol.invokeLater(self.cancel)
         return
     with self._lock:
         for cmd in self._pending.values():
             cmd.token.cancel()
         del self._queue[:]

 def dispose(self):
     assert protocol.isDispatchThread()
     id = self.getID()
     assert id
     peers = protocol.getLocator().getPeers()
     assert peers.get(id) == self
     del peers[id]
     self.sendPeerRemovedEvent()

 def cancel(self):
     assert protocol.isDispatchThread()
     if channel.state != STATE_OPEN: return False
     with channel.out_lock:
         if msg.is_sent: return False
         msg.is_canceled = True
     del channel.out_tokens[msg.token.getID()]
     return True

 def terminate(self, error):
     assert protocol.isDispatchThread()
     if self.state == STATE_CLOSED: return
     try:
         self.__sendEndOfStream(500)
     except Exception as x:
         if not error: error = x
     self._close(error)

 def close(self):
     assert protocol.isDispatchThread()
     if self.state == STATE_CLOSED: return
     try:
         self.__sendEndOfStream(10000)
         self._close(None)
     except Exception as x:
         self._close(x)

 def sendEvent(self, service, name, args):
     assert protocol.isDispatchThread()
     if not (self.state == STATE_OPEN or self.state == STATE_OPENING and isinstance(service, locator.LocatorService)):
         raise Exception("Channel is closed")
     msg = Message('E')
     msg.service = str(service)
     msg.name = name
     msg.data = args
     self.addToOutQueue(msg)

 def cancel(self):
     assert protocol.isDispatchThread()
     with self._lock:
         if self.isDone(): return False
         self.__canceled = True
         self.__error = Exception("Canceled")
         if self.__channel:
             self.__channel.removeChannelListener(self.__channel_listener)
         self._lock.notifyAll()
     return True

 def __init__(self, attrs):
     super(AbstractPeer, self).__init__(attrs)
     assert protocol.isDispatchThread()
     id = self.getID()
     assert id
     peers = protocol.getLocator().getPeers()
     if isinstance(peers.get(id), RemotePeer):
         peers.get(id).dispose()
     assert id not in peers
     peers[id] = self
     self.sendPeerAddedEvent()

 def _waitForCommand(self, token, timeout=None):
     assert not protocol.isDispatchThread()
     with self._lock:
         while token.id in self._pending:
             self._lock.wait(timeout)
             if timeout: break
         else:
             if self._queue:
                 self._lock.wait(timeout)
                 while token.id in self._pending:
                     self._lock.wait(timeout)
                     if timeout: break

 def removeEventListener(self, service, listener):
     assert protocol.isDispatchThread()
     svc_name = str(service)
     list = self.event_listeners.get(svc_name)
     if not list: return
     for i in range(len(list)):
         if list[i] is listener:
             if len(list) == 1:
                 del self.event_listeners[svc_name]
             else:
                 del list[i]
             return

 def done(self, result):
     with self._lock:
         assert protocol.isDispatchThread()
         if self.__canceled: return
         assert not self.__is_done
         assert not self.__error
         assert self.__result is None
         self.__result = result
         self.__is_done = True
         if self.__channel:
             self.__channel.removeChannelListener(self.__channel_listener)
         self._lock.notifyAll()

 def reset(self, data=None):
     """
     Force cache to become valid, cancel pending data retrieval if data is provided.
     @param data - up-to-date data object (optional)
     """
     assert protocol.isDispatchThread()
     if data is not None and self._command is not None:
         self._command.cancel()
         self._command = None
     if not self.__disposed:
         self.__data = data
         self.__error = None
         self.__valid = True
     self.post()