doc/core/howto/listings/trial/calculus/remote_2.py

   1 # -*- test-case-name: calculus.test.test_remote_1 -*-
   2
   3 from twisted.protocols import basic
   4 from twisted.internet import protocol
   5 from twisted.python import log
   6 from calculus.base_3 import Calculation
   7
   8
   9
  10 class CalculationProxy(object):
  11     def __init__(self):
  12         self.calc = Calculation()
  13         for m in ['add', 'subtract', 'multiply', 'divide']:
  14             setattr(self, 'remote_%s' % m, getattr(self.calc, m))
  15
  16
  17
  18 class RemoteCalculationProtocol(basic.LineReceiver):
  19     def __init__(self):
  20         self.proxy = CalculationProxy()
  21
  22
  23     def lineReceived(self, line):
  24         op, a, b = line.split()
  25         op = getattr(self.proxy, 'remote_%s' % (op,))
  26         try:
  27             result = op(a, b)
  28         except TypeError:
  29             log.err()
  30             self.sendLine("error")
  31         else:
  32             self.sendLine(str(result))
  33
  34
  35
  36 class RemoteCalculationFactory(protocol.Factory):
  37     protocol = RemoteCalculationProtocol
  38
  39
  40
  41 def main():
  42     from twisted.internet import reactor
  43     from twisted.python import log
  44     import sys
  45     log.startLogging(sys.stdout)
  46     reactor.listenTCP(0, RemoteCalculationFactory())
  47     reactor.run()
  48
  49
  50 if __name__ == "__main__":
  51     main()