Eventful/src/eventful/router.py at dependabot/github_actions/actions/setup-python-7 · Forebase/Eventful · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
"""
Topic-based event routing with wildcard support.
"""

from __future__ import annotations

from eventful.event import Event
from eventful.listener import Listener


class Router:
    """
    Router for matching events to listeners based on topic patterns.

    Supports hierarchical topics with * (single-level) and ** (multi-level) wildcards.
    """

    def __init__(self) -> None:
        self._listeners: dict[str, list[Listener]] = {}
        self._wildcard_listeners: list[tuple[str, Listener]] = []

    def add_listener(self, topic: str, listener: Listener) -> None:
        """
        Add a listener for a topic pattern.

        Parameters
        ----------
        topic : str
            Topic pattern (may contain wildcards)
        listener : Listener
            Listener to add
        """
        if '*' in topic or '**' in topic:
            self._wildcard_listeners.append((topic, listener))
        else:
            if topic not in self._listeners:
                self._listeners[topic] = []
            self._listeners[topic].append(listener)

    def remove_listener(self, topic: str, listener: Listener) -> None:
        """
        Remove a listener from a topic pattern.

        Parameters
        ----------
        topic : str
            Topic pattern
        listener : Listener
            Listener to remove
        """
        if '*' in topic or '**' in topic:
            self._wildcard_listeners = [(t, l) for t, l in self._wildcard_listeners
                                        if not (t == topic and l == listener)]
        else:
            if topic in self._listeners:
                self._listeners[topic] = [l for l in self._listeners[topic] if l != listener]
                if not self._listeners[topic]:
                    del self._listeners[topic]

    def get_listeners(self, event: Event) -> list[Listener]:
        """
        Get all listeners that match the given event.

        Parameters
        ----------
        event : Event
            Event to match

        Returns
        -------
        List[Listener]
            List of matching listeners
        """
        listeners = []
        event_type = event.type

        # Exact matches
        if event_type in self._listeners:
            listeners.extend(self._listeners[event_type])

        # Wildcard matches
        for pattern, listener_obj in self._wildcard_listeners:
            if self._matches_pattern(event_type, pattern):
                listeners.append(listener_obj)

        # Filter by listener-specific criteria
        return [l for l in listeners if l.matches(event)]

    def _matches_pattern(self, event_type: str, pattern: str) -> bool:
        """
        Check if event type matches a wildcard pattern.

        Parameters
        ----------
        event_type : str
            Event type to match
        pattern : str
            Pattern with wildcards

        Returns
        -------
        bool
            True if pattern matches
        """
        if pattern == event_type:
            return True

        pattern_parts = pattern.split('.')
        event_parts = event_type.split('.')

        i = j = 0
        while i < len(pattern_parts) and j < len(event_parts):
            if pattern_parts[i] == '**':
                # Multi-level wildcard - match remaining event parts
                return True
            elif pattern_parts[i] == '*':
                # Single-level wildcard - match exactly one part
                i += 1
                j += 1
            elif pattern_parts[i] == event_parts[j]:
                i += 1
                j += 1
            else:
                return False

        # Both should be exhausted for exact match
        return i == len(pattern_parts) and j == len(event_parts)