HuangYi/simple_rpc_with_allegra

   1 import pickle
   2 from allegra import (
   3         async_loop, async_chat,
   4         async_server, async_client )
   5 
   6 def dump_call(funcname, args, kwargs):
   7     para = (args, kwargs)
   8     para = pickle.dumps( para )
   9     data = funcname + '\t' + para
  10     length = len(data)
  11     return '%s\r\n%s' % ( hex(length), data )
  12 
  13 def parse_call(data):
  14     funcname, para = data.split('\t', 1)
  15     para = pickle.loads(para)
  16     return funcname, para
  17 
  18 class Dispatcher(async_chat.Dispatcher):
  19     ''' the dispatcher for both server and client '''
  20     terminator = '\r\n'
  21     def __init__(self, obj):
  22         super(Dispatcher, self).__init__()
  23         self.buffer = []
  24         self.obj = obj
  25         self.obj.remote = Remote(self.async_chat_push)
  26 
  27     def handle_connect(self):
  28         self.obj.handle_connect()
  29 
  30     def handle_close(self):
  31         self.close()
  32         self.obj.handle_close()
  33 
  34     def collect_incoming_data(self, data):
  35         self.buffer.append(data)
  36 
  37     def found_terminator(self):
  38         data = ''.join(self.buffer)
  39         self.buffer = []
  40         if not data:
  41             return True
  42         if self.terminator == '\r\n':
  43             length = int(data, 16)
  44             self.set_terminator(length)
  45         else:
  46             self.set_terminator('\r\n')
  47             self.call(data)
  48 
  49     def call(self, data):
  50         funcname, para = parse_call(data)
  51         try:
  52             func = getattr(self.obj, funcname)
  53         except AttributeError:
  54             return
  55         else:
  56             try:
  57                 func(*para[0], **para[1])
  58             except TypeError:
  59                 print 'parameter error'
  60 
  61 class Remote(object):
  62     ''' wrapper for remote object '''
  63     def __init__(self, send):
  64         self.send = send
  65     def __getattr__(self, name):
  66         def method(*args, **kw):
  67             self.send( dump_call( name, args, kw ) )
  68         return method
  69 
  70 def serve(obj, ip, port):
  71     ''' start a server '''
  72     return async_server.Listen( lambda : Dispatcher(obj), (ip, port), 6.0, 5 )
  73 
  74 def connect(obj, ip, port):
  75     ''' connect a client '''
  76     return async_client.connect( Dispatcher(obj), (ip, port), 3)

   1 from allegra import async_loop
   2 from async_rpc import serve
   3 
   4 class Temp(object):
   5     def __init__(self):
   6         self.remote = None
   7     def handle_connect(self):
   8         print 'connected'
   9     def handle_close(self):
  10         print 'closed'
  11     def say(self, str):
  12         print str
  13         self.remote.test(1)
  14     def test(self, a, b=1):
  15         print a,b
  16 
  17 t = Temp()
  18 serve(t, 'localhost', 9000)
  19 async_loop.dispatch()

   1 from allegra import async_loop
   2 from async_rpc import connect
   3 
   4 class Temp(object):
   5     def handle_connect(self):
   6         print 'connected'
   7     def handle_close(self):
   8         print 'closed'
   9     def say(self, str):
  10         print str
  11         self.remote.say(str)
  12     def test(self, a, b=1):
  13         print a,b
  14 
  15 t = Temp()
  16 connect(t, 'localhost', 9000)
  17 t.say('hello')
  18 t.remote.test(1,2)
  19 async_loop.dispatch()