#NEWLINE# @ misc . cachedproperty #NEWLINE# def banner( self ) : #NEWLINE# #INDENT# 'A banner that can be useful to display before running.' #NEWLINE# connection_details = self . _server . connection_details #NEWLINE# transport = connection_details . transport #NEWLINE# if transport . driver_version : #NEWLINE# #INDENT# transport_driver = ( '%s v%s' % ( transport . driver_name , transport . driver_version ) ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# transport_driver = transport . driver_name #NEWLINE# #UNINDENT# try : #NEWLINE# #INDENT# hostname = socket . getfqdn ( ) #NEWLINE# #UNINDENT# except socket . error : #NEWLINE# #INDENT# hostname = '???' #NEWLINE# #UNINDENT# try : #NEWLINE# #INDENT# pid = os . getpid ( ) #NEWLINE# #UNINDENT# except OSError : #NEWLINE# #INDENT# pid = '???' #NEWLINE# #UNINDENT# chapters = { #NEWLINE# #INDENT# 'Connection details' : { #NEWLINE# #INDENT# 'Driver' : transport_driver , #NEWLINE# 'Exchange' : self . _exchange , #NEWLINE# 'Topic' : self . _topic , #NEWLINE# 'Transport' : transport . driver_type , #NEWLINE# 'Uri' : connection_details . uri , #NEWLINE# #UNINDENT# } , #NEWLINE# 'Powered by' : { #NEWLINE# #INDENT# 'Executor' : reflection . get_class_name ( self . _executor ) , #NEWLINE# 'Thread count' : getattr ( self . _executor , 'max_workers' , '???' ) , #NEWLINE# #UNINDENT# } , #NEWLINE# 'Supported endpoints' : [ str ( ep ) for ep in self . _endpoints ] , #NEWLINE# 'System details' : { #NEWLINE# #INDENT# 'Hostname' : hostname , #NEWLINE# 'Pid' : pid , #NEWLINE# 'Platform' : platform . platform ( ) , #NEWLINE# 'Python' : sys . version . split ( '\n' , 1 ) [ 0 ] . strip ( ) , #NEWLINE# 'Thread id' : tu . get_ident ( ) , #NEWLINE# #UNINDENT# } , #NEWLINE# #UNINDENT# } #NEWLINE# return banner . make_banner ( 'WBE worker' , chapters )