add: 数据监听功能

src/queue_sqlite/constant/message_type.py

@@ -2,6 +2,5 @@ from enum import Enum
 
 
 class MessageType(Enum):
-    NOTIFICATION = "notification"
+    TASK = "task" # 任务
-    SYSTEM = "system"
+    LISTEN = "listen" # 监听
-    TASK = "task"

src/queue_sqlite/core/__init__.pyi

@@ -1,4 +1,4 @@
-from ..task_cycle import TaskMounter
+from ..mounter import TaskMounter
 from typing import Callable, List
 from ..constant import MessageStatus
 

src/queue_sqlite/mounter/__init__.py

@@ -1,5 +1,3 @@
 from .task_mounter import TaskMounter
 
-__all__ = [
+__all__ = ["TaskMounter"]
-    "TaskMounter"
-]

src/queue_sqlite/mounter/base_mounter.py

@@ -0,0 +1,42 @@
+from typing import Callable, List
+
+
+class BaseMounter:
+    @classmethod
+    def mount_(cls, function: Callable):
+        setattr(cls, function.__name__, function)
+    
+    @staticmethod
+    def mount(meta: dict = {}):
+        """带参数的装饰器"""
+        def decorator(function: Callable):
+            # 使用自定义名称或函数原名
+            function.meta = meta # type: ignore
+            setattr(BaseMounter, function.__name__, function)
+            return function
+        return decorator
+    
+    @classmethod
+    def get_function(cls, name: str):
+        return getattr(cls, name, None)
+
+    @classmethod
+    def get_task_list(cls) -> List[str]:
+        """获取所有挂载的任务函数名称列表"""
+        task_list = []
+        
+        # 遍历类属性字典
+        for attr_name, attr_value in vars(cls).items():
+            # 过滤条件：
+            # 1. 必须是可调用对象（函数）
+            # 2. 不是类自带的特殊方法（非双下划线开头）
+            # 3. 不是类方法本身（如 mount_, get_task_list 等）
+            if (
+                callable(attr_value) 
+                and not attr_name.startswith("__") 
+                and attr_name not in ["mount_", "mount", "get_function", "get_task_list"]
+            ):
+                task_list.append(attr_name)
+                
+        return task_list
+    

src/queue_sqlite/mounter/listen_mounter.py

@@ -0,0 +1,41 @@
+from typing import Callable, List
+
+
+class ListenMounter:
+    @classmethod
+    def mount_Listener(cls, function: Callable):
+        setattr(cls, function.__name__, function)
+    
+    @staticmethod
+    def listener():
+        """带参数的装饰器"""
+        def decorator(function: Callable):
+            # 使用自定义名称或函数原名
+            setattr(ListenMounter, function.__name__, function)
+            return function
+        return decorator
+    
+    @classmethod
+    def get_Listener_function(cls, name: str):
+        return getattr(cls, name, None)
+
+    @classmethod
+    def get_Listener_list(cls) -> List[str]:
+        """获取所有挂载的监听函数名称列表"""
+        listener_list = []
+        
+        # 遍历类属性字典
+        for attr_name, attr_value in vars(cls).items():
+            # 过滤条件：
+            # 1. 必须是可调用对象（函数）
+            # 2. 不是类自带的特殊方法（非双下划线开头）
+            # 3. 不是类方法本身（如 mount_Listener, listener 等）
+            if (
+                callable(attr_value) 
+                and not attr_name.startswith("__") 
+                and attr_name not in ["mount_Listener", "listener", "get_Listener_function", "get_Listener_list"]
+            ):
+                listener_list.append(attr_name)
+                
+        return listener_list
+    

src/queue_sqlite/queue_operation/listen_operation.py

@@ -0,0 +1,71 @@
+from ..mounter.listen_mounter import ListenMounter
+import sqlite3
+import threading
+
+
+class ListenOperation:
+    def __init__(self, db_dir):
+        self.db_dir = db_dir
+        self.listen_fields = ListenMounter.get_Listener_list()
+        self.create_table()
+
+    def create_table(self):
+        if len(self.listen_fields) == 0:
+            return
+        sql = f"""
+            CREATE TABLE IF NOT EXISTS listen_table (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                key Text,
+                value Text
+            );
+        """
+        self.conn = sqlite3.connect(
+            self.db_dir,
+            check_same_thread = False,
+        )
+        self.conn.execute(sql)
+        self.conn.execute("PRAGMA journal_mode=WAL;")
+        self.conn.execute("PRAGMA synchronous=NORMAL;")
+        self.conn.execute("PRAGMA cache_size=-20000;")
+        self.conn.execute("PRAGMA mmap_size=1073741824;")
+        self.conn.execute("PRAGMA temp_store=MEMORY;")
+        self.conn.execute("PRAGMA busy_timeout=5000;")
+        self.conn.commit()
+
+        # 删除原有数据
+        sql = f"""
+            DELETE FROM listen_table
+        """
+        self.conn.execute(sql)
+        self.conn.commit()
+
+        for listen_field in self.listen_fields:
+            sql = f"""
+                INSERT INTO 
+                    listen_table (key, value)
+                VALUES 
+                    (?, ?)
+            """
+            self.conn.execute(sql, (listen_field, "null"))
+            self.conn.commit()
+    
+    def listen_data(self, key, value) -> tuple[bool, str]:
+        sql = f"""
+            Select value from listen_table where key = '{key}'
+        """
+        cursor = self.conn.execute(sql)
+        result = cursor.fetchone()
+        if result is None:
+            return False, "key not found"
+        if result[0] == value:
+            return False, "value not changed"
+        else:
+            return True, result[0]
+        
+    def update_listen_data(self, key, value):
+        sql = f"""
+            UPDATE listen_table SET value = '{value}' WHERE key = '{key}'
+        """
+        self.conn.execute(sql)
+        self.conn.commit()
+

src/queue_sqlite/queue_operation/queue_operation.py

@@ -8,7 +8,6 @@ from ..core import core
 from ..constant import MessageStatus
 
 
-
 class QueueOperation():
     def __init__(self, shard_num: int = 4, queue_name: str = "default"):
         self.shard_num = shard_num

src/queue_sqlite/scheduler/__init__.py

@@ -1,13 +1,20 @@
 from .receive_scheduler import ReceiveScheduler
 from .task_scheduler import TaskScheduler
+from .listen_data_scheduler import ListenDataScheduler
 from ..model import MessageItem
 from typing import Callable
 from ..queue_operation import QueueOperation
+from ..queue_operation.listen_operation import ListenOperation
 from .cleanup_scheduler import CleanupScheduler
+import os
 
 class QueueScheduler:
     def __init__(self, receive_thread_num=1, task_thread_num=1, shard_num=4):
         self.queue_operation = QueueOperation(shard_num)
+        self.listen_operation = ListenOperation(os.path.join(self.queue_operation.db_dir, "listen.db"))
+        self.listen_operation.create_table()
+        
+        self.listen_scheduler = ListenDataScheduler(self.listen_operation)
         self.receive_scheduler = ReceiveScheduler(self.queue_operation, receive_thread_num)
         self.task_scheduler = TaskScheduler(self.queue_operation, task_thread_num)
         self.cleanup_scheduler = CleanupScheduler(self.queue_operation)
@@ -15,13 +22,18 @@ class QueueScheduler:
     def send_message(self, message: MessageItem, callback: Callable):
         self.receive_scheduler.send_message(message, callback)
 
+    def update_listen_data(self, key, value):
+        self.listen_operation.update_listen_data(key, value)
+
     def start_queue_scheduler(self):
         self.receive_scheduler.start_receive_thread()
         self.task_scheduler.start_task_thread()
         self.cleanup_scheduler.start_cleanup()
+        self.listen_scheduler.start_listen_data()
 
     def stop_queue_scheduler(self):
         self.receive_scheduler.stop_receive_thread()
+        self.listen_scheduler.stop_listen_data()
         self.task_scheduler.stop_task_thread()
         self.cleanup_scheduler.stop_cleanup()
         # self.queue_operation.close_all_connections()

src/queue_sqlite/scheduler/listen_data_scheduler.py

@@ -0,0 +1,46 @@
+from ..queue_operation.listen_operation import ListenOperation
+from concurrent.futures import ThreadPoolExecutor
+from ..mounter.listen_mounter import ListenMounter
+import threading
+
+
+class ListenDataScheduler:
+    def __init__(self, listen_operation: ListenOperation):
+        self.listen_operation = listen_operation
+        self.is_running = False
+        self.executor = ThreadPoolExecutor(max_workers=1)
+        self.listen_thread = None
+        self.listen_data = {field: "null" for field in ListenMounter.get_Listener_list()}
+
+    def _process_listen_data(self, key, value):
+        listen_function = ListenMounter.get_Listener_function(key)
+        if listen_function:
+            try:
+                listen_function(value)
+            except Exception as e:
+                ValueError(f"Error in {key} listener function: {e}")
+
+    def listen(self):
+        while self.is_running:
+            for key, value in self.listen_data.items():
+                status, new_value = self.listen_operation.listen_data(key, value)
+                if status:
+                    self.listen_data[key] = new_value
+                    self.executor.submit(self._process_listen_data, key, new_value)
+
+    def start_listen_data(self):
+        if self.is_running:
+            return
+        self.is_running = True
+        self.listen_thread = threading.Thread(target=self.listen)
+        self.listen_thread.start()
+
+    def stop_listen_data(self):
+        if not self.is_running:
+            return
+        self.is_running = False
+        if self.listen_thread and self.listen_thread.is_alive():
+            self.listen_thread.join(timeout=2)
+
+        self.executor.shutdown(wait=True)
+    

src/queue_sqlite/scheduler/task_scheduler.py

@@ -1,8 +1,8 @@
 from ..queue_operation import QueueOperation
 from ..model import MessageItem
 from ..constant import MessageStatus
-from ..task_cycle.task_cycle import TaskCycle
+from ..cycle.task_cycle import TaskCycle
-from ..task_cycle.task_mounter import TaskMounter 
+from ..mounter.task_mounter import TaskMounter 
 from concurrent.futures import ThreadPoolExecutor
 from datetime import datetime
 import time

tests/listen/__init__.py

@@ -0,0 +1,3 @@
+from .listen_data import key_1
+
+__all__ = ["key_1"]

tests/listen/listen_data.py

@@ -0,0 +1,5 @@
+from queue_sqlite.mounter.listen_mounter import ListenMounter
+
+@ListenMounter.listener()
+def key_1(data):
+    print(data)

tests/tasks/example.py

@@ -1,4 +1,4 @@
-from queue_sqlite.task_cycle.task_mounter import TaskMounter
+from queue_sqlite.mounter.task_mounter import TaskMounter
 from queue_sqlite.model import MessageItem
 
 

tests/test_core.py

@@ -1,5 +1,5 @@
 from queue_sqlite.core import core
-from queue_sqlite.task_cycle import TaskMounter
+from queue_sqlite.mounter import TaskMounter
 from queue_sqlite.model import MessageItem
 
 

tests/test_listen.py

@@ -0,0 +1,19 @@
+from listen import *
+from queue_sqlite.scheduler import QueueScheduler
+import time
+
+
+class TestListen:
+    def test_listen_data(self):
+        scheduler = QueueScheduler()
+        scheduler.start_queue_scheduler()
+        scheduler.update_listen_data("key_1", "value_1")
+        time.sleep(0.001)
+        scheduler.update_listen_data("key_1", "value_2")
+        time.sleep(0.001)
+        scheduler.update_listen_data("key_1", "value_3")
+        time.sleep(0.001)
+        scheduler.update_listen_data("key_1", "value_4")
+        time.sleep(0.001)
+        scheduler.stop_queue_scheduler()
+        print(scheduler.listen_scheduler.listen_data)