Lib/test/test_asynchat.py

   1 # test asynchat -- requires threading
   2
   3 import thread # If this fails, we can't test this module
   4 import asyncore, asynchat, socket, threading, time
   5
   6 HOST = "127.0.0.1"
   7 PORT = 54321
   8
   9 class echo_server(threading.Thread):
  10
  11     def run(self):
  12         sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  13         sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
  14         sock.bind((HOST, PORT))
  15         sock.listen(1)
  16         conn, client = sock.accept()
  17         buffer = ""
  18         while "\n" not in buffer:
  19             data = conn.recv(10)
  20             if not data:
  21                 break
  22             buffer = buffer + data
  23         while buffer:
  24             n = conn.send(buffer)
  25             buffer = buffer[n:]
  26         conn.close()
  27         sock.close()
  28
  29 class echo_client(asynchat.async_chat):
  30
  31     def __init__(self):
  32         asynchat.async_chat.__init__(self)
  33         self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
  34         self.connect((HOST, PORT))
  35         self.set_terminator("\n")
  36         self.buffer = ""
  37
  38     def handle_connect(self):
  39         print "Connected"
  40
  41     def collect_incoming_data(self, data):
  42         self.buffer = self.buffer + data
  43
  44     def found_terminator(self):
  45         print "Received:", `self.buffer`
  46         self.buffer = ""
  47         self.close()
  48
  49 def main():
  50     s = echo_server()
  51     s.start()
  52     time.sleep(1) # Give server time to initialize
  53     c = echo_client()
  54     c.push("hello ")
  55     c.push("world\n")
  56     asyncore.loop()
  57
  58 main()