at last - working json proxy

diff --git a/ProxyServer.py b/ProxyServer.py
index 18a22689e8a27fa7494ee46d7b7d8cacdc482bee..5f1a731f7eec9b471eed63953fd5e0927e8f3974 100644 (file)
--- a/ProxyServer.py
+++ b/ProxyServer.py
@@ -128,6 +128,7 @@ class ProxyServerRequestHandler(object):
         self.hr = args[0]
         print "on_query", reqtype, repr(self.hr.headers), str(self.hr.headers)
         if not hasattr(self.client, "proxy"):
+            print "new proxy"
             self.client.proxy = ProxyConnection()
             self.client.proxy.connect()
 
@@ -257,7 +258,6 @@ class ProxyServerRequestHandler(object):
             print 'proxy wants client to close_connection'
             try:
                 yield self.client.connectionClosed()
-                pass
             except httpd.ConnectionClosed:
                 print 'close_connection done'
                 pass

diff --git a/httpd.py b/httpd.py
index 6dc90c4acaa36816e9dea7b24990bdec198152de..eeee6e5b376d0e11b902514036cfd0ce32982a33 100644 (file)
--- a/httpd.py
+++ b/httpd.py
@@ -281,7 +281,13 @@ class Protocol(object):
             if _debug: print 'parse connection closed'
                     
     def writeMessage(self, message):
-        self.writeQueue.put(message)
+        try:
+            yield self.stream.write(message)
+        except ConnectionClosed:
+            yield self.connectionClosed()
+        except:
+            print traceback.print_exc()
+        #self.writeQueue.put(message)
             
     def parseHandshake(self):
         '''Parses the rtmp handshake'''
@@ -292,6 +298,8 @@ class Protocol(object):
     
     def parseRequests(self):
         '''Parses complete messages until connection closed. Raises ConnectionLost exception.'''
+
+        print "parseRequests start", repr(self)
         self.hr = MultitaskHTTPRequestHandler(self.stream, self.remote, None)
         self.hr.close_connection = 1
         self.cookies = CookieJar()
@@ -302,6 +310,9 @@ class Protocol(object):
             # over to "standard" HTTPRequestHandler, the data's already
             # there.
             print "parseRequests"
+            readok = (yield multitask.readable(self.stream.sock, 5000))
+            print "readok", readok
+            print
             raw_requestline = (yield self.stream.readline())
             if _debug: print "parseRequests, line", raw_requestline
             if not raw_requestline:
@@ -324,8 +335,9 @@ class Protocol(object):
                 raise ConnectionClosed 
 
             self.hr.raw_requestline = raw_requestline
-            pos = self.hr.rfile.tell()
-            #self.hr.rfile.truncate(0)
+            #pos = self.hr.rfile.tell()
+            pos = 0
+            self.hr.rfile.truncate(0)
             self.hr.rfile.write(data)
             print "parseRequests write after"
             self.hr.rfile.seek(pos)
@@ -344,6 +356,7 @@ class Protocol(object):
 
     def write(self):
         '''Writes messages to stream'''
+        print "starting protocol write loop", repr(self)
         while True:
             while self.writeQueue.empty(): (yield multitask.sleep(0.01))
             data = self.writeQueue.get() # TODO this should be used using multitask.Queue and remove previous wait.
@@ -364,6 +377,7 @@ class Protocol(object):
                 yield self.connectionClosed()
             except:
                 print traceback.print_exc()
+        print "ending protocol write loop", repr(self)
 
 class Command(object):
     ''' Class for command / data messages'''
@@ -489,7 +503,7 @@ class Client(Protocol):
         self.objectEncoding = 0.0
         self.queue = multitask.Queue() # receive queue used by application
         multitask.add(self.parse())
-        multitask.add(self.write())
+        #multitask.add(self.write())
 
     def recv(self):
         '''Generator to receive new Message (msg, arg) on this stream, or (None,None) if stream is closed.'''
@@ -498,8 +512,12 @@ class Client(Protocol):
     def connectionClosed(self):
         '''Called when the client drops the connection'''
         if _debug: 'Client.connectionClosed'
-        self.writeMessage(None)
-        yield self.queue.put((None,None))
+        if self.stream.sock:
+            yield self.stream.close()
+        else:
+            yield None
+        #self.writeMessage(None)
+        #yield self.queue.put((None,None))
             
     def messageReceived(self, msg):
         if _debug: print 'messageReceived cmd=', msg.command, msg.path
@@ -666,6 +684,7 @@ class HTTPServer(object):
                             inst = self.clients[session][0]
                         else:
                             inst = app()
+                        self.clients[session] = [inst]; inst._clients=self.clients[session]
                         msg.server = inst # whew! just in time!
                         try: 
                             methodname = "on%s" % msg.command
@@ -679,8 +698,6 @@ class HTTPServer(object):
                             yield client.rejectConnection(reason='Exception on %s' % methodname)
                             continue
                         if result is True or result is None:
-                            if session not in self.clients: 
-                                self.clients[session] = [inst]; inst._clients=self.clients[session]
                             if result is None:
                                 msg.wfile.seek(0)
                                 data = msg.wfile.read()
@@ -692,6 +709,7 @@ class HTTPServer(object):
                                         print 'close_connection requested'
                                     try:
                                         yield client.connectionClosed()
+                                        raise ConnectionClosed
                                     except ConnectionClosed:
                                         if _debug:
                                             print 'close_connection done'