[pypy-svn] r39717 - pypy/dist/pypy/lib/distributed

[fijal at codespeak.net]

Author: fijal
Date: Fri Mar  2 17:08:14 2007
New Revision: 39717

Modified:
   pypy/dist/pypy/lib/distributed/socklayer.py
Log:
Make socklayer working (provided greensock2 is on path)


Modified: pypy/dist/pypy/lib/distributed/socklayer.py
==============================================================================
--- pypy/dist/pypy/lib/distributed/socklayer.py	(original)
+++ pypy/dist/pypy/lib/distributed/socklayer.py	Fri Mar  2 17:08:14 2007
@@ -1,5 +1,6 @@
 
 from pypeers.pipe.gsocket import GreenSocket
+from pypeers.msgstruct import decodemessage, message
 from socket import socket, AF_INET, SOCK_STREAM
 import marshal
 import sys
@@ -9,29 +10,39 @@
     if TRACE:
         print >>sys.stderr, msg
 
-def receive(conn):
-    all = []
-    data = conn.recv(10000)
-    trace("received %s" % data)
-    return marshal.loads(data)
-
-def send(conn, data):
-    trace("sending %s" % (data,))
-    conn.send(marshal.dumps(data))
-    trace("done")
+class SocketWrapper(object):
+    def __init__(self, conn):
+        self.buffer = ""
+        self.conn = conn
+
+    def receive(self):
+        msg, self.buffer = decodemessage(self.buffer)
+        while msg is None:
+            self.buffer += self.conn.recv(8192)
+            msg, self.buffer = decodemessage(self.buffer)
+        assert msg[0] == 'c'
+        trace("received %s" % msg[1])
+        return marshal.loads(msg[1])
+
+    def send(self, data):
+        trace("sending %s" % (data,))
+        self.conn.sendall(message('c', marshal.dumps(data)))
+        trace("done")
 
-def socket_listener(address=('', 12121)):
+def socket_listener(address=('', 12122)):
     s = GreenSocket(AF_INET, SOCK_STREAM)
     s.bind(address)
     s.listen(1)
     print "Waiting for connection"
     conn, addr = s.accept()
-    
-    return lambda data : send(conn, data), lambda : receive(conn)
+
+    sw = SocketWrapper(conn)
+    return sw.send, sw.receive
 
 def socket_connecter(address):
     s = GreenSocket(AF_INET, SOCK_STREAM)
     print "Connecting %s" % (address,)
     s.connect(address)
 
-    return lambda data : send(s, data), lambda : receive(s)
+    sw = SocketWrapper(s)
+    return sw.send, sw.receive