Changeset 45 for ucomet.py

Timestamp:

02/08/08 19:24:35 (4 years ago)

Author:

mike

Message:

Got ucommet working

Files:

• ucomet.py (modified) (14 diffs)

ucomet.py

@@ -1 +1 @@
 from twisted.web2 import resource, http, http_headers, stream
-from random import random
+import random
 import string
 import cjson
@@ -14 +14 @@
     usedIds.add(None)
     @staticmethod
-    def getId(chars=32):
+    def makeId(chars=32):
         id = None
         while id in Utils.usedIds:
-            id = ''.join(random.sample(string.letters+string.digits, 12)) 
+            id = ''.join(random.sample(string.letters+string.digits, chars)) 
         Utils.usedIds.add(id)
         return id
@@ -40 +40 @@
                 jsonp = "jsonp"
         newstream = stream.ProducerStream()
-        self.streams.append(newstream)
-        if len(self.streams) > len(self.message_queue):
-            amsg = []
-        else:
+        self.streams.append((newstream,jsonp))
+        amsg = []
+        if len(self.streams) <= len(self.message_queue):
             amsg = self.message_queue.pop() 
         if addMessage: amsg.insert(0,addMessage)
         self.message_queue.append(amsg)
-        return newstream
+        print "Created Stream!!"
+        return makeResponse(newstream)
         
     
     def connect(self, message, args):
+        print "message connect  %s" % message
         msg = {
             "channel":   "/meta/connect",
@@ -58 +59 @@
             "advice":    {"reconnect": "retry"}
         }
-        return self.addStream(message, args, msg)
+        str = self.addStream(message, args, msg)
+        if len(self.message_queue[0]) > 1:
+            self.deliver()
+        return str
     
     def flushStreams(self):
@@ -71 +75 @@
                 self.message_queue[0].append(message)
         
-        if len(self.streams): #and len(self.message_queue[0]):
+        if len(self.streams) and len(self.message_queue[0]):
             stream,jsoncallback = self.streams.pop(0)
+            print "delivering %s" % ( "%s(%s)" % (jsoncallback,
+                                      encodejson(self.message_queue[0])))
             stream.write( "%s(%s)" % (jsoncallback,
-                                      encodejson(self.message_queue[0])))
+                                      encodejson(self.message_queue.pop(0))))
             
             stream.finish()
             del stream
+        else:
+            "might be delivering eventually %s" % message
 
 class ClientSubscription(object):
@@ -106 +114 @@
                  ClientSubscriptionType = ClientSubscription):
         self.ClientSubscriptionType = ClientSubscriptionType
-        self.connection = None
+        self.connection = Connection()
         self.subscriptions = {}
         if id:
@@ -114 +122 @@
             
     def connect(self,msg,args):
-        if not self.connection:
-            self.connection = Connection()
         return self.connection.connect(msg, args)
         
@@ -134 +140 @@
             return self._subunsubscribe(msg, args, nmsg)
         else:
-            return makeResponse(
-                encodejson([{"error":"already subscribed"}]),
-                           code = HTTP_ERROR)
+            return {"error":"already subscribed"}
         
     def unsubscribe(self,msg,args):
@@ -150 +154 @@
             return self._subunsubscribe(msg, args, nmsg)
         else:
-            return makeResponse(
-                encodejson([{"error":"not even subscribed"}]),
-                           code = HTTP_ERROR)
+            return {"error":"not even subscribed"}
     
     def dispatch(self,msg,args):
         channel = msg["channel"]
         data = msg["data"]
-        if channel in self.suscriptions:
-            self.subscriptions[channel].messageReceived(data)            
-        else:
-            return makeResponse(
-                encodejson([{"error":"not subscribed to %s" % channel}]),
-                           code = HTTP_ERROR)
-        
+        if channel in self.subscriptions:
+            #conn = self.connection.addStream(msg, args)
+            self.subscriptions[channel].messageReceived(data)  
+            return {"successful":True,"channel":channel}
+        else:
+            return {"error":"not subscribed to %s" % channel}        
         
     def _subunsubscribe(self,msg,args,nmsg):
         self.connection.flushStreams()
-        stream = self.connection.addStream(msg, args,nmsg)
-        self.connection.deliver()
-        return stream
+        #stream = self.connection.addStream(msg, args,nmsg)
+        #self.connection.deliver()
+        return nmsg
     
     
     def sendMessage(self, message, channel):
         if self.connection is not None:
-            self.connection.deliver({"data":message, "channel":channel})
+            self.connection.deliver({"data":message, "channel":channel, "clientId":self.id})
         else:
             #FIXME make own exception
@@ -186 +187 @@
     newHeaders.addRawHeader("Content-type", type)
     for (k,v) in headers.items(): newHeaders.addRawHeader(k, v)
+    print "RESPONSE STREAM: %s" % stream
     return http.Response(code=code, headers=newHeaders, stream=stream)
     
@@ -213 +215 @@
         if type(messages) is not list:
             return makeResponse("Messages must be encapsulated in array",
-                              type="text/plain", code=HTTP_ERROR)  
+                              type="text/plain", code=HTTP_ERROR)
+        print "messages %s" % messages
+        stream = None
+        responses = []
         connection = None
-        responses = []
         for m in messages:
             
@@ -224 +228 @@
                 clientId = m["clientId"]
                 if clientId in self.clients:
-                    client = self.client[client]
+                    client = self.clients[clientId]
                     connection = client.connection #it still may be None
             
-            if type(m) is dict:
+            if type(m) is not dict:
                 responses.append({"error":"each message must be in dict format"})
+                print("NOT DICT")
                 continue
             
             if "channel" not in m:
                 responses.append({"error":"channel is required in message"})
+                print("NO CHANNEL")
                 continue
             
@@ -238 +244 @@
             
             if channel == "/meta/handshake":
-                return self.handshake(args,m)
+                resp = self.handshake(m,args)
+                return resp
+            
+            
+            
+            
             if channel == "/meta/connect" or \
                channel == "/meta/reconnect":
-                connection = client.connect(args,m)
-            if channel == "/meta/subscribe":
-                connection = client.subscribe(args,m)
-            if channel == "/meta/unsubscribe":
-                connection = client.unsubscribe(args,m)
-            else: connection = client.dispatch(args,m)
-        
-        return connection
-            
-            
-        
-    def handshake(self, args, message):
+                print "M %s" %m
+                stream = client.connect(m,args)
+            elif channel == "/meta/subscribe":
+                responses.append(client.subscribe(m,args))
+            elif channel == "/meta/unsubscribe":
+                responses.append(client.unsubscribe(m,args))
+            else: 
+                responses.append(client.dispatch(m,args))
+        
+        if not stream or not client:
+            return makeResponse(encodejson(responses))
+        else:
+            for r in responses:
+                connection.deliver(r)
+        
+        return stream
+            
+            
+        
+    def handshake(self, message, args):
         #FIXME add error handlign if we're already hand shaken
         
         client = self.ClientType(ClientSubscriptionType=self.ClientSubscriptionType)
         self.clients[client.id] = client
-        return {
+        return makeResponse(encodejson([{
             "channel"                 :"/meta/handshake",
             "version"                 :self.version,
@@ -266 +285 @@
             "error"                   :None,
             "id"                      :message["id"]
-        }
-    
-            
-            
-            
-            
-    
-    
+        }]))
+    
+            
+            
+            
+            
+    
+