b75e83a35c
A new hub: This closely mirrors the poll hub with some of the internal logic changed to reflect zmq's flags. A green module for zmq: This subclasses Context and Socket to ensure calls are non blocking. A (very sparse) beginings of a test module. An example: A melding of the pyzmq chat example and the eventlet telnet chat example. TODO zmq_poll chokes if the sockets passed to it come from different contexts. As context is the entry point to everything else then it would make sense to include a check in here that each thread has only one context instance. By context being the entry point I mean: ctx = zmq.Context() socket = ctx.socket(zmq.<type-of-socket>) This call to socket is repeated for each socket you want and ctx must be the same one for each thread. Tests. I'd like to get to the point f having all zmq socket pairs tested - and perhaps a nice benchmark suite too.
64 lines
1.6 KiB
Python
64 lines
1.6 KiB
Python
import eventlet, sys
|
|
from eventlet.green import socket, zmq
|
|
from eventlet.hubs import use_hub
|
|
use_hub('zeromq')
|
|
|
|
ADDR = 'ipc:///tmp/chat'
|
|
|
|
ctx = zmq.Context()
|
|
|
|
def publish(writer):
|
|
|
|
print "connected"
|
|
socket = ctx.socket(zmq.SUB)
|
|
|
|
socket.setsockopt(zmq.SUBSCRIBE, "")
|
|
socket.connect(ADDR)
|
|
eventlet.sleep(0.1)
|
|
|
|
while True:
|
|
msg = socket.recv_pyobj()
|
|
str_msg = "%s: %s" % msg
|
|
writer.write(str_msg)
|
|
writer.flush()
|
|
|
|
|
|
PORT=3001
|
|
|
|
def read_chat_forever(reader, pub_socket):
|
|
|
|
line = reader.readline()
|
|
who = 'someone'
|
|
while line:
|
|
print "Chat:", line.strip()
|
|
if line.startswith('name:'):
|
|
who = line.split(':')[-1].strip()
|
|
|
|
try:
|
|
pub_socket.send_pyobj((who, line))
|
|
except socket.error, e:
|
|
# ignore broken pipes, they just mean the participant
|
|
# closed its connection already
|
|
if e[0] != 32:
|
|
raise
|
|
line = reader.readline()
|
|
print "Participant left chat."
|
|
|
|
try:
|
|
print "ChatServer starting up on port %s" % PORT
|
|
server = eventlet.listen(('0.0.0.0', PORT))
|
|
pub_socket = ctx.socket(zmq.PUB)
|
|
pub_socket.bind(ADDR)
|
|
eventlet.spawn_n(publish,
|
|
sys.stdout)
|
|
while True:
|
|
new_connection, address = server.accept()
|
|
|
|
print "Participant joined chat."
|
|
eventlet.spawn_n(publish,
|
|
new_connection.makefile('w'))
|
|
eventlet.spawn_n(read_chat_forever,
|
|
new_connection.makefile('r'),
|
|
pub_socket)
|
|
except (KeyboardInterrupt, SystemExit):
|
|
print "ChatServer exiting." |