[文档]defon_message(self,body,message):# just increase clock as clients usually don't# have a valid clock to adjust with.self._forward_clock()try:self.node.handle_message(body,message)exceptKeyErrorasexc:error('No such control command: %s',exc)exceptExceptionasexc:error('Control command error: %r',exc,exc_info=True)self.reset()
[文档]defon_stop(self):ifself._node_stopped:self._node_shutdown.set()debug('Waiting for broadcast thread to shutdown...')self._node_stopped.wait()self._node_stopped=self._node_shutdown=None
[文档]defloop(self,c):resets=[self._resets]shutdown=self._node_shutdown=threading.Event()stopped=self._node_stopped=threading.Event()try:withc.connection_for_read()asconnection:info('pidbox: Connected to %s.',connection.as_uri())self._do_reset(c,connection)whilenotshutdown.is_set()andc.connection:ifresets[0]<self._resets:resets[0]+=1self._do_reset(c,connection)try:connection.drain_events(timeout=1.0)exceptsocket.timeout:passfinally:stopped.set()