mirror of
https://github.com/wassname/catalyst.git
synced 2026-07-12 19:26:32 +08:00
Removes vestigial code for zmq based generators.
This commit is contained in:
@@ -1,18 +0,0 @@
|
||||
import zmq
|
||||
|
||||
import zipline.protocol as zp
|
||||
|
||||
def gen_from_zmq(poller, unframe, namestring):
|
||||
"""
|
||||
A generator that takes an initialized zmq poller and yields
|
||||
messages from the poller until it gets a zp.CONTROL_PROTOCOL.DONE.
|
||||
"""
|
||||
while True:
|
||||
message = poller.recv()
|
||||
# Done protocol should now be a message type so that
|
||||
# done messages can also have source_ids.
|
||||
if message.type == zp.CONTROL_PROTOCOL.DONE:
|
||||
yield done_message(message.source_id)
|
||||
break
|
||||
else:
|
||||
yield unframe(message)
|
||||
@@ -1,19 +0,0 @@
|
||||
import zmq
|
||||
import zipline.protocol as zp
|
||||
|
||||
def gen_from_pull_socket(socket_uri, context, unframe):
|
||||
"""
|
||||
A generator that takes a socket_uri, and yields
|
||||
messages from the poller until it gets a zp.CONTROL_PROTOCOL.DONE.
|
||||
"""
|
||||
pull_socket = context.socket(zmq.PULL)
|
||||
pull_socket.connect(socket_uri)
|
||||
poller = zmq.Poller()
|
||||
poller.register(pull_socket, zmq.POLLIN)
|
||||
|
||||
return gen_from_poller(poller, pull_socket, unframe)
|
||||
|
||||
|
||||
# this generator needs to know about the source_ids coming in via
|
||||
# the poller, and need to yield DONE messages for each
|
||||
# source_id.
|
||||
Reference in New Issue
Block a user