monitor/Rpyc/AsyncNetProxy.py

   1 from NetProxy import NetProxyWrapper\r
   2 from Lib import raise_exception\r
   3 \r
   4 \r
   5 class InvalidAsyncResultState(Exception):\r
   6     pass\r
   7 \r
   8 \r
   9 class AsyncNetProxy(NetProxyWrapper):\r
  10     """wraps an exiting synchronous netproxy to make is asynchronous \r
  11     (remote operations return AsyncResult objects)"""\r
  12     __doc__ = NetProxyWrapper.__doc__\r
  13 \r
  14     def __request__(self, handler, *args):\r
  15         res = AsyncResult(self.__dict__["____conn"])\r
  16         self.__dict__["____conn"].async_request(res.callback, handler, self.__dict__["____oid"], *args)\r
  17         return res\r
  18 \r
  19     # must return a string... and it's not meaningful to return the repr of an async result\r
  20     def __repr__(self, *args):\r
  21         return self.__request__("handle_repr", *args).result\r
  22     def __str__(self, *args):\r
  23         return self.__request__("handle_str", *args).result      \r
  24         \r
  25 \r
  26 class AsyncResult(object):\r
  27     """represents the result of an asynchronous operation"""\r
  28     STATE_PENDING = "pending"\r
  29     STATE_READY = "ready"\r
  30     STATE_EXCEPTION = "exception"\r
  31     STATE_UNKNOWN = "unknown"\r
  32     \r
  33     def __init__(self, conn):\r
  34         self.conn = conn\r
  35         self._state = self.STATE_PENDING\r
  36         self._result = None\r
  37         self._on_ready = None\r
  38     \r
  39     def __repr__(self):\r
  40         return "<AsyncResult (%s) at 0x%08x>" % (self._state, id(self))\r
  41     \r
  42     def callback(self, event, obj):\r
  43         if event == "result":\r
  44             self._state = self.STATE_READY\r
  45             self._result = obj\r
  46         elif event == "exception":\r
  47             self._state = self.STATE_EXCEPTION\r
  48             self._result = obj\r
  49         else:\r
  50             self._state = self.STATE_UNKNOWN\r
  51             self._result = obj\r
  52             \r
  53         if self._on_ready is not None:\r
  54             self._on_ready(self)\r
  55     \r
  56     def _get_on_ready(self):\r
  57         return self._ready_callback\r
  58 \r
  59     def _set_on_ready(self, obj):\r
  60         self._on_ready = obj\r
  61         if self._state != self.STATE_PENDING:\r
  62             self._on_ready(self)\r
  63     \r
  64     def _get_is_ready(self):\r
  65         if self._state == self.STATE_PENDING:\r
  66             self.conn.poll()\r
  67         return self._state != self.STATE_PENDING\r
  68     \r
  69     def _get_result(self):\r
  70         while self._state == self.STATE_PENDING:\r
  71             self.conn.serve()\r
  72         if self._state == self.STATE_READY:\r
  73             return self._result\r
  74         elif self._state == self.STATE_EXCEPTION:\r
  75             raise_exception(*self._result)\r
  76         else:\r
  77             raise InvalidAsyncResultState(self._state)\r
  78             \r
  79     is_ready = property(_get_is_ready, doc = "indicates whether or not the result is ready")\r
  80     result = property(_get_result, doc = "the value of the async result (may block)")\r
  81     on_ready = property(_get_on_ready, _set_on_ready, doc =\r
  82         "if not None, specifies a callback which is called when the result is ready")\r