[PROGRESS]

Author: ultimatecoder

tests/unit/test_subject.py

@@ -1,7 +1,7 @@
 import unittest
 from unittest import mock
 
-from yamq import subject, observer
+from yamq import subject, observer, message
 
 
 class TestSubjectSTOMP(unittest.TestCase):
@@ -60,9 +60,73 @@ def test_subscribers_are_preserved(self):
 
         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"
         })
+
+        obj_1.delete()
+        obj_2.delete()
+        obj_3.delete()
+
+    def test_get(self):
+        obj_1 = subject.SubjectSTOMP(name="test_queue_1", loop=self.loop)
+        returned_obj = subject.SubjectSTOMP.get(obj_1.name)
+        self.assertIs(obj_1, returned_obj)
+
+        returned_obj = subject.SubjectSTOMP.get("test_queue_2")
+        self.assertIsNone(returned_obj)
+
+        obj_1.delete()
+
+    def test_delete(self):
+        obj_1 = subject.SubjectSTOMP(name="test_queue_1", loop=self.loop)
+        returned_obj = subject.SubjectSTOMP.get("test_queue_1")
+
+        self.assertIs(obj_1, returned_obj)
+
+        obj_1.delete()
+        returned_obj = subject.SubjectSTOMP.get("test_queue_1")
+
+        self.assertIsNone(returned_obj)
+
+    def test_unsubscribe(self):
+        obj_1 = subject.SubjectSTOMP(name="test_queue_1", loop=self.loop)
+
+        observer_1 = observer.ObserverSTOMP(self.loop, self.transport)
+        observer_2 = observer.ObserverSTOMP(self.loop, self.transport)
+
+        obj_1.subscribe(observer_1, "auto")
+        obj_1.subscribe(observer_2, "auto")
+
+        self.assertDictEqual(obj_1.observers, {
+            observer_1: "auto",
+            observer_2: "auto"
+        })
+
+        obj_1.unsubscribe(observer_1)
+
+        self.assertDictEqual(obj_1.observers, {observer_2: "auto"})
+
+        obj_1.unsubscribe(observer_2)
+        self.assertDictEqual(obj_1.observers, {})
+
+        # It must call the subject_obj.delete() if there are no observers.
+        self.assertIsNone(subject.SubjectSTOMP.get('test_queue_1'))
+        obj_1.delete()
+
+    def test_z_notify(self):
+        obj_1 = subject.SubjectSTOMP(name="test_queue_1", loop=self.loop)
+
+        observer_1 = observer.ObserverSTOMP(self.loop, self.transport)
+        observer_2 = observer.ObserverSTOMP(self.loop, self.transport)
+
+        message_obj = message.Message("Test Message")
+
+        observer_1.subscribe(obj_1, "auto", "1")
+        observer_2.subscribe(obj_1, "auto", "2")
+
+        obj_1.notify(message_obj)
+
+        obj_1.delete()

yamq/observer.py

@@ -1,6 +1,6 @@
 from bidict import bidict
 
-from yamq import stomp
+import stomp
 
 
 class Observer:
@@ -21,7 +21,6 @@ def __init__(self, loop, 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
 
@@ -30,7 +29,6 @@ def unsubscribe(self, subscription_id):
         subject.unsubscribe(self)
 
     def update_auto(self, message_frame):
-        import pdb; pdb.set_trace()  # XXX BREAKPOINT
         self.transport(stomp.dumps(message_frame))
 
     def update_client(self, message_frame):
@@ -44,8 +42,7 @@ def delete(self):
             subject.unsubscribe(self)
 
     def update(self, subject, message, ack):
-        import pdb; pdb.set_trace()  # XXX BREAKPOINT
-        subscrption_id = self.subscriptions.inv[subject]
+        subscription_id = self.subscriptions.inv[subject]
         frame = stomp.MessageFrame(
             message=message,
             destination=subject.name,

yamq/server.py

@@ -52,13 +52,11 @@ def connect(self):
         pass
 
     def send(self, destination, raw_message, **headers):
-        import pdb; pdb.set_trace()  # XXX BREAKPOINT
         user_subject = subject.SubjectSTOMP(name=destination, loop=event_loop)
         message_obj = message.Message(raw_message)
         user_subject.notify(message_obj)
 
     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)
 

yamq/stomp/parse.py

@@ -22,7 +22,7 @@ def loads(raw_frame):
     return Frame(
         command=lines[0],
         headers=headers,
-        body=lines[body_start:]
+        body='\r\n'.join(lines[body_start:])
     )
 
 

yamq/subject.py

@@ -33,15 +33,12 @@ def __new__(cls, name, loop):
             obj = cls._objects[name]
         except KeyError as e:
             obj = super().__new__(cls)
+            obj.observers = {}
+            obj.name = name
+            obj.loop = loop
             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,)
 
@@ -65,9 +62,9 @@ def unsubscribe(self, observer):
             # TODO: I am not sure you should silently pass
             pass
 
-        if not self.objservers:
+        if not self.observers:
             self.delete()
 
     def notify(self, message):
-        for observer, ack_type in self.observers:
-            observer.update(message, ack_type, subject=self)
+        for observer, ack_type in self.observers.items():
+            observer.update(self, message, ack_type)