Client/Server split in {Client/Server}Base

bombela · bombela · commit 04ed696312b9 · 2012-03-25T02:54:08.000-07:00
diff --git a/zerorpc/core.py b/zerorpc/core.py
@@ -39,93 +39,11 @@
 from .heartbeat import HeartBeatOnChannel
 
 
-class DecoratorBase(object):
-    pattern = None
-
-    def __init__(self, functor):
-        self._functor = functor
-        self.__doc__ = functor.__doc__
-        self.__name__ = functor.__name__
-
-    def __get__(self, instance, type_instance=None):
-        if instance is None:
-            return self
-        return self.__class__(self._functor.__get__(instance, type_instance))
-
-    def __call__(self, *args, **kargs):
-        return self._functor(*args, **kargs)
-
-    def _zerorpc_args(self):
-        try:
-            args_spec = self._functor._zerorpc_args()
-        except AttributeError:
-            try:
-                args_spec = inspect.getargspec(self._functor)
-            except TypeError:
-                try:
-                    args_spec = inspect.getargspec(self._functor.__call__)
-                except (AttributeError, TypeError):
-                    args_spec = None
-        return args_spec
-
-
-class PatternReqRep():
-
-    def process_call(self, context, bufchan, event, functor):
-        result = context.middleware_call_procedure(functor, *event.args)
-        bufchan.emit('OK', (result,))
+class ServerBase(object):
 
-    def accept_answer(self, event):
-        return True
-
-    def process_answer(self, context, bufchan, event, method, timeout,
-            raise_remote_error):
-        result = event.args[0]
-        if event.name == 'ERR':
-            raise_remote_error(event)
-        bufchan.close()
-        bufchan.channel.close()
-        bufchan.channel.channel.close()
-        return result
-
-
-class rep(DecoratorBase):
-    pattern = PatternReqRep()
-
-
-class PatternReqStream():
-
-    def process_call(self, context, bufchan, event, functor):
-        for result in iter(context.middleware_call_procedure(functor,
-                *event.args)):
-            bufchan.emit('STREAM', result)
-        bufchan.emit('STREAM_DONE', None)
-
-    def accept_answer(self, event):
-        return event.name in ('STREAM', 'STREAM_DONE')
-
-    def process_answer(self, context, bufchan, event, method, timeout,
-            raise_remote_error):
-        def iterator(event):
-            while event.name == 'STREAM':
-                yield event.args
-                event = bufchan.recv()
-            if event.name == 'ERR':
-                raise_remote_error(event)
-            bufchan.close()
-            bufchan.channel.channel.close()
-        return iterator(event)
-
-
-class stream(DecoratorBase):
-    pattern = PatternReqStream()
-
-
-class Server(SocketBase):
-    def __init__(self, methods=None, name=None, context=None, pool_size=None,
+    def __init__(self, channel, methods=None, name=None, pool_size=None,
             heartbeat=5):
-        super(Server, self).__init__(zmq.XREP, context)
-        self._multiplexer = ChannelMultiplexer(self._events)
+        self._multiplexer = ChannelMultiplexer(channel)
 
         if methods is None:
             methods = self
@@ -155,12 +73,10 @@ def __init__(self, methods=None, name=None, context=None, pool_size=None,
 
     def __del__(self):
         self.close()
-        super(Server, self).__del__()
 
     def close(self):
         self.stop()
         self._multiplexer.close()
-        super(Server, self).close()
 
     def _zerorpc_inspect(self, method=None, long_doc=True):
         if method:
@@ -241,14 +157,13 @@ def stop(self):
             self._acceptor_task.kill(block=False)
 
 
-class Client(SocketBase):
-    patterns = [PatternReqStream(), PatternReqRep()]
+class ClientBase(object):
 
-    def __init__(self, context=None, timeout=30, heartbeat=5,
+    def __init__(self, channel, patterns, timeout=30, heartbeat=5,
             passive_heartbeat=False):
-        super(Client, self).__init__(zmq.XREQ, context=context)
-        self._multiplexer = ChannelMultiplexer(self._events,
+        self._multiplexer = ChannelMultiplexer(channel,
                 ignore_broadcast=True)
+        self._patterns = patterns
         self._timeout = timeout
         self._heartbeat_freq = heartbeat
         self._passive_heartbeat = passive_heartbeat
@@ -272,7 +187,7 @@ def _raise_remote_error(self, event):
             raise RemoteError('RemoteError', msg, None)
 
     def _select_pattern(self, event):
-        for pattern in Client.patterns:
+        for pattern in self._patterns:
             if pattern.accept_answer(event):
                 return pattern
         msg = 'Unable to find a pattern for: {0}'.format(event)
@@ -322,6 +237,114 @@ def __getattr__(self, method):
         return lambda *args, **kargs: self(method, *args, **kargs)
 
 
+class DecoratorBase(object):
+    pattern = None
+
+    def __init__(self, functor):
+        self._functor = functor
+        self.__doc__ = functor.__doc__
+        self.__name__ = functor.__name__
+
+    def __get__(self, instance, type_instance=None):
+        if instance is None:
+            return self
+        return self.__class__(self._functor.__get__(instance, type_instance))
+
+    def __call__(self, *args, **kargs):
+        return self._functor(*args, **kargs)
+
+    def _zerorpc_args(self):
+        try:
+            args_spec = self._functor._zerorpc_args()
+        except AttributeError:
+            try:
+                args_spec = inspect.getargspec(self._functor)
+            except TypeError:
+                try:
+                    args_spec = inspect.getargspec(self._functor.__call__)
+                except (AttributeError, TypeError):
+                    args_spec = None
+        return args_spec
+
+
+class PatternReqRep():
+
+    def process_call(self, context, bufchan, event, functor):
+        result = context.middleware_call_procedure(functor, *event.args)
+        bufchan.emit('OK', (result,))
+
+    def accept_answer(self, event):
+        return True
+
+    def process_answer(self, context, bufchan, event, method, timeout,
+            raise_remote_error):
+        result = event.args[0]
+        if event.name == 'ERR':
+            raise_remote_error(event)
+        bufchan.close()
+        bufchan.channel.close()
+        bufchan.channel.channel.close()
+        return result
+
+
+class rep(DecoratorBase):
+    pattern = PatternReqRep()
+
+
+class PatternReqStream():
+
+    def process_call(self, context, bufchan, event, functor):
+        for result in iter(context.middleware_call_procedure(functor,
+                *event.args)):
+            bufchan.emit('STREAM', result)
+        bufchan.emit('STREAM_DONE', None)
+
+    def accept_answer(self, event):
+        return event.name in ('STREAM', 'STREAM_DONE')
+
+    def process_answer(self, context, bufchan, event, method, timeout,
+            raise_remote_error):
+        def iterator(event):
+            while event.name == 'STREAM':
+                yield event.args
+                event = bufchan.recv()
+            if event.name == 'ERR':
+                raise_remote_error(event)
+            bufchan.close()
+            bufchan.channel.channel.close()
+        return iterator(event)
+
+
+class stream(DecoratorBase):
+    pattern = PatternReqStream()
+
+
+class Server(SocketBase, ServerBase):
+
+    def __init__(self, methods=None, name=None, context=None, pool_size=None,
+            heartbeat=5):
+        SocketBase.__init__(self, zmq.XREP, context)
+        ServerBase.__init__(self, self._events, methods, name, pool_size, heartbeat)
+
+    def close(self):
+        ServerBase.close(self)
+        SocketBase.close(self)
+
+
+class Client(SocketBase, ClientBase):
+    patterns = [PatternReqStream(), PatternReqRep()]
+
+    def __init__(self, context=None, timeout=30, heartbeat=5,
+            passive_heartbeat=False):
+        SocketBase.__init__(self, zmq.XREQ, context=context)
+        ClientBase.__init__(self, self._events, Client.patterns, timeout,
+                heartbeat, passive_heartbeat)
+
+    def close(self):
+        ClientBase.close(self)
+        SocketBase.close(self)
+
+
 class Pusher(SocketBase):
 
     def __init__(self, context=None, zmq_socket=zmq.PUSH):
