[VOTR-185] autodelete component queues (#94)

* Revert "[Inf 250] feat(queue clean up): unbind component at shutdown (#86)"

This reverts commit 7099422.

* autodelete component queues

ergo/amqp_invoker.py

@@ -61,11 +61,12 @@ def __init__(self, invocable: FunctionInvocable) -> None:
         component_queue_name = f"{self._invocable.config.func}".replace("/", ":")
         if component_queue_name.startswith(":"):
             component_queue_name = component_queue_name[1:]
-        self._component_queue = kombu.Queue(name=component_queue_name, exchange=self._exchange, routing_key=str(SubTopic(self._invocable.config.subtopic)), durable=False, channel=self._connection.channel())
+        self._component_queue = kombu.Queue(name=component_queue_name, exchange=self._exchange, routing_key=str(SubTopic(self._invocable.config.subtopic)), durable=False, autodelete=True)
         instance_queue_name = f"{component_queue_name}:{instance_id()}"
         self._instance_queue = kombu.Queue(name=instance_queue_name, exchange=self._exchange, routing_key=str(SubTopic(instance_id())), auto_delete=True)
         error_queue_name = f"{component_queue_name}:error"
         self._error_queue = kombu.Queue(name=error_queue_name, exchange=self._exchange, routing_key=error_queue_name, durable=False)
+
         self._terminating = threading.Event()
         self._pending_invocations = threading.Semaphore()
         self._handler_lock = threading.Lock()
@@ -78,12 +79,8 @@ def start(self) -> int:
             consumer: kombu.Consumer = conn.Consumer(queues=[self._component_queue, self._instance_queue], prefetch_count=PREFETCH_COUNT, accept=["json"])
             consumer.register_callback(self._start_handle_message_thread)
             consumer.consume()
-
             while not self._terminating.is_set():
                 try:
-                    # lazy load binding
-                    # to prevent unbinding all consumers w same subtopic at shutdown
-                    self._component_queue.queue_bind(channel=self._connection.channel())
                     # wait up to 1s for the next message before sending a heartbeat
                     conn.drain_events(timeout=1)
                 except socket.timeout:
@@ -149,6 +146,5 @@ def _producer(self) -> kombu.Producer:
     def _shutdown(self, signum: int, *_: Any) -> None:
         self._terminating.set()
         self._pending_invocations.acquire(blocking=True, timeout=TERMINATION_GRACE_PERIOD)
-        self._component_queue.queue_unbind()
         self._connection.close()
         os.kill(os.getpid(), 0)

ergo/version.py

@@ -7,7 +7,7 @@
 import subprocess
 import sys
 
-VERSION = '0.9.2'
+VERSION = '0.10.0'
 
 
 def get_version() -> str: