monitor/Rpyc/Channel.py

   1 from threading import RLock\r
   2 import struct\r
   3 \r
   4 \r
   5 class Channel(object):\r
   6     """a channel transfers packages over a stream. a package is any blob of data,\r
   7     up to 4GB in size. channels are gauranteed to be thread-safe"""\r
   8     HEADER_FORMAT = ">L" # byte order must be the same at both sides!\r
   9     HEADER_SIZE = struct.calcsize(HEADER_FORMAT)\r
  10 \r
  11     def __init__(self, stream):\r
  12         self.lock = RLock()\r
  13         self.stream = stream\r
  14 \r
  15     def __repr__(self):\r
  16         return "<%s(%r)>" % (self.__class__.__name__, self.stream)\r
  17 \r
  18     def send(self, data):\r
  19         """sends a package"""\r
  20         try:\r
  21             self.lock.acquire()\r
  22             header = struct.pack(self.HEADER_FORMAT, len(data))\r
  23             self.stream.write(header + data)\r
  24         finally:\r
  25             self.lock.release()\r
  26         \r
  27     def recv(self):\r
  28         """receives a package (blocking)"""\r
  29         try:\r
  30             self.lock.acquire()\r
  31             length, = struct.unpack(self.HEADER_FORMAT, self.stream.read(self.HEADER_SIZE))\r
  32             return self.stream.read(length)\r
  33         finally:\r
  34             self.lock.release()\r
  35     \r
  36     def close(self):\r
  37         return self.stream.close()\r
  38 \r
  39     def fileno(self):\r
  40         return self.stream.fileno()\r
  41 \r
  42     def is_available(self):\r
  43         return self.stream.is_available()\r
  44 \r
  45     def wait(self):\r
  46         return self.stream.wait()\r
  47 \r
  48 \r