|
1
|
""" |
|
2
|
GraphNotifier — real-time graph change notifications via Redis pub/sub. |
|
3
|
|
|
4
|
Agents can publish change events when they mutate graph nodes or edges and |
|
5
|
subscribe to receive those events, enabling reactive coordination in a swarm. |
|
6
|
|
|
7
|
Usage: |
|
8
|
notifier = GraphNotifier("redis://localhost:6379") |
|
9
|
|
|
10
|
# Publisher side |
|
11
|
notifier.publish(EventType.NODE_CREATED, {"label": "Function", "name": "my_fn"}) |
|
12
|
|
|
13
|
# Subscriber side (blocking — run in a thread) |
|
14
|
def handler(event_type, data): |
|
15
|
print(f"Event: {event_type}, data: {data}") |
|
16
|
|
|
17
|
notifier.subscribe([EventType.NODE_CREATED, EventType.EDGE_CREATED], handler) |
|
18
|
""" |
|
19
|
|
|
20
|
from __future__ import annotations |
|
21
|
|
|
22
|
import json |
|
23
|
import logging |
|
24
|
import threading |
|
25
|
from enum import Enum |
|
26
|
from typing import Any, Callable |
|
27
|
|
|
28
|
logger = logging.getLogger(__name__) |
|
29
|
|
|
30
|
_CHANNEL_PREFIX = "navegador:events" |
|
31
|
|
|
32
|
|
|
33
|
class EventType(str, Enum): |
|
34
|
"""Graph change event types.""" |
|
35
|
|
|
36
|
NODE_CREATED = "node_created" |
|
37
|
NODE_UPDATED = "node_updated" |
|
38
|
NODE_DELETED = "node_deleted" |
|
39
|
EDGE_CREATED = "edge_created" |
|
40
|
EDGE_UPDATED = "edge_updated" |
|
41
|
EDGE_DELETED = "edge_deleted" |
|
42
|
GRAPH_CLEARED = "graph_cleared" |
|
43
|
SNAPSHOT_PUSHED = "snapshot_pushed" |
|
44
|
|
|
45
|
|
|
46
|
def _channel_name(event_type: EventType | str) -> str: |
|
47
|
val = event_type.value if isinstance(event_type, EventType) else str(event_type) |
|
48
|
return f"{_CHANNEL_PREFIX}:{val}" |
|
49
|
|
|
50
|
|
|
51
|
class GraphNotifier: |
|
52
|
""" |
|
53
|
Publish and subscribe to graph change events over Redis pub/sub. |
|
54
|
|
|
55
|
Parameters |
|
56
|
---------- |
|
57
|
redis_url: |
|
58
|
URL of the Redis server. |
|
59
|
redis_client: |
|
60
|
Optional pre-built Redis client (for testing / DI). |
|
61
|
""" |
|
62
|
|
|
63
|
def __init__(self, redis_url: str, *, redis_client: Any = None) -> None: |
|
64
|
self.redis_url = redis_url |
|
65
|
self._redis = redis_client or self._connect_redis(redis_url) |
|
66
|
self._subscriptions: list[threading.Thread] = [] |
|
67
|
|
|
68
|
# ── Internal ────────────────────────────────────────────────────────────── |
|
69
|
|
|
70
|
@staticmethod |
|
71
|
def _connect_redis(url: str) -> Any: |
|
72
|
try: |
|
73
|
import redis # type: ignore[import] |
|
74
|
except ImportError as exc: |
|
75
|
raise ImportError("Install redis: pip install redis") from exc |
|
76
|
return redis.from_url(url) |
|
77
|
|
|
78
|
# ── Public API ──────────────────────────────────────────────────────────── |
|
79
|
|
|
80
|
def publish(self, event_type: EventType | str, data: dict[str, Any]) -> int: |
|
81
|
""" |
|
82
|
Publish a change event to all subscribers. |
|
83
|
|
|
84
|
Parameters |
|
85
|
---------- |
|
86
|
event_type: |
|
87
|
One of the ``EventType`` enum values (or a raw string). |
|
88
|
data: |
|
89
|
Arbitrary JSON-serialisable payload describing the change. |
|
90
|
|
|
91
|
Returns |
|
92
|
------- |
|
93
|
int |
|
94
|
Number of clients that received the message. |
|
95
|
""" |
|
96
|
channel = _channel_name(event_type) |
|
97
|
payload = json.dumps( |
|
98
|
{ |
|
99
|
"event_type": event_type.value if isinstance(event_type, EventType) else event_type, |
|
100
|
"data": data, |
|
101
|
} |
|
102
|
) |
|
103
|
result = self._redis.publish(channel, payload) |
|
104
|
logger.debug("Published %s to channel %s (%d receivers)", event_type, channel, result) |
|
105
|
return result |
|
106
|
|
|
107
|
def subscribe( |
|
108
|
self, |
|
109
|
event_types: list[EventType | str], |
|
110
|
callback: Callable[[str, dict[str, Any]], None], |
|
111
|
*, |
|
112
|
run_in_thread: bool = False, |
|
113
|
) -> threading.Thread | None: |
|
114
|
""" |
|
115
|
Subscribe to one or more event types and invoke *callback* for each. |
|
116
|
|
|
117
|
Parameters |
|
118
|
---------- |
|
119
|
event_types: |
|
120
|
List of ``EventType`` values to listen for. |
|
121
|
callback: |
|
122
|
Callable receiving ``(event_type: str, data: dict)``. |
|
123
|
run_in_thread: |
|
124
|
If ``True``, run the blocking listen loop in a daemon thread and |
|
125
|
return that thread. If ``False`` (default), block the calling |
|
126
|
thread. |
|
127
|
|
|
128
|
Returns |
|
129
|
------- |
|
130
|
threading.Thread | None |
|
131
|
The daemon thread if ``run_in_thread=True``, else ``None``. |
|
132
|
""" |
|
133
|
channels = [_channel_name(et) for et in event_types] |
|
134
|
pubsub = self._redis.pubsub() |
|
135
|
pubsub.subscribe(*channels) |
|
136
|
logger.info("Subscribed to channels: %s", channels) |
|
137
|
|
|
138
|
def _listen() -> None: |
|
139
|
for message in pubsub.listen(): |
|
140
|
if message["type"] != "message": |
|
141
|
continue |
|
142
|
try: |
|
143
|
payload = json.loads(message["data"]) |
|
144
|
et = payload.get("event_type", "") |
|
145
|
d = payload.get("data", {}) |
|
146
|
callback(et, d) |
|
147
|
except (json.JSONDecodeError, TypeError, KeyError) as exc: |
|
148
|
logger.warning("Failed to decode pubsub message: %s", exc) |
|
149
|
|
|
150
|
if run_in_thread: |
|
151
|
t = threading.Thread(target=_listen, daemon=True) |
|
152
|
t.start() |
|
153
|
self._subscriptions.append(t) |
|
154
|
return t |
|
155
|
|
|
156
|
_listen() |
|
157
|
return None |
|
158
|
|
|
159
|
def close(self) -> None: |
|
160
|
"""Close the Redis connection.""" |
|
161
|
try: |
|
162
|
self._redis.close() |
|
163
|
except Exception: |
|
164
|
pass |
|
165
|
|