Skip to content

Commit ea64c0a

Browse files
bombelabombela
committed
Extract decorators and patterns from core
A little bit of reorganisation to better separate and present the code.
1 parent e2d5e75 commit ea64c0a

File tree

4 files changed

+144
-92
lines changed

4 files changed

+144
-92
lines changed

‎zerorpc/__init__.py‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -30,3 +30,4 @@
3030
from .events import *
3131
from .core import *
3232
from .heartbeat import *
33+
from .decorators import *

‎zerorpc/core.py‎

Lines changed: 7 additions & 92 deletions
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,6 @@
2323
# SOFTWARE.
2424

2525

26-
import inspect
2726
import sys
2827
import traceback
2928
import gevent.pool
@@ -38,7 +37,8 @@
3837
from .socket import SocketBase
3938
from .heartbeat import HeartBeatOnChannel
4039
from .context import Context
41-
40+
from .decorators import DecoratorBase, rep
41+
import patterns
4242

4343
class ServerBase(object):
4444

@@ -171,12 +171,11 @@ def stop(self):
171171

172172
class ClientBase(object):
173173

174-
def __init__(self, channel, patterns, context=None, timeout=30, heartbeat=5,
174+
def __init__(self, channel, context=None, timeout=30, heartbeat=5,
175175
passive_heartbeat=False):
176176
self._multiplexer = ChannelMultiplexer(channel,
177177
ignore_broadcast=True)
178178
self._context = context or Context.get_instance()
179-
self._patterns = patterns
180179
self._timeout = timeout
181180
self._heartbeat_freq = heartbeat
182181
self._passive_heartbeat = passive_heartbeat
@@ -195,7 +194,7 @@ def _raise_remote_error(self, event):
195194
raise RemoteError('RemoteError', msg, None)
196195

197196
def _select_pattern(self, event):
198-
for pattern in self._patterns:
197+
for pattern in patterns.patterns_list:
199198
if pattern.accept_answer(event):
200199
return pattern
201200
msg = 'Unable to find a pattern for: {0}'.format(event)
@@ -211,7 +210,7 @@ def _process_response(self, method, bufchan, timeout):
211210

212211
pattern = self._select_pattern(event)
213212
return pattern.process_answer(self._context, bufchan, event, method,
214-
timeout, self._raise_remote_error)
213+
self._raise_remote_error)
215214
except:
216215
bufchan.close()
217216
bufchan.channel.close()
@@ -246,89 +245,6 @@ def __getattr__(self, method):
246245
return lambda *args, **kargs: self(method, *args, **kargs)
247246

248247

249-
class DecoratorBase(object):
250-
pattern = None
251-
252-
def __init__(self, functor):
253-
self._functor = functor
254-
self.__doc__ = functor.__doc__
255-
self.__name__ = functor.__name__
256-
257-
def __get__(self, instance, type_instance=None):
258-
if instance is None:
259-
return self
260-
return self.__class__(self._functor.__get__(instance, type_instance))
261-
262-
def __call__(self, *args, **kargs):
263-
return self._functor(*args, **kargs)
264-
265-
def _zerorpc_args(self):
266-
try:
267-
args_spec = self._functor._zerorpc_args()
268-
except AttributeError:
269-
try:
270-
args_spec = inspect.getargspec(self._functor)
271-
except TypeError:
272-
try:
273-
args_spec = inspect.getargspec(self._functor.__call__)
274-
except (AttributeError, TypeError):
275-
args_spec = None
276-
return args_spec
277-
278-
279-
class PatternReqRep():
280-
281-
def process_call(self, context, bufchan, event, functor):
282-
result = context.middleware_call_procedure(functor, *event.args)
283-
bufchan.emit('OK', (result,), context.middleware_get_task_context())
284-
285-
def accept_answer(self, event):
286-
return True
287-
288-
def process_answer(self, context, bufchan, event, method, timeout,
289-
raise_remote_error):
290-
result = event.args[0]
291-
if event.name == 'ERR':
292-
raise_remote_error(event)
293-
bufchan.close()
294-
bufchan.channel.close()
295-
bufchan.channel.channel.close()
296-
return result
297-
298-
299-
class rep(DecoratorBase):
300-
pattern = PatternReqRep()
301-
302-
303-
class PatternReqStream():
304-
305-
def process_call(self, context, bufchan, event, functor):
306-
xheader = context.middleware_get_task_context()
307-
for result in iter(context.middleware_call_procedure(functor,
308-
*event.args)):
309-
bufchan.emit('STREAM', result, xheader)
310-
bufchan.emit('STREAM_DONE', None, xheader)
311-
312-
def accept_answer(self, event):
313-
return event.name in ('STREAM', 'STREAM_DONE')
314-
315-
def process_answer(self, context, bufchan, event, method, timeout,
316-
raise_remote_error):
317-
def iterator(event):
318-
while event.name == 'STREAM':
319-
yield event.args
320-
event = bufchan.recv()
321-
if event.name == 'ERR':
322-
raise_remote_error(event)
323-
bufchan.close()
324-
bufchan.channel.channel.close()
325-
return iterator(event)
326-
327-
328-
class stream(DecoratorBase):
329-
pattern = PatternReqStream()
330-
331-
332248
class Server(SocketBase, ServerBase):
333249

334250
def __init__(self, methods=None, name=None, context=None, pool_size=None,
@@ -346,13 +262,12 @@ def close(self):
346262

347263

348264
class Client(SocketBase, ClientBase):
349-
patterns = [PatternReqStream(), PatternReqRep()]
350265

351266
def __init__(self, connect_to=None, context=None, timeout=30, heartbeat=5,
352267
passive_heartbeat=False):
353268
SocketBase.__init__(self, zmq.XREQ, context=context)
354-
ClientBase.__init__(self, self._events, Client.patterns, context,
355-
timeout, heartbeat, passive_heartbeat)
269+
ClientBase.__init__(self, self._events, context, timeout, heartbeat,
270+
passive_heartbeat)
356271
if connect_to:
357272
self.connect(connect_to)
358273

‎zerorpc/decorators.py‎

Lines changed: 65 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,65 @@
1+
# -*- coding: utf-8 -*-
2+
# Open Source Initiative OSI - The MIT License (MIT):Licensing
3+
#
4+
# The MIT License (MIT)
5+
# Copyright (c) 2012 DotCloud Inc ([email protected])
6+
#
7+
# Permission is hereby granted, free of charge, to any person obtaining a copy of
8+
# this software and associated documentation files (the "Software"), to deal in
9+
# the Software without restriction, including without limitation the rights to
10+
# use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies
11+
# of the Software, and to permit persons to whom the Software is furnished to do
12+
# so, subject to the following conditions:
13+
#
14+
# The above copyright notice and this permission notice shall be included in all
15+
# copies or substantial portions of the Software.
16+
#
17+
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
18+
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
19+
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
20+
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
21+
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
22+
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
23+
# SOFTWARE.
24+
25+
import inspect
26+
27+
from .patterns import *
28+
29+
30+
class DecoratorBase(object):
31+
pattern = None
32+
33+
def __init__(self, functor):
34+
self._functor = functor
35+
self.__doc__ = functor.__doc__
36+
self.__name__ = functor.__name__
37+
38+
def __get__(self, instance, type_instance=None):
39+
if instance is None:
40+
return self
41+
return self.__class__(self._functor.__get__(instance, type_instance))
42+
43+
def __call__(self, *args, **kargs):
44+
return self._functor(*args, **kargs)
45+
46+
def _zerorpc_args(self):
47+
try:
48+
args_spec = self._functor._zerorpc_args()
49+
except AttributeError:
50+
try:
51+
args_spec = inspect.getargspec(self._functor)
52+
except TypeError:
53+
try:
54+
args_spec = inspect.getargspec(self._functor.__call__)
55+
except (AttributeError, TypeError):
56+
args_spec = None
57+
return args_spec
58+
59+
60+
class rep(DecoratorBase):
61+
pattern = ReqRep()
62+
63+
64+
class stream(DecoratorBase):
65+
pattern = ReqStream()

‎zerorpc/patterns.py‎

Lines changed: 71 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,71 @@
1+
# -*- coding: utf-8 -*-
2+
# Open Source Initiative OSI - The MIT License (MIT):Licensing
3+
#
4+
# The MIT License (MIT)
5+
# Copyright (c) 2012 DotCloud Inc ([email protected])
6+
#
7+
# Permission is hereby granted, free of charge, to any person obtaining a copy of
8+
# this software and associated documentation files (the "Software"), to deal in
9+
# the Software without restriction, including without limitation the rights to
10+
# use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies
11+
# of the Software, and to permit persons to whom the Software is furnished to do
12+
# so, subject to the following conditions:
13+
#
14+
# The above copyright notice and this permission notice shall be included in all
15+
# copies or substantial portions of the Software.
16+
#
17+
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
18+
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
19+
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
20+
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
21+
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
22+
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
23+
# SOFTWARE.
24+
25+
26+
27+
class ReqRep():
28+
29+
def process_call(self, context, bufchan, event, functor):
30+
result = context.middleware_call_procedure(functor, *event.args)
31+
bufchan.emit('OK', (result,), context.middleware_get_task_context())
32+
33+
def accept_answer(self, event):
34+
return True
35+
36+
def process_answer(self, context, bufchan, event, method,
37+
raise_remote_error):
38+
result = event.args[0]
39+
if event.name == 'ERR':
40+
raise_remote_error(event)
41+
bufchan.close()
42+
bufchan.channel.close()
43+
bufchan.channel.channel.close()
44+
return result
45+
46+
47+
class ReqStream():
48+
49+
def process_call(self, context, bufchan, event, functor):
50+
xheader = context.middleware_get_task_context()
51+
for result in iter(context.middleware_call_procedure(functor,
52+
*event.args)):
53+
bufchan.emit('STREAM', result, xheader)
54+
bufchan.emit('STREAM_DONE', None, xheader)
55+
56+
def accept_answer(self, event):
57+
return event.name in ('STREAM', 'STREAM_DONE')
58+
59+
def process_answer(self, context, bufchan, event, method,
60+
raise_remote_error):
61+
def iterator(event):
62+
while event.name == 'STREAM':
63+
yield event.args
64+
event = bufchan.recv()
65+
if event.name == 'ERR':
66+
raise_remote_error(event)
67+
bufchan.close()
68+
bufchan.channel.channel.close()
69+
return iterator(event)
70+
71+
patterns_list = [ReqStream(), ReqRep()]

0 commit comments

Comments
 (0)