fix CI error

Author: aeeeeeep

objwatch/sinks/consumer.py

@@ -68,7 +68,13 @@ def _connect(self) -> None:
             self.logger.info(f"Subscribed to topic: {self.topic.decode('utf-8') if self.topic else 'all topics'}")
         except zmq.ZMQError as e:
             self.logger.error(f"Failed to connect to ZeroMQ endpoint {self.endpoint}: {e}")
-            raise
+            # Clean up resources if partially initialized
+            if self.socket:
+                self.socket.close()
+                self.socket = None
+            if self.context:
+                self.context.term()
+                self.context = None
 
     def _disconnect(self) -> None:
         """
@@ -106,25 +112,23 @@ def _run(self) -> None:
         """
         The main run loop that listens for messages and writes them to file.
         """
-        self._connect()
-
         try:
             with open(self.output_file, 'a', encoding='utf-8') as f:
                 self.logger.info(f"Writing events to file: {self.output_file}")
 
                 while self.running:
                     try:
                         if self.socket is None:
-                            self.logger.error("Socket is None, attempting to reconnect")
+                            self.logger.info("Attempting to connect to ZeroMQ endpoint...")
                             self._connect()
                             if self.socket is None:
+                                self.logger.error("Failed to establish connection, will retry")
                                 time.sleep(0.1)
                                 continue
 
                         # Receive multipart message [topic, payload]
                         msg_parts = self.socket.recv_multipart()
                         if len(msg_parts) == 2:
-                            received_topic = msg_parts[0].decode('utf-8')
                             payload = msgpack.unpackb(msg_parts[1], raw=False)
 
                             # Process and write the event to file
@@ -134,6 +138,12 @@ def _run(self) -> None:
                     except zmq.Again:
                         # Timeout occurred, continue the loop
                         continue
+                    except zmq.ZMQError as e:
+                        self.logger.error(f"ZeroMQ error: {e}")
+                        # Reset socket to trigger reconnection
+                        self.socket = None
+                        time.sleep(0.1)
+                        continue
                     except Exception as e:
                         self.logger.error(f"Error processing message: {e}")
                         continue

objwatch/sinks/factory.py

@@ -18,7 +18,7 @@ def get_sink(config: ObjWatchConfig) -> BaseSink:
         BaseSink: The configured sink instance.
     """
     if config.output_mode == 'zmq':
-        return ZeroMQSink(endpoint=config.zmq_endpoint)
+        return ZeroMQSink(endpoint=config.zmq_endpoint, topic=config.zmq_topic)
     else:
         # Default to StandardSink
         # It handles output file internally if config.output is set

tests/test_zmq_e2e.py

@@ -22,7 +22,7 @@ def setUp(self):
         # Use a unique port for each test to avoid conflicts
         self.endpoint = "tcp://127.0.0.1:5555"
         self.topic = "test_topic"
-        self.consumer_output = tempfile.mktemp(suffix=".log")
+        self.consumer_output = tempfile.NamedTemporaryFile(suffix=".log", delete=False).name
 
         # Clean up any existing output file
         if os.path.exists(self.consumer_output):
@@ -126,7 +126,7 @@ def test_zmq_topic_filtering(self):
         Note: This test may fail occasionally due to ZeroMQ's asynchronous nature and SUB socket's "slow joiner" problem.
         """
         # Simplified test: create one consumer with a specific topic and send matching messages
-        consumer_output = tempfile.mktemp(suffix=".log")
+        consumer_output = tempfile.NamedTemporaryFile(suffix=".log", delete=False).name
 
         # Create consumer with topic "test_topic"
         consumer = ZeroMQFileConsumer(

tests/test_zmq_integration.py

@@ -70,7 +70,8 @@ def test_zmq_integration():
             with open(output, 'r') as f:
                 content = f.read()
 
-            print(f"Output file contains {content.count('\n')} lines")
+            newline = '\n'
+            print(f"Output file contains {content.count(newline)} lines")
             print("First 3 lines:")
             for line in content.split('\n')[:3]:
                 if line:
@@ -81,20 +82,20 @@ def test_zmq_integration():
             print(f"Found {test_messages_found}/5 test messages in the output file")
             if test_messages_found > 0:
                 print("Test PASSED: ZeroMQ integration works correctly!")
-                return True
+                assert True  # For pytest
             else:
                 print("Test FAILED: No test messages found in the output file")
-                return False
+                assert False  # For pytest
         else:
             print(f"Test FAILED: Output file {output} was not created")
-            return False
+            assert False  # For pytest
 
     except Exception as e:
         print(f"Test FAILED with exception: {e}")
         import traceback
 
         traceback.print_exc()
-        return False
+        assert False  # For pytest
 
     finally:
         # Clean up