Unify RabbitMQ handling with persistent, auto-reconnecting connections

detection_service/main.py

@@ -4,7 +4,7 @@
 from collections import defaultdict
 from scansynclib.logging import logger
 import pika
-from scansynclib.helpers import reconnect_rabbitmq, setup_rabbitmq_connection
+from scansynclib.rabbitmq import RabbitMQClient
 from scansynclib.config import config
 import json
 
@@ -94,7 +94,11 @@ def main():
 
     ensure_scan_directory_exists(SCAN_DIR)
     cleanup_dangling_documents()
-    connection, channel = setup_rabbitmq_connection(RABBITQUEUE)
+    client = RabbitMQClient(name="detection")
+    while not client.ensure_connection():
+        logger.warning("RabbitMQ is not available. Retrying detection startup in 5 seconds...")
+        time.sleep(5)
+    client.declare_queue(RABBITQUEUE)
 
     logger.info(f"Scanning {SCAN_DIR} for new files...")
     known_files = get_all_files(SCAN_DIR)
@@ -103,11 +107,17 @@ def main():
 
     while True:
         try:
-            # Check if the connection is still alive
-            if connection.is_closed or channel.is_closed:
-                connection, channel = reconnect_rabbitmq([RABBITQUEUE])
+            # Keep the single connection alive and reconnect if it was dropped.
+            if not client.is_open():
+                logger.warning("RabbitMQ connection lost. Reconnecting...")
+                if not client.ensure_connection():
+                    logger.warning("RabbitMQ reconnect failed. Retrying in 5 seconds...")
+                    time.sleep(5)
+                    continue
+                client.declare_queue(RABBITQUEUE)
             else:
-                connection.process_data_events(1)
+                # Give pika a chance to send heartbeats between scans.
+                client.process_events(1)
 
             current_files = get_all_files(SCAN_DIR)
             new_files = current_files - known_files
@@ -121,9 +131,10 @@ def main():
             if pending_files and last_file_time and (time.time() - last_file_time >= DUPLICATE_DETECTION_WINDOW):
                 logger.info(f"Processing {len(pending_files)} pending files after {DUPLICATE_DETECTION_WINDOW}s wait...")
                 grouped_files = group_files_by_content(pending_files)
-                publish_new_files(channel, RABBITQUEUE, grouped_files)
-                pending_files = []  # Pending-Liste leeren
-                last_file_time = None
+                if client.is_open():
+                    publish_new_files(client.channel, RABBITQUEUE, grouped_files)
+                    pending_files = []  # Pending-Liste leeren
+                    last_file_time = None
 
             known_files = current_files
 

docker-compose.yml

@@ -138,7 +138,7 @@ services:
       - app-network
 
   rabbitmq:
-    image: "rabbitmq:4.0.7-management"
+    image: "rabbitmq:4-management"
     restart: unless-stopped
     container_name: "rabbitmq"
     environment:

file_naming_service/main.py

@@ -4,8 +4,7 @@
 
 from scansynclib.ProcessItem import ProcessItem, ProcessStatus, FileNamingStatus
 from scansynclib.logging import logger
-from scansynclib.helpers import connect_rabbitmq, forward_to_rabbitmq
-import time
+from scansynclib.helpers import consume, forward_to_rabbitmq
 import pika.exceptions
 from scansynclib.openai_helper import generate_filename_openai
 from scansynclib.ollama_helper import generate_filename_ollama
@@ -92,37 +91,28 @@ def callback(ch, method, properties, body):
         logger.exception(f"Failed processing {item.filename}.")
         execute_query("UPDATE file_naming_jobs SET file_naming_status = ?, error_description = ?, finished = DATETIME('now', 'localtime') WHERE id = ?", (FileNamingStatus.FAILED.name, str(e), item.file_naming_db_id))
     finally:
+        ack_ok = True
         try:
             ch.basic_ack(delivery_tag=method.delivery_tag)
-        except pika.exceptions.AMQPConnectionError:
-            logger.error("Connection lost while acknowledging message. Reconnecting...")
-            connection, channel = connect_rabbitmq([RABBITQUEUE], heartbeat=120)
-            channel.basic_ack(delivery_tag=method.delivery_tag)
-        if isinstance(item, ProcessItem):
-            item_file_naming_db_id = getattr(item, "file_naming_db_id", None)
-            if item_file_naming_db_id:
-                item.file_naming_status = get_latest_file_naming_status(item)
-            item.status = ProcessStatus.SYNC_PENDING
-            update_scanneddata_database(item, {"file_status": item.status.value})
-            forward_to_rabbitmq("upload_queue", item)
-        else:
-            logger.error("Item is None, cannot forward to upload queue.")
+        except pika.exceptions.AMQPError:
+            ack_ok = False
+            # The connection was lost before we could acknowledge. The unified
+            # consumer will reconnect and the broker will redeliver the message.
+            logger.error("Connection lost while acknowledging message. It will be redelivered after reconnect.")
+        if ack_ok:
+            if isinstance(item, ProcessItem):
+                item_file_naming_db_id = getattr(item, "file_naming_db_id", None)
+                if item_file_naming_db_id:
+                    item.file_naming_status = get_latest_file_naming_status(item)
+                item.status = ProcessStatus.SYNC_PENDING
+                update_scanneddata_database(item, {"file_status": item.status.value})
+                forward_to_rabbitmq("upload_queue", item)
+            else:
+                logger.error("Item is None, cannot forward to upload queue.")
 
 
 def start_consuming_with_reconnect():
-    global channel, connection
-    while True:
-        try:
-            connection, channel = connect_rabbitmq([RABBITQUEUE], heartbeat=120)
-            channel.basic_consume(queue=RABBITQUEUE, on_message_callback=callback)
-            logger.info("File naming service started, waiting for messages...")
-            channel.start_consuming()
-        except pika.exceptions.AMQPConnectionError as e:
-            logger.error(f"Connection lost: {e}. Reconnecting in 5 seconds...")
-            time.sleep(5)
-        except Exception as e:
-            logger.exception(f"Unexpected error: {e}. Restarting consumer...")
-            time.sleep(5)
+    consume(RABBITQUEUE, callback, heartbeat=120)
 
 
 if __name__ == "__main__":

metadata_service/main.py

@@ -5,17 +5,14 @@
 from scansynclib.ProcessItem import ItemType, ProcessItem, ProcessStatus, OneDriveDestination
 from PIL import Image
 from pypdf import PdfReader
-import pika
-import pika.exceptions
 from scansynclib.sqlite_wrapper import execute_query, update_scanneddata_database
-from scansynclib.helpers import connect_rabbitmq, move_to_failed
+from scansynclib.helpers import consume, publish, move_to_failed
 from scansynclib.config import config
 import pymupdf
 import pickle
 
 RABBITQUEUE = "metadata_queue"
 TIMEOUT_PDF_VALIDATION = 300
-channel, connection = None, None
 
 
 def on_created(filepaths: list):
@@ -165,12 +162,7 @@ def on_created(filepaths: list):
             logger.exception(f"Error reading PDF file: {item.local_file_path}")
     item.status = ProcessStatus.OCR_PENDING
     update_scanneddata_database(item, {"file_status": item.status.value})
-    channel.basic_publish(
-                    exchange="",
-                    routing_key="ocr_queue",
-                    body=pickle.dumps(item),
-                    properties=pika.BasicProperties(delivery_mode=2)
-                )
+    publish("ocr_queue", pickle.dumps(item))
     logger.info(f"Added {item.local_file_path} to OCR queue")
 
 
@@ -236,19 +228,7 @@ def callback(ch, method, properties, body):
 
 
 def start_consuming_with_reconnect():
-    global channel, connection
-    while True:
-        try:
-            connection, channel = connect_rabbitmq([RABBITQUEUE, "ocr_queue"], heartbeat=600)
-            channel.basic_consume(queue=RABBITQUEUE, on_message_callback=callback)
-            logger.info("Metadata service started, waiting for messages...")
-            channel.start_consuming()
-        except pika.exceptions.AMQPConnectionError as e:
-            logger.error(f"Connection lost: {e}. Reconnecting in 5 seconds...")
-            time.sleep(5)
-        except Exception as e:
-            logger.exception(f"Unexpected error: {e}. Restarting consumer...")
-            time.sleep(5)
+    consume([RABBITQUEUE, "ocr_queue"], callback, heartbeat=600)
 
 
 # Start the consumer with reconnect logic

ocr_service/main.py

@@ -1,13 +1,11 @@
 from scansynclib.logging import logger
 from scansynclib.ProcessItem import ProcessItem, ProcessStatus, OCRStatus
 from scansynclib.sqlite_wrapper import execute_query, update_scanneddata_database
-from scansynclib.helpers import connect_rabbitmq, forward_to_rabbitmq, extract_text
+from scansynclib.helpers import consume, forward_to_rabbitmq, extract_text
 import pickle
 import ocrmypdf
 import os
 from datetime import datetime
-import time
-import pika.exceptions
 from scansynclib.settings import settings
 
 logger.info("Starting OCR service...")
@@ -124,18 +122,7 @@ def start_processing(item: ProcessItem):
 
 
 def start_consuming_with_reconnect():
-    while True:
-        try:
-            connection, channel = connect_rabbitmq([RABBITQUEUE], heartbeat=600)
-            channel.basic_consume(queue=RABBITQUEUE, on_message_callback=callback)
-            logger.info("OCR service started, waiting for messages...")
-            channel.start_consuming()
-        except pika.exceptions.AMQPConnectionError as e:
-            logger.error(f"Connection lost: {e}. Reconnecting in 5 seconds...")
-            time.sleep(5)
-        except Exception as e:
-            logger.exception(f"Unexpected error: {e}. Restarting consumer...")
-            time.sleep(5)
+    consume(RABBITQUEUE, callback, heartbeat=600)
 
 
 # Start the consumer with reconnect logic

scansynclib/scansynclib.egg-info/SOURCES.txt

@@ -1,16 +1,17 @@
 pyproject.toml
-./scansynclib/ProcessItem.py
-./scansynclib/__init__.py
-./scansynclib/config.py
-./scansynclib/helpers.py
-./scansynclib/logging.py
-./scansynclib/ollama_helper.py
-./scansynclib/onedrive_api.py
-./scansynclib/onedrive_smb_manager.py
-./scansynclib/openai_helper.py
-./scansynclib/settings.py
-./scansynclib/settings_schema.py
-./scansynclib/sqlite_wrapper.py
+scansynclib/ProcessItem.py
+scansynclib/__init__.py
+scansynclib/config.py
+scansynclib/helpers.py
+scansynclib/logging.py
+scansynclib/ollama_helper.py
+scansynclib/onedrive_api.py
+scansynclib/onedrive_smb_manager.py
+scansynclib/openai_helper.py
+scansynclib/rabbitmq.py
+scansynclib/settings.py
+scansynclib/settings_schema.py
+scansynclib/sqlite_wrapper.py
 scansynclib.egg-info/PKG-INFO
 scansynclib.egg-info/SOURCES.txt
 scansynclib.egg-info/dependency_links.txt

scansynclib/scansynclib/helpers.py

@@ -1,14 +1,20 @@
 from datetime import datetime, timedelta
 import os
-import pickle
 import re
-import pika
-import socket
-import time
 from scansynclib.ProcessItem import ProcessItem
 from scansynclib.config import config
-import pika.exceptions
 from scansynclib.logging import logger
+# RabbitMQ handling now lives in the unified scansynclib.rabbitmq module. The
+# functions are re-exported here for backwards compatibility with existing
+# imports (e.g. ``from scansynclib.helpers import forward_to_rabbitmq``).
+from scansynclib.rabbitmq import (  # noqa: F401
+    RabbitMQClient,
+    connect_rabbitmq,
+    consume,
+    forward_to_rabbitmq,
+    publish,
+    publish_to_exchange,
+)
 from pypdf import PdfReader
 
 SMB_TAG_COLORS = [
@@ -32,80 +38,6 @@
         ]
 
 
-def connect_rabbitmq(queue_names: list = None, heartbeat: int = 30):
-    """
-    Establishes a connection to a RabbitMQ server and declares multiple queues.
-
-    This function attempts to connect to a RabbitMQ server up to 10 times,
-    with a 2-second delay between each attempt. If the connection is
-    successful, it declares durable queues with the specified names.
-
-    Args:
-        queue_names (list): A list of RabbitMQ queue names to declare.
-        heartbeat (int): The heartbeat timeout in seconds for the RabbitMQ connection.
-
-    Returns:
-        tuple: A tuple containing the RabbitMQ connection and channel objects
-               if the connection is successful.
-        None: If the connection could not be established after 10 attempts.
-
-    Raises:
-        None: The function handles `socket.gaierror` and
-              `pika.exceptions.AMQPConnectionError` internally.
-    """
-    for i in range(10):
-        try:
-            connection = pika.BlockingConnection(pika.ConnectionParameters(host="rabbitmq", heartbeat=heartbeat))
-            channel = connection.channel()
-            if queue_names:
-                for queue_name in queue_names:
-                    channel.queue_declare(queue=queue_name, durable=True)
-            channel.basic_qos(prefetch_count=1)
-            return connection, channel
-        except (socket.gaierror, pika.exceptions.AMQPConnectionError):
-            time.sleep(2)
-    logger.critical("Couldn't connect to RabbitMQ.")
-    return None
-
-
-def setup_rabbitmq_connection(queue_name):
-    try:
-        connection, channel = connect_rabbitmq([queue_name])
-        logger.debug(f"Connected to RabbitMQ on {channel.channel_number}")
-        logger.debug(f"Connected to queue {queue_name}")
-        return connection, channel
-    except Exception as e:
-        logger.critical(f"Failed to connect to RabbitMQ: {e}")
-        exit(1)
-
-
-def reconnect_rabbitmq(queue_name):
-    while True:
-        try:
-            logger.warning("Attempting to reconnect to RabbitMQ...")
-            connection, channel = setup_rabbitmq_connection(queue_name)
-            logger.info("Reconnected to RabbitMQ successfully.")
-            return connection, channel
-        except Exception as e:
-            logger.critical(f"Failed to reconnect to RabbitMQ: {e}")
-            time.sleep(5)  # Wait before retrying
-
-
-def forward_to_rabbitmq(queue_name: str, item: ProcessItem):
-    try:
-        connection, channel = connect_rabbitmq([queue_name])
-        channel.basic_publish(
-            exchange='',
-            routing_key=queue_name,
-            body=pickle.dumps(item),
-            properties=pika.BasicProperties(delivery_mode=2)  # Make message persistent
-        )
-        logger.info(f"Item {item.filename} forwarded to {queue_name}.")
-        connection.close()
-    except Exception as e:
-        logger.error(f"Failed to forward item {item.filename} to RabbitMQ queue {queue_name}: {e}")
-
-
 def parse_timestamp(timestamp: str) -> datetime:
     try:
         return datetime.strptime(timestamp, "%Y-%m-%d %H:%M:%S")