Rpyc/Servers/ServerUtils.py

   1 import os\r
   2 import socket\r
   3 import sys\r
   4 import gc\r
   5 from threading import Thread\r
   6 from Rpyc.Connection import Connection\r
   7 from Rpyc.Stream import SocketStream, PipeStream\r
   8 from Rpyc.Channel import Channel\r
   9 from Rpyc.Lib import DEFAULT_PORT\r
  10 \r
  11 \r
  12 class Logger(object):\r
  13     def __init__(self, logfile = None, active = True):\r
  14         self.logfile = logfile\r
  15         self.active = active\r
  16     def __call__(self, *args):\r
  17         if not self.logfile:\r
  18             return\r
  19         if not self.active:\r
  20             return\r
  21         text = " ".join([str(a) for a in args])\r
  22         self.logfile.write("[%d] %s\n" % (os.getpid(), text))\r
  23         self.logfile.flush()\r
  24         \r
  25 log = Logger(sys.stdout)\r
  26 \r
  27 def _serve(chan):\r
  28     conn = Connection(chan)\r
  29     try:\r
  30         try:\r
  31             while True:\r
  32                 conn.serve()\r
  33         except EOFError:\r
  34             pass\r
  35     finally:\r
  36         conn.close()\r
  37         gc.collect()\r
  38 \r
  39 def serve_stream(stream, authenticate = False, users = None):\r
  40     chan = Channel(stream)\r
  41     \r
  42     if authenticate:\r
  43         from Rpyc.Authentication import accept\r
  44         log("requiring authentication")\r
  45         if accept(chan, users):\r
  46             log("authenication successful")\r
  47         else:\r
  48             log("authentication failed")\r
  49             return\r
  50     \r
  51     _serve(chan)\r
  52 \r
  53 def create_listener_socket(port):\r
  54     sock = socket.socket()\r
  55     sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)\r
  56     #sock.bind(("", port))\r
  57     sock.bind(("localhost", port))\r
  58     sock.listen(4)\r
  59     log("listening on", sock.getsockname())\r
  60     return sock\r
  61 \r
  62 def serve_socket(sock, **kw):\r
  63     sockname = sock.getpeername()\r
  64     log("welcome", sockname)\r
  65     try:\r
  66         try:\r
  67             serve_stream(SocketStream(sock), **kw)\r
  68         except socket.error:\r
  69             pass\r
  70     finally:\r
  71         log("goodbye", sockname)\r
  72 \r
  73 def serve_pipes(incoming, outgoing, **kw):\r
  74     serve_stream(PipeStream(incoming, outgoing), **kw)\r
  75 \r
  76 def threaded_server(port = DEFAULT_PORT, **kwargs):\r
  77     sock = create_listener_socket(port)\r
  78     while True:\r
  79         newsock, name = sock.accept()\r
  80         t = Thread(target = serve_socket, args = (newsock,), kwargs = kwargs)\r
  81         t.setDaemon(True)\r
  82         t.start()\r
  83 \r
  84 def start_threaded_server(*args, **kwargs):\r
  85     """starts the threaded_server on a separate thread. this turns the \r
  86     threaded_server into a mix-in you can place anywhere in your code"""\r
  87     t = Thread(target = threaded_server, args = args, kwargs = kwargs)\r
  88     t.setDaemon(True)\r
  89     t.start()\r
  90 \r