<s> """<STR_LIT>""" <EOL> import sys <EOL> import os <EOL> import rpyc <EOL> from plumbum import cli <EOL> from rpyc . utils . server import ThreadedServer , ForkingServer , OneShotServer <EOL> from rpyc . utils . classic import DEFAULT_SERVER_PORT , DEFAULT_SERVER_SSL_PORT <EOL> from rpyc . utils . registry import REGISTRY_PORT <EOL> from rpyc . utils . registry import UDPRegistryClient , TCPRegistryClient <EOL> from rpyc . utils . authenticators import SSLAuthenticator <EOL> from rpyc . lib import setup_logger <EOL> from rpyc . core import SlaveService <EOL> class ClassicServer ( cli . Application ) : <EOL> mode = cli . SwitchAttr ( [ "<STR_LIT>" , "<STR_LIT>" ] , cli . Set ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) , <EOL> default = "<STR_LIT>" , help = "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> port = cli . SwitchAttr ( [ "<STR_LIT>" , "<STR_LIT>" ] , cli . Range ( <NUM_LIT:0> , <NUM_LIT> ) , default = None , <EOL> help = "<STR_LIT>" % <EOL> ( DEFAULT_SERVER_PORT , DEFAULT_SERVER_SSL_PORT ) , group = "<STR_LIT>" ) <EOL> host = cli . SwitchAttr ( [ "<STR_LIT>" ] , str , default = "<STR_LIT>" , help = "<STR_LIT>" <EOL> "<STR_LIT>" , group = "<STR_LIT>" ) <EOL> ipv6 = cli . Flag ( [ "<STR_LIT>" ] , help = "<STR_LIT>" , group = "<STR_LIT>" ) <EOL> logfile = cli . SwitchAttr ( "<STR_LIT>" , str , default = None , help = "<STR_LIT>" <EOL> "<STR_LIT>" , group = "<STR_LIT>" ) <EOL> quiet = cli . Flag ( [ "<STR_LIT>" , "<STR_LIT>" ] , help = "<STR_LIT>" , <EOL> group = "<STR_LIT>" ) <EOL> ssl_keyfile = cli . SwitchAttr ( "<STR_LIT>" , cli . ExistingFile , <EOL> help = "<STR_LIT>" , group = "<STR_LIT>" , <EOL> requires = [ "<STR_LIT>" ] ) <EOL> ssl_certfile = cli . SwitchAttr ( "<STR_LIT>" , cli . ExistingFile , <EOL> help = "<STR_LIT>" , group = "<STR_LIT>" , <EOL> requires = [ "<STR_LIT>" ] ) <EOL> ssl_cafile = cli . SwitchAttr ( "<STR_LIT>" , cli . ExistingFile , <EOL> help = "<STR_LIT>" <EOL> "<STR_LIT>" , group = "<STR_LIT>" , requires = [ "<STR_LIT>" ] ) <EOL> auto_register = cli . Flag ( "<STR_LIT>" , help = "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> group = "<STR_LIT>" ) <EOL> registry_type = cli . SwitchAttr ( "<STR_LIT>" , cli . Set ( "<STR_LIT>" , "<STR_LIT>" ) , <EOL> default = "<STR_LIT>" , help = "<STR_LIT>" , group = "<STR_LIT>" ) <EOL> registry_port = cli . SwitchAttr ( "<STR_LIT>" , cli . Range ( <NUM_LIT:0> , <NUM_LIT> ) , default = REGISTRY_PORT , <EOL> help = "<STR_LIT>" , group = "<STR_LIT>" ) <EOL> registry_host = cli . SwitchAttr ( "<STR_LIT>" , str , default = None , <EOL> help = "<STR_LIT>" <EOL> "<STR_LIT>" , group = "<STR_LIT>" ) <EOL> def main ( self ) : <EOL> if self . registry_type == "<STR_LIT>" : <EOL> if self . registry_host is None : <EOL> self . registry_host = "<STR_LIT>" <EOL> self . registrar = UDPRegistryClient ( ip = self . registry_host , port = self . registry_port ) <EOL> else : <EOL> if self . registry_host is None : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> self . registrar = TCPRegistryClient ( ip = self . registry_host , port = self . registry_port ) <EOL> if self . ssl_keyfile : <EOL> self . authenticator = SSLAuthenticator ( self . ssl_keyfile , self . ssl_certfile , <EOL> self . ssl_cafile ) <EOL> default_port = DEFAULT_SERVER_SSL_PORT <EOL> else : <EOL> self . authenticator = None <EOL> default_port = DEFAULT_SERVER_PORT <EOL> if self . port is None : <EOL> self . port = default_port <EOL> setup_logger ( self . quiet , self . logfile ) <EOL> if self . mode == "<STR_LIT>" : <EOL> self . _serve_mode ( ThreadedServer ) <EOL> elif self . mode == "<STR_LIT>" : <EOL> self . _serve_mode ( ForkingServer ) <EOL> elif self . mode == "<STR_LIT>" : <EOL> self . _serve_oneshot ( ) <EOL> elif self . mode == "<STR_LIT>" : <EOL> self . _serve_stdio ( ) <EOL> def _serve_mode ( self , factory ) : <EOL> t = factory ( SlaveService , hostname = self . host , port = self . port , <EOL> reuse_addr = True , ipv6 = self . ipv6 , authenticator = self . authenticator , <EOL> registrar = self . registrar , auto_register = self . auto_register ) <EOL> t . start ( ) <EOL> def _serve_oneshot ( self ) : <EOL> t = OneShotServer ( SlaveService , hostname = self . host , port = self . port , <EOL> reuse_addr = True , ipv6 = self . ipv6 , authenticator = self . authenticator , <EOL> registrar = self . registrar , auto_register = self . auto_register ) <EOL> sys . stdout . write ( "<STR_LIT>" ) <EOL> sys . stdout . write ( "<STR_LIT>" % ( t . host , t . port ) ) <EOL> sys . stdout . flush ( ) <EOL> t . start ( ) <EOL> def _serve_stdio ( self ) : <EOL> origstdin = sys . stdin <EOL> origstdout = sys . stdout <EOL> sys . stdin = open ( os . devnull , "<STR_LIT:r>" ) <EOL> sys . stdout = open ( os . devnull , "<STR_LIT:w>" ) <EOL> sys . stderr = open ( os . devnull , "<STR_LIT:w>" ) <EOL> conn = rpyc . classic . connect_pipes ( origstdin , origstdout ) <EOL> try : <EOL> try : <EOL> conn . serve_all ( ) <EOL> except KeyboardInterrupt : <EOL> print ( "<STR_LIT>" ) <EOL> finally : <EOL> conn . close ( ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> ClassicServer . run ( ) </s>
