Update SelectConsumerThread to pass exceptions to self.error_func…

… to match `BlockingConsumerThread` Handle channel/connection closed exceptions explicitly in `SelectConsumerThread`
NeonGeckoCom · Jan 2, 2025 · be8aa96 · be8aa96
1 parent f845c3f
commit be8aa96
Showing 1 changed file with 8 additions and 0 deletions.
diff --git a/neon_mq_connector/consumers/select_consumer.py b/neon_mq_connector/consumers/select_consumer.py
@@ -206,9 +206,17 @@ def run(self):
                 super(SelectConsumerThread, self).run()
                 self.connection: pika.SelectConnection = self.create_connection()
                 self.connection.ioloop.start()
+            except (pika.exceptions.ChannelClosed,
+                    pika.exceptions.ConnectionClosed) as e:
+                LOG.info(f"Closed {e.reply_code}: {e.reply_text}")
+                if not self._stopping:
+                    # Connection was unexpectedly closed
+                    self._close_connection()
+                    self.error_func(self, e)
             except Exception as e:
                 LOG.error(f"Failed to start io loop on consumer thread {self.name!r}: {e}")
                 self._close_connection()
+                self.error_func(self, e)
 
     def _close_connection(self, mark_consumer_as_dead: bool = True):
         try: