Skip to content

Commit 5547843

Browse files
bombelabombela
committed
Standardize events objects
Also simplify WrappedEvents works.
1 parent ae784b2 commit 5547843

File tree

3 files changed

+63
-22
lines changed

3 files changed

+63
-22
lines changed

‎zerorpc/channel.py‎

Lines changed: 28 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -102,6 +102,10 @@ def channel(self, from_event=None):
102102
def active_channels(self):
103103
return self._active_channels
104104

105+
@property
106+
def context(self):
107+
return self._events.context
108+
105109

106110
class Channel(object):
107111

@@ -128,28 +132,33 @@ def close(self):
128132
del self._multiplexer._active_channels[self._channel_id]
129133
self._channel_id = None
130134

131-
def emit(self, name, args, xheader={}):
135+
def create_event(self, name, args, xheader={}):
132136
event = self._multiplexer.create_event(name, args, xheader)
133-
134137
if self._channel_id is None:
135138
self._channel_id = event.header['message_id']
136139
self._multiplexer._active_channels[self._channel_id] = self
137140
else:
138141
event.header['response_to'] = self._channel_id
142+
return event
143+
144+
def emit(self, name, args, xheader={}):
145+
event = self.create_event(name, args, xheader)
146+
self._multiplexer.emit_event(event, self._zmqid)
139147

140-
# TODO debug middleware
141-
# print time.time(), 'channel emit', event
148+
def emit_event(self, event):
142149
self._multiplexer.emit_event(event, self._zmqid)
143150

144151
def recv(self, timeout=None):
145152
try:
146153
event = self._queue.get(timeout=timeout)
147154
except gevent.queue.Empty:
148155
raise TimeoutExpired(timeout)
149-
# TODO debug middleware
150-
# print time.time(), 'channel recv', event
151156
return event
152157

158+
@property
159+
def context(self):
160+
return self._multiplexer.context
161+
153162

154163
class BufferedChannel(object):
155164

@@ -193,20 +202,27 @@ def _recver(self):
193202
else:
194203
self._input_queue.put(event)
195204

196-
def emit(self, name, args, xheader={}, block=True, timeout=None):
205+
def create_event(self, name, args, xheader={}):
206+
return self._channel.create_event(name, args, xheader)
207+
208+
def emit_event(self, event, block=True, timeout=None):
197209
if self._remote_queue_open_slots == 0:
198210
if not block:
199211
return False
200212
self._remote_can_recv.clear()
201213
self._remote_can_recv.wait(timeout=timeout)
202214
self._remote_queue_open_slots -= 1
203215
try:
204-
self._channel.emit(name, args, xheader)
216+
self._channel.emit_event(event)
205217
except:
206218
self._remote_queue_open_slots += 1
207219
raise
208220
return True
209221

222+
def emit(self, name, args, xheader={}, block=True, timeout=None):
223+
event = self.create_event(name, args, xheader)
224+
return self.emit_event(event, block, timeout)
225+
210226
def _request_data(self):
211227
open_slots = self._input_queue_size - self._input_queue_reserved
212228
self._input_queue_reserved += open_slots
@@ -230,3 +246,7 @@ def recv(self, timeout=None):
230246
@property
231247
def channel(self):
232248
return self._channel
249+
250+
@property
251+
def context(self):
252+
return self._channel.context

‎zerorpc/events.py‎

Lines changed: 20 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -113,7 +113,7 @@ def __init__(self, name, args, context, header=None):
113113
self._name = name
114114
self._args = args
115115
if header is None:
116-
context = context or Context.get_instance()
116+
context = context
117117
self._header = {
118118
'message_id': context.new_msgid(),
119119
'v': 3
@@ -257,6 +257,10 @@ def recv(self):
257257
def setsockopt(self, *args):
258258
return self._socket.setsockopt(*args)
259259

260+
@property
261+
def context(self):
262+
return self._context
263+
260264

261265
class WrappedEvents(object):
262266

@@ -271,18 +275,24 @@ def recv_is_available(self):
271275
return self._channel.recv_is_available
272276

273277
def create_event(self, name, args, xheader={}):
274-
wrapped_event = Event(name, args, None)
275-
wrapped_event.header.update(xheader)
276-
return wrapped_event
278+
event = Event(name, args, self._channel.context)
279+
event.header.update(xheader)
280+
return event
277281

278282
def emit_event(self, event, identity=None):
279-
return self._channel.emit('w', event.pack())
283+
event_payload = (event.header, event.name, event.args)
284+
wrapper_event = self._channel.create_event('w', event_payload)
285+
self._channel.emit_event(wrapper_event)
280286

281287
def emit(self, name, args, xheader={}):
282-
wrapped_event = self.create_event(name, args, xheader)
283-
return self._channel.emit('w', wrapped_event.pack())
288+
wrapper_event = self.create_event(name, args, xheader)
289+
self.emit_event(wrapper_event)
284290

285291
def recv(self, timeout=None):
286-
event = self._channel.recv()
287-
wrapped_event = Event.unpack(event.args)
288-
return wrapped_event
292+
wrapper_event = self._channel.recv()
293+
(header, name, args) = wrapper_event.args
294+
return Event(name, args, None, header)
295+
296+
@property
297+
def context(self):
298+
return self._channel.context

‎zerorpc/heartbeat.py‎

Lines changed: 15 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -95,12 +95,19 @@ def _lost_remote_exception(self):
9595
return LostRemote('Lost remote after {0}s heartbeat'.format(
9696
self._heartbeat_freq * 2))
9797

98-
def emit(self, name, args, xheader={}):
99-
if self._lost_remote:
100-
raise self._lost_remote_exception()
98+
def create_event(self, name, args, xheader={}):
10199
if self._compat_v2 and name == '_zpc_more':
102100
name = '_zpc_hb'
103-
self._channel.emit(name, args, xheader)
101+
return self._channel.create_event(name, args, xheader)
102+
103+
def emit_event(self, event):
104+
if self._lost_remote:
105+
raise self._lost_remote_exception()
106+
self._channel.emit_event(event)
107+
108+
def emit(self, name, args, xheader={}):
109+
event = self.create_event(name, args, xheader)
110+
self.emit_event(event)
104111

105112
def recv(self, timeout=None):
106113
if self._lost_remote:
@@ -116,3 +123,7 @@ def recv(self, timeout=None):
116123
@property
117124
def channel(self):
118125
return self._channel
126+
127+
@property
128+
def context(self):
129+
return self._channel.context

0 commit comments

Comments
 (0)