publisher implementation

Author: DanielePalaia

examples/getting_started/main.py

@@ -1,9 +1,12 @@
+from proton import Message
+
 from rabbitmq_amqp_python_client import (
     BindingSpecification,
     Connection,
     ExchangeSpecification,
     QueueSpecification,
     QueueType,
+    exchange_address,
 )
 
 
@@ -31,29 +34,17 @@ def main() -> None:
         )
     )
 
-    """
     addr = exchange_address(exchange_name, routing_key)
-    """
 
-    """
-    publisher = connection.publisher(addr, "getting-started-publisher")
-    """
+    publisher = connection.publisher(addr)
 
-    """
-    message = Message(
-        body='test',
-        address='/queues/getting-started-exchangemessage',
-    )
+    publisher.publish(Message(body="test"))
 
-    publisher.Publish(message)
     publisher.close()
-    """
 
     # management.unbind(binding_exchange_queue_path)
 
-    """
-    management.purge_queue(queue_info.name)
-    """
+    # management.purge_queue(queue_info.name)
 
     # management.delete_queue(queue_name)
 

rabbitmq_amqp_python_client/__init__.py

@@ -1,12 +1,14 @@
 from importlib import metadata
 
+from .address_helper import exchange_address
 from .common import QueueType
 from .connection import Connection
 from .entities import (
     BindingSpecification,
     ExchangeSpecification,
     QueueSpecification,
 )
+from .publisher import Publisher
 
 try:
     __version__ = metadata.version(__package__)
@@ -23,4 +25,6 @@
     "QueueSpecification",
     "BindingSpecification",
     "QueueType",
+    "Publisher",
+    "exchange_address",
 ]

rabbitmq_amqp_python_client/address_helper.py

@@ -1,8 +1,11 @@
 from .entities import BindingSpecification
 
 
-def exchange_address(name: str) -> str:
-    path = "/exchanges/" + name
+def exchange_address(exchange_name: str, routing_key: str = "") -> str:
+    if routing_key == "":
+        path = "/exchanges/" + exchange_name
+    else:
+        path = "/exchanges/" + exchange_name + "/" + routing_key
 
     return path
 

rabbitmq_amqp_python_client/connection.py

@@ -3,6 +3,7 @@
 from proton.utils import BlockingConnection
 
 from .management import Management
+from .publisher import Publisher
 
 logger = logging.getLogger(__name__)
 
@@ -30,6 +31,10 @@ def management(self) -> Management:
     def close(self) -> None:
         self._conn.close()
 
+    def publisher(self, destination: str) -> Publisher:
+        publisher = Publisher(self._conn, destination)
+        return publisher
+
     # TODO: returns the current status of the connection.
     # def status(self) -> int:
     #    pass

rabbitmq_amqp_python_client/publisher.py

@@ -0,0 +1,41 @@
+import logging
+from typing import Optional
+
+from proton import Message
+from proton.utils import (
+    BlockingConnection,
+    BlockingReceiver,
+    BlockingSender,
+)
+
+from .options import SenderOption
+
+logger = logging.getLogger(__name__)
+
+
+class Publisher:
+    def __init__(self, conn: BlockingConnection, addr: str):
+        self._sender: Optional[BlockingSender] = None
+        self._receiver: Optional[BlockingReceiver] = None
+        self._conn = conn
+        self._addr = addr
+        self._open()
+
+    def _open(self) -> None:
+        print("addr is " + str(self._addr))
+        if self._sender is None:
+            logger.debug("Creating Sender")
+            self._sender = self._create_sender(self._addr)
+
+    def publish(self, message: Message) -> None:
+        if self._sender is not None:
+            self._sender.send(message)
+
+    def close(self) -> None:
+        if self._sender is not None:
+            self._sender.close()
+        # if self._receiver is not None:
+        #    self._receiver.close()
+
+    def _create_sender(self, addr: str) -> BlockingSender:
+        return self._conn.create_sender(addr, options=SenderOption(addr))