On Apr 20, 12:25 pm, Dun Peal dunpea...@gmail.com wrote:
Hi,
I'm writing and testing an asyncore-based server. Unfortunately, it
doesn't seem to work. The code below is based on the official docs and
examples, and starts a listening and sending dispatcher, where the
sending dispatcher connects and sends a message to the listener - yet
Handler.handle_read() never gets called, and I'm not sure why. Any
ideas?
Thanks, D.
import asyncore, socket, sys
COMM_PORT = 9345
class Handler(asyncore.dispatcher):
def handle_read(self):
print 'This never prints'
class Listener(asyncore.dispatcher):
def __init__(self, port=COMM_PORT):
asyncore.dispatcher.__init__(self)
self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
self.set_reuse_addr()
self.bind(('', port))
self.listen(5)
def handle_accept(self):
client, addr = self.accept()
print 'This prints.'
return Handler(client)
class Sender(asyncore.dispatcher):
def __init__(self, host):
asyncore.dispatcher.__init__(self)
self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
self.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self.connect( (host, COMM_PORT) )
self.buffer = 'Msg\r\n'
def handle_connect(self):
pass
def writable(self):
return len(self.buffer) 0
def handle_write(self):
sent = self.send(self.buffer)
self.buffer = self.buffer[sent:]
def test_communication():
from multiprocessing import Process
def listener():
l = Listener()
asyncore.loop(timeout=10, count=1)
lis = Process(target=listener)
lis.start()
def sender():
s = Sender('localhost')
asyncore.loop(timeout=10, count=1)
sen = Process(target=sender)
sen.start()
lis.join()
test_communication()
You didn't let the program run long enough for the later events to
happen. loop(count=1) basically means one I/O event will be processed
- in the case of your example, that's an accept(). Then asyncore is
done and it never gets to your custom handle_read.
So you can try passing a higher count to loop, or you can add your own
loop around the loop call. Or you can switch to Twisted which
actually makes testing a lot easier than this - no need to spawn
multiple processes or call accept or recv yourself. Here's a somewhat
equivalent Twisted-based version of your program:
from twisted.internet.protocol import ServerFactory, Protocol
from twisted.internet import reactor
factory = ServerFactory()
factory.protocol = Protocol
reactor.listenTCP(0, factory)
reactor.run()
It's hard to write the equivalent unit test, because the test you
wrote for the asyncore-based version is testing lots of low level
details which, as you can see, don't actually appear in the Twisted-
based version because Twisted does them for you already. However,
once you get past all that low-level stuff and get to the part where
you actually implement some of your application logic, you might have
tests for your protocol implementation that look something like this:
from twisted.trial.unittest import TestCase
from twisted.test.proto_helpers import StringTransport
from yourapp import Handler # Or a better name
class HandlerTests(TestCase):
def test_someMessage(self):
When the X message is received, the Y response is sent
back.
transport = StringTransport()
protocol = Handler()
protocol.makeConnection(transport)
protocol.dataReceived(X)
self.assertEqual(transport.value(), Y)
Hope this helps,
Jean-Paul
--
http://mail.python.org/mailman/listinfo/python-list
