[PROGRESS]

Author: ultimatecoder

requirements.txt

@@ -0,0 +1 @@
+bidict==0.13.1

tests/unit/test_subject.py

@@ -0,0 +1,68 @@
+import unittest
+from unittest import mock
+
+from yamq import subject, observer
+
+
+class TestSubjectSTOMP(unittest.TestCase):
+
+    def setUp(self):
+        self.loop = unittest.mock.Mock()
+        self.transport = unittest.mock.Mock()
+
+    def test_object_creation(self):
+        obj_1 = subject.SubjectSTOMP(name="test_queue_1", loop=self.loop)
+
+        self.assertDictEqual(obj_1.observers, {})
+        obj_1.delete()
+
+    def test_subjects_are_unique(self):
+        self.assertDictEqual(subject.SubjectSTOMP._objects, {})
+
+        obj_1 = subject.SubjectSTOMP(name="test_queue_1", loop=self.loop)
+
+        self.assertDictEqual(subject.SubjectSTOMP._objects, {obj_1.name: obj_1})
+
+        obj_2 = subject.SubjectSTOMP(name="test_queue_2", loop=self.loop)
+
+        self.assertDictEqual(subject.SubjectSTOMP._objects, {
+            obj_1.name: obj_1,
+            obj_2.name: obj_2
+        })
+
+        obj_3 = subject.SubjectSTOMP(name=obj_1.name, loop=self.loop)
+
+        self.assertIs(obj_3, obj_1)
+        self.assertDictEqual(subject.SubjectSTOMP._objects, {
+            obj_1.name: obj_1,
+            obj_2.name: obj_2
+        })
+
+        obj_1.delete()
+        obj_2.delete()
+        obj_3.delete()
+
+    def test_subscribers_are_preserved(self):
+        observer_1 = observer.ObserverSTOMP(self.loop, self.transport)
+        observer_2 = observer.ObserverSTOMP(self.loop, self.transport)
+
+        obj_1 = subject.SubjectSTOMP(name="test_queue_1", loop=self.loop)
+        obj_1.subscribe(observer_1)
+        obj_1.subscribe(observer_2)
+
+        self.assertDictEqual(obj_1.observers, {
+            observer_1: "auto",
+            observer_2: "auto"
+        })
+
+        obj_2 = subject.SubjectSTOMP(name="test_queue_2", loop=self.loop)
+        obj_2.subscribe(observer_1)
+
+        self.assertDictEqual(obj_2.observers, {observer_1: "auto"})
+
+        import pdb; pdb.set_trace()  # XXX BREAKPOINT
+        obj_3 = subject.SubjectSTOMP(name="test_queue_1", loop=self.loop)
+        self.assertDictEqual(obj_3.observers, {
+            observer_1: "auto",
+            observer_2: "auto"
+        })

yamq/message.py

@@ -1,15 +1,13 @@
 from itertools import count
 
-from yamq import utils
-
 
 class Message():
     """Don't forget to call obj.delete() to delete object from the memory!"""
 
     _messages = {}
     _last_id = count()
 
-    def __init__(self, message, content_type):
+    def __init__(self, message, content_type='text/plain'):
         self._id = next(Message._last_id)
         Message._messages[self._id] = self
         self.message = message

yamq/observer.py

@@ -1,4 +1,6 @@
-import stomp
+from bidict import bidict
+
+from yamq import stomp
 
 
 class Observer:
@@ -16,16 +18,43 @@ class ObserverSTOMP:
     def __init__(self, loop, transport):
         self.loop = loop
         self.transport = transport
+        self.subscriptions = bidict()  # subscription_id: subject_obj
+
+    def subscribe(self, subject, ack_type, subscription_id):
+        import pdb; pdb.set_trace()  # XXX BREAKPOINT
+        subject.subscribe(self, ack_type)
+        self.subscriptions[subscription_id] = subject
+
+    def unsubscribe(self, subscription_id):
+        subject = self.subscriptions.get(subscription_id)
+        subject.unsubscribe(self)
+
+    def update_auto(self, message_frame):
+        import pdb; pdb.set_trace()  # XXX BREAKPOINT
+        self.transport(stomp.dumps(message_frame))
 
-    async def ack_auto(self, frame):
+    def update_client(self, message_frame):
         pass
 
-    async def update(self, subject, message, message_id, subscription_id, ack):
+    def update_client_individual(self, mesasge_frame):
+        pass
+
+    def delete(self):
+        for _, subject in self.subscriptions:
+            subject.unsubscribe(self)
+
+    def update(self, subject, message, ack):
+        import pdb; pdb.set_trace()  # XXX BREAKPOINT
+        subscrption_id = self.subscriptions.inv[subject]
         frame = stomp.MessageFrame(
-            subject,
-            message_id,
-            subscription_id,
-            message,
-            ack
+            message=message,
+            destination=subject.name,
+            subscription_id=subscription_id,
+            ack=ack
         )
-        pass
+        if ack == 'client-individual':
+            self.update_client_inidividual(frame)
+        elif ack == 'client':
+            self.update_client(frame)
+        else:
+            self.update_auto(frame)

yamq/server.py

@@ -7,24 +7,12 @@
 import observer
 import message
 
-SUBJECTS = {}
-#MESSAGES = {}
-
-#MESSAGE_ID = -1
-
-
-#def create_message(message):
-#    MESSAGE_ID += 1
-#    MESSAGES[MESSAGE_ID] = message
-#    return MESSAGE_ID
-
 
 class STOMP_Server(asyncio.Protocol):
     """Minimum supported version 1.2"""
 
     def connection_made(self, transport):
         self.transport = transport
-        self.subscription = {}
         self.observer = observer.ObserverSTOMP(event_loop, self.transport)
 
     def data_received(self, data):
@@ -64,31 +52,18 @@ def connect(self):
         pass
 
     def send(self, destination, raw_message, **headers):
-        user_subject = SUBJECTS.get(destination)
-        if not user_subject:
-            user_subject = subject.SubjectSTOMP(
-                name=destination, loop=event_loop
-            )
-            SUBJECTS[destination] = subject
+        import pdb; pdb.set_trace()  # XXX BREAKPOINT
+        user_subject = subject.SubjectSTOMP(name=destination, loop=event_loop)
         message_obj = message.Message(raw_message)
-        self.loop.call_soon(user_subject.notify(message_obj))
+        user_subject.notify(message_obj)
 
-    def subscribe(self, subject_id, destination, ack="auto", **headers):
-        user_subject = SUBJECTS.get(destination)
-        if not user_subject:
-            user_subject = subject.SubjectSTOMP(
-                name=destination, loop=event_loop
-            )
-            SUBJECTS[destination] = subject
-        self.subscription[subject_id] = subject
-        user_subject.subscribe(self.observer, subscrption_id, ack)
+    def subscribe(self, subscription_id, destination, ack="auto", **headers):
+        import pdb; pdb.set_trace()  # XXX BREAKPOINT
+        user_subject = subject.SubjectSTOMP(destination, loop=event_loop)
+        self.observer.subscribe(user_subject, ack, subscription_id)
 
-    def unsubscribe(self, subject_id, **headers):
-        user_subject = self.subscription.get(subject_id)
-        if not user_subject:
-            # TODO: Return Error frame here
-            pass
-        user_subject.unsubscribe(self.observer)
+    def unsubscribe(self, subscription_id, **headers):
+        self.objserver.unsubscribe(subscription_id)
 
     def ack(self, id, **headers):
         pass
@@ -97,8 +72,10 @@ def nack(self, id, **headers):
         pass
 
     def disconnect(self, receipt, **headers):
-        for user_subject in self.subscription.values():
-            user_subject.unsubscribe(self.observer)
+        self.observer.delete()
+
+    def connection_lost(self, exc):
+        self.observer.delete()
 
 
 #class YampServer(asyncio.Protocol):

yamq/stomp/frame.py

@@ -29,15 +29,15 @@ def __init__(self, version="1.2", heart_beat=None, session=None,
 
 class MessageFrame(Frame):
 
-    def __init__(self, destination, message_id, subscription_id, body="",
-                 ack=None):
+    def __init__(self, message, destination, subscription_id, ack=None):
         command = "MESSAGE"
         headers = {
             'destination': destination,
-            'message-id': message_id,
+            'message-id': message._id,
+            'content-type': message.content_type,
             'subscription': subscription_id
         }
 
         if ack:
             headers['ack'] = ack
-        super().__init__(command, headers, body)
+        super().__init__(command, headers, message.message)

yamq/subject.py

@@ -26,15 +26,36 @@ async def notify(self, message):
 
 class SubjectSTOMP:
 
+    _objects = {}
+
+    def __new__(cls, name, loop):
+        try:
+            obj = cls._objects[name]
+        except KeyError as e:
+            obj = super().__new__(cls)
+            cls._objects[name] = obj
+        return obj
+
     def __init__(self, name, loop):
+        print("Init called for : {}".format(name))
         self.observers = {}  # observer_object: ack_type
         self.name = name
         self.loop = loop
 
     def __repr__(self):
         return "<Subject object: %s>" % (self.name,)
 
-    def subscribe(self, observer, subscription_id, ack="auto"):
+    @classmethod
+    def get(cls, name):
+        return cls._objects.get(name)
+
+    def delete(self):
+        try:
+            del self.__class__._objects[self.name]
+        except KeyError:
+            pass
+
+    def subscribe(self, observer, ack="auto"):
         self.observers[observer] = ack
 
     def unsubscribe(self, observer):
@@ -44,9 +65,9 @@ def unsubscribe(self, observer):
             # TODO: I am not sure you should silently pass
             pass
 
+        if not self.objservers:
+            self.delete()
+
     def notify(self, message):
-        """Follows PUSH stratergy."""
-        for observer, ack in self.observers:
-            await observer.update(
-                self.name, message.message, message._id, ack
-            )
+        for observer, ack_type in self.observers:
+            observer.update(message, ack_type, subject=self)