我们从Python开源项目中，提取了以下19个代码示例，用于说明如何使用asyncore.compact_traceback()。
def reap(self): # find DNS requests that have timed out now = int(time.time()) if now - self.last_reap_time > 180: # reap every 3 minutes self.last_reap_time = now # update before we forget for k, (host, unpack, callback, when) in self.request_map.items(): if now - when > 180: # over 3 minutes old del self.request_map[k] try: # same code as in handle_read callback(host, 0, None) # timeout val is (0,None) except Exception: (file, fun, line), t, v, tbinfo = \ asyncore.compact_traceback() self.log_info('%s %s %s' % (t, v, tbinfo), 'error')
def trace (self, name = ""): if self.logger: self.logger.trace (name) else: print(asyncore.compact_traceback ()) #---------------------------------------------------------------- # database connect #----------------------------------------------------------------
def trace (multirows = False): (file, fun, line), t, v, tbinfo = asyncore.compact_traceback() try: v = str (v) except: v = repr (v) line = "%s %s Traceback: %s" % (t, v, tbinfo) if multirows: line = trace ().replace ("] [", "\n - ") line = line.replace ("Traceback: [", "\n -----------\n + Traceback\n ===========\n - ") line = line [:-1] + "\n -----------" return line
def handle_error (self): dummy, exception_class, exception_str, tbinfo = asyncore.compact_traceback() self.has_result = False self.logger.trace () self.handle_close (exception_class, exception_str)
def handle_error (self): self.trace () self.__code = 900 self.__resp = asyncore.compact_traceback() [1].__name__ self.close()
def test_compact_traceback(self): try: raise Exception("I don't like spam!") except: real_t, real_v, real_tb = sys.exc_info() r = asyncore.compact_traceback() else: self.fail("Expected exception") (f, function, line), t, v, info = r self.assertEqual(os.path.split(f)[-1], 'test_asyncore.py') self.assertEqual(function, 'test_compact_traceback') self.assertEqual(t, real_t) self.assertEqual(v, real_v) self.assertEqual(info, '[%s|%s|%s]' % (f, function, line))
def handle_error(self): # don't close the socket on error (file, fun, line), t, v, tbinfo = asyncore.compact_traceback() self.log_info('Problem in Clock (%s:%s %s)' % (t, v, tbinfo), 'error')
def handle_error(self): # don't close the socket on error (file, fun, line), t, v, tbinfo = asyncore.compact_traceback() self.log_info('Problem in ICP (%s:%s %s)' % (t, v, tbinfo), 'error')
def handle_read(self): self.recv(8192) try: self.lock.acquire() for thunk in self.thunks: try: thunk() except: (file, fun, line), t, v, tbinfo = \ asyncore.compact_traceback() print('exception in trigger thunk: (%s:%s %s)' % ( t, v, tbinfo)) self.thunks = [] finally: self.lock.release()
def handle_read(self): self.recv(8192) try: self.lock.acquire() for thunk in self.thunks: try: thunk() except: (file, fun, line), t, v, tbinfo = \ asyncore.compact_traceback() print('exception in trigger thunk: (%s:%s %s)' % (t, v, tbinfo)) self.thunks = [] finally: self.lock.release()
def handle_read(self): reply, whence = self.socket.recvfrom(512) # for security reasons we may want to double-check # that <whence> is the server we sent the request to. id = (ord(reply[0]) << 8) + ord(reply[1]) if id in self.request_map: host, unpack, callback, when = self.request_map[id] del self.request_map[id] ttl, answer = unpack(reply) try: callback(host, ttl, answer) except Exception: (file, fun, line), t, v, tbinfo = asyncore.compact_traceback() self.log_info('%s %s %s' % (t, v, tbinfo), 'error')
def found_terminator(self): line = self.in_buffer if not len(line): return sp = string.find(line, ' ') if sp != -1: line = [line[:sp], line[sp + 1:]] else: line = [line] command = string.lower(line[0]) # watch especially for 'urgent' abort commands. if string.find(command, 'abor') != -1: # strip off telnet sync chars and the like... while command and command[0] not in string.letters: command = command[1:] fun_name = 'cmd_%s' % command if command != 'pass': self.log('<== %s' % repr(self.in_buffer)[1:-1]) else: self.log('<== %s' % line[0] + ' <password>') self.in_buffer = '' if not hasattr(self, fun_name): self.command_not_understood(line[0]) return fun = getattr(self, fun_name) if (not self.authorized) and (command not in ( 'user', 'pass', 'help', 'quit')): self.respond('530 Please log in with USER and PASS') elif (not self.check_command_authorization(command)): self.command_not_authorized(command) else: try: fun(line) except Exception: self.server.total_exceptions.increment() (file, fun, line), t, v, tbinfo = asyncore.compact_traceback() if self.client_dc: try: self.client_dc.close() except Exception: pass self.respond( '451 Server Error: %s, %s: file: %s line: %s' % ( t, v, file, line, ) )
