Skip to content

Dispatcher

gradysim.protocol.plugin.dispatcher

This module contains a function that creates a dispatcher which wraps a protocol instance and it's methods. Implements a call chain for each of the protocol interface's methods.

Use this module through the create_dispatcher][gradysim.protocol.plugin.dispatcher.create_dispatcher] method, never instantiate the ProtocolWrapper directly.

This module is useful if you want to implement an plugin on or some other functionality that relies on overriding the protocol's methods. Protocols using this plugin will still be compatible with all execution modes as only the protocol itself is tampered with and not any of the layers that allow it to run on different environments.

Beware that this module uses monkey patching and may result in broken protocols if someone else tries to tamper with the protocol's methods.

DispatchReturn

Bases: Enum

Return value for the dispatched funcions. INTERRUPT should be returned if your handler completely handled the call and it shouldn't be passed forward. CONTINUE should be called if the call should continue down the call chain.

Source code in gradysim/protocol/plugin/dispatcher.py 
24
25
26
27
28
29
30
31
class DispatchReturn(Enum):
    """
    Return value for the dispatched funcions. INTERRUPT should be returned if your handler completely handled
    the call and it shouldn't be passed forward. CONTINUE should be called if the call should continue down
    the call chain.
    """
    INTERRUPT = 1
    CONTINUE = 2

ProtocolWrapper

Do not use this class directly, instead use create_dispatcher.

Wraps the protocol's calls into a call chain. Instead of going directly to the protocol's methods calls to the protocol interface will be passed down a chain of registered handlers. The protocol's own method is at the end of the chain. Methods should return a value in the DispatchReturn Enum, INTERRUPT do interrupt the chain and CONTINUE if the message should be passed through.

Source code in gradysim/protocol/plugin/dispatcher.py 
 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
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
class ProtocolWrapper:
    """
    Do not use this class directly, instead use
    [create_dispatcher][gradysim.protocol.plugin.dispatcher.create_dispatcher].

    Wraps the protocol's calls into a call chain. Instead of going directly to the protocol's methods calls to the
    protocol interface will be passed down a chain of registered handlers. The protocol's own method is at the end
    of the chain. Methods should return a value in the DispatchReturn Enum, INTERRUPT do interrupt the chain and
    CONTINUE if the message should be passed through.
    """
    _protocol: IProtocol

    _handle_initialize_chain: List[Callable[[IProtocol, int], None]]
    _handle_timer_chain: List[Callable[[IProtocol, str], DispatchReturn]]
    _handle_telemetry_chain: List[Callable[[IProtocol, Telemetry], DispatchReturn]]
    _handle_packet_chain: List[Callable[[IProtocol, str], DispatchReturn]]
    _finish_queue_chain: List[Callable[[], None]]

    def __init__(self, protocol: IProtocol):
        """
        Instantiates a protocol wrapper. Should not be instantiated directly, create a dispatcher using the
        [create_dispatcher][gradysim.protocol.plugin.dispatcher.create_dispatcher] method.

        **Do not instantiate this class directly**

        Args:
            protocol: Protocol whose calls will be wrapped
        """
        self._protocol = protocol

        self._handle_initialize_chain = []
        self._handle_timer_chain = []
        self._handle_telemetry_chain = []
        self._handle_packet_chain = []
        self._finish_queue_chain = []

        _wrap_functionality(protocol, 'initialize', self._handle_initialize_chain)
        _wrap_functionality(protocol, 'handle_timer', self._handle_timer_chain)
        _wrap_functionality(protocol, 'handle_telemetry', self._handle_telemetry_chain)
        _wrap_functionality(protocol, 'handle_packet', self._handle_packet_chain)
        _wrap_functionality(protocol, 'finish', self._finish_queue_chain)

    def register_initialize(self, handler: Callable[[IProtocol, int], None]) -> None:
        """
        Registers a handler for the [initialize][gradysim.protocol.interface.IProtocol.initialize] method. Handlers should
        have the same signature as the initialize method. DispatcherReturn is not supported for this method, the call
        chain is always followed.

        Args:
            handler: Handler being registered
        """
        self._handle_initialize_chain.insert(0, handler)

    def register_handle_timer(self, handler: Callable[[IProtocol, str], DispatchReturn]) -> None:
        """
        Registers a handler for the [handle_timer][gradysim.protocol.interface.IProtocol.handle_timer] method. Handlers should
        have the same signature as the handle_timer method but return a value in the DispatcherReturn enum.

        Args:
            handler: Handler being registered
        """
        self._handle_timer_chain.insert(0, handler)

    def register_handle_telemetry(self, handler: Callable[[IProtocol, Telemetry], DispatchReturn]) -> None:
        """
        Registers a handler for the [handle_telemetry][gradysim.protocol.interface.IProtocol.handle_telemetry] method. Handlers
        should have the same signature as the handle_telemetry method but return a value in the DispatcherReturn enum.

        Args:
            handler: Handler being registered
        """
        self._handle_telemetry_chain.insert(0, handler)

    def register_handle_packet(self, handler: Callable[[IProtocol, str], DispatchReturn]) -> None:
        """
        Registers a handler for the [handle_packet][gradysim.protocol.interface.IProtocol.handle_packet] method. Handlers should
        have the same signature as the handle_packet method but return a value in the DispatcherReturn enum.

        Args:
            handler: Handler being registered
        """
        self._handle_packet_chain.insert(0, handler)

    def register_finish(self, handler: Callable[[IProtocol], None]) -> None:
        """
        Registers a handler for the [handle_timer][gradysim.protocol.interface.IProtocol.finish] method. Handlers should
        have the same signature as the finish. This method doesn't support returning DispatchReturn values, the
        call chain is always followed.

        Not following this chain, since it is only called once, could result in improper finalization of protocols and
        other handlers.

        Args:
            handler: Handler being registered
        """
        self._finish_queue_chain.insert(0, handler)

    def unregister_initialize(self, handler: Callable[[IProtocol, int], None]) -> None:
        """
        Unregisters a handle_initialize handler. Raises ValueError if handler was not registered

        Args:
            handler: Handler instance being unregistered
        """
        self._handle_initialize_chain.remove(handler)

    def unregister_handle_timer(self, handler: Callable[[IProtocol, str], DispatchReturn]) -> None:
        """
        Unregisters a handle_timer handler. Raises ValueError if handler was not registered

        Args:
            handler: Handler instance being unregistered
        """
        self._handle_timer_chain.remove(handler)

    def unregister_handle_telemetry(self, handler: Callable[[IProtocol, Telemetry], DispatchReturn]) -> None:
        """
        Unregisters a handle_timer handler. Raises ValueError if handler was not registered

        Args:
            handler: Handler instance being unregistered
        """
        self._handle_telemetry_chain.remove(handler)

    def unregister_handle_packet(self, handler: Callable[[IProtocol, str], DispatchReturn]) -> None:
        """
        Unregisters a handle_timer handler. Raises ValueError if handler was not registered

        Args:
            handler: Handler instance being unregistered
        """
        self._handle_packet_chain.remove(handler)

    def unregister_finish(self, handler: Callable[[IProtocol], None]) -> None:
        """
        Unregisters a handle_timer handler. Raises ValueError if handler was not registered

        Args:
            handler: Handler instance being unregistered
        """
        self._finish_queue_chain.remove(handler)

__init__(protocol)

Instantiates a protocol wrapper. Should not be instantiated directly, create a dispatcher using the create_dispatcher method.

Do not instantiate this class directly

Parameters:

Name Type Description Default
protocol IProtocol

Protocol whose calls will be wrapped

 required
Source code in gradysim/protocol/plugin/dispatcher.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def __init__(self, protocol: IProtocol):
    """
    Instantiates a protocol wrapper. Should not be instantiated directly, create a dispatcher using the
    [create_dispatcher][gradysim.protocol.plugin.dispatcher.create_dispatcher] method.

    **Do not instantiate this class directly**

    Args:
        protocol: Protocol whose calls will be wrapped
    """
    self._protocol = protocol

    self._handle_initialize_chain = []
    self._handle_timer_chain = []
    self._handle_telemetry_chain = []
    self._handle_packet_chain = []
    self._finish_queue_chain = []

    _wrap_functionality(protocol, 'initialize', self._handle_initialize_chain)
    _wrap_functionality(protocol, 'handle_timer', self._handle_timer_chain)
    _wrap_functionality(protocol, 'handle_telemetry', self._handle_telemetry_chain)
    _wrap_functionality(protocol, 'handle_packet', self._handle_packet_chain)
    _wrap_functionality(protocol, 'finish', self._finish_queue_chain)

register_finish(handler)

Registers a handler for the handle_timer method. Handlers should have the same signature as the finish. This method doesn't support returning DispatchReturn values, the call chain is always followed.

Not following this chain, since it is only called once, could result in improper finalization of protocols and other handlers.

Parameters:

Name Type Description Default
handler Callable[[IProtocol], None]

Handler being registered

 required
Source code in gradysim/protocol/plugin/dispatcher.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
def register_finish(self, handler: Callable[[IProtocol], None]) -> None:
    """
    Registers a handler for the [handle_timer][gradysim.protocol.interface.IProtocol.finish] method. Handlers should
    have the same signature as the finish. This method doesn't support returning DispatchReturn values, the
    call chain is always followed.

    Not following this chain, since it is only called once, could result in improper finalization of protocols and
    other handlers.

    Args:
        handler: Handler being registered
    """
    self._finish_queue_chain.insert(0, handler)

register_handle_packet(handler)

Registers a handler for the handle_packet method. Handlers should have the same signature as the handle_packet method but return a value in the DispatcherReturn enum.

Parameters:

Name Type Description Default
handler Callable[[IProtocol, str], DispatchReturn]

Handler being registered

 required
Source code in gradysim/protocol/plugin/dispatcher.py 
119
120
121
122
123
124
125
126
127
def register_handle_packet(self, handler: Callable[[IProtocol, str], DispatchReturn]) -> None:
    """
    Registers a handler for the [handle_packet][gradysim.protocol.interface.IProtocol.handle_packet] method. Handlers should
    have the same signature as the handle_packet method but return a value in the DispatcherReturn enum.

    Args:
        handler: Handler being registered
    """
    self._handle_packet_chain.insert(0, handler)

register_handle_telemetry(handler)

Registers a handler for the handle_telemetry method. Handlers should have the same signature as the handle_telemetry method but return a value in the DispatcherReturn enum.

Parameters:

Name Type Description Default
handler Callable[[IProtocol, Telemetry], DispatchReturn]

Handler being registered

 required
Source code in gradysim/protocol/plugin/dispatcher.py 
109
110
111
112
113
114
115
116
117
def register_handle_telemetry(self, handler: Callable[[IProtocol, Telemetry], DispatchReturn]) -> None:
    """
    Registers a handler for the [handle_telemetry][gradysim.protocol.interface.IProtocol.handle_telemetry] method. Handlers
    should have the same signature as the handle_telemetry method but return a value in the DispatcherReturn enum.

    Args:
        handler: Handler being registered
    """
    self._handle_telemetry_chain.insert(0, handler)

register_handle_timer(handler)

Registers a handler for the handle_timer method. Handlers should have the same signature as the handle_timer method but return a value in the DispatcherReturn enum.

Parameters:

Name Type Description Default
handler Callable[[IProtocol, str], DispatchReturn]

Handler being registered

 required
Source code in gradysim/protocol/plugin/dispatcher.py 
 99
100
101
102
103
104
105
106
107
def register_handle_timer(self, handler: Callable[[IProtocol, str], DispatchReturn]) -> None:
    """
    Registers a handler for the [handle_timer][gradysim.protocol.interface.IProtocol.handle_timer] method. Handlers should
    have the same signature as the handle_timer method but return a value in the DispatcherReturn enum.

    Args:
        handler: Handler being registered
    """
    self._handle_timer_chain.insert(0, handler)

register_initialize(handler)

Registers a handler for the initialize method. Handlers should have the same signature as the initialize method. DispatcherReturn is not supported for this method, the call chain is always followed.

Parameters:

Name Type Description Default
handler Callable[[IProtocol, int], None]

Handler being registered

 required
Source code in gradysim/protocol/plugin/dispatcher.py 
88
89
90
91
92
93
94
95
96
97
def register_initialize(self, handler: Callable[[IProtocol, int], None]) -> None:
    """
    Registers a handler for the [initialize][gradysim.protocol.interface.IProtocol.initialize] method. Handlers should
    have the same signature as the initialize method. DispatcherReturn is not supported for this method, the call
    chain is always followed.

    Args:
        handler: Handler being registered
    """
    self._handle_initialize_chain.insert(0, handler)

unregister_finish(handler)

Unregisters a handle_timer handler. Raises ValueError if handler was not registered

Parameters:

Name Type Description Default
handler Callable[[IProtocol], None]

Handler instance being unregistered

 required
Source code in gradysim/protocol/plugin/dispatcher.py 
179
180
181
182
183
184
185
186
def unregister_finish(self, handler: Callable[[IProtocol], None]) -> None:
    """
    Unregisters a handle_timer handler. Raises ValueError if handler was not registered

    Args:
        handler: Handler instance being unregistered
    """
    self._finish_queue_chain.remove(handler)

unregister_handle_packet(handler)

Unregisters a handle_timer handler. Raises ValueError if handler was not registered

Parameters:

Name Type Description Default
handler Callable[[IProtocol, str], DispatchReturn]

Handler instance being unregistered

 required
Source code in gradysim/protocol/plugin/dispatcher.py 
170
171
172
173
174
175
176
177
def unregister_handle_packet(self, handler: Callable[[IProtocol, str], DispatchReturn]) -> None:
    """
    Unregisters a handle_timer handler. Raises ValueError if handler was not registered

    Args:
        handler: Handler instance being unregistered
    """
    self._handle_packet_chain.remove(handler)

unregister_handle_telemetry(handler)

Unregisters a handle_timer handler. Raises ValueError if handler was not registered

Parameters:

Name Type Description Default
handler Callable[[IProtocol, Telemetry], DispatchReturn]

Handler instance being unregistered

 required
Source code in gradysim/protocol/plugin/dispatcher.py 
161
162
163
164
165
166
167
168
def unregister_handle_telemetry(self, handler: Callable[[IProtocol, Telemetry], DispatchReturn]) -> None:
    """
    Unregisters a handle_timer handler. Raises ValueError if handler was not registered

    Args:
        handler: Handler instance being unregistered
    """
    self._handle_telemetry_chain.remove(handler)

unregister_handle_timer(handler)

Unregisters a handle_timer handler. Raises ValueError if handler was not registered

Parameters:

Name Type Description Default
handler Callable[[IProtocol, str], DispatchReturn]

Handler instance being unregistered

 required
Source code in gradysim/protocol/plugin/dispatcher.py 
152
153
154
155
156
157
158
159
def unregister_handle_timer(self, handler: Callable[[IProtocol, str], DispatchReturn]) -> None:
    """
    Unregisters a handle_timer handler. Raises ValueError if handler was not registered

    Args:
        handler: Handler instance being unregistered
    """
    self._handle_timer_chain.remove(handler)

unregister_initialize(handler)

Unregisters a handle_initialize handler. Raises ValueError if handler was not registered

Parameters:

Name Type Description Default
handler Callable[[IProtocol, int], None]

Handler instance being unregistered

 required
Source code in gradysim/protocol/plugin/dispatcher.py 
143
144
145
146
147
148
149
150
def unregister_initialize(self, handler: Callable[[IProtocol, int], None]) -> None:
    """
    Unregisters a handle_initialize handler. Raises ValueError if handler was not registered

    Args:
        handler: Handler instance being unregistered
    """
    self._handle_initialize_chain.remove(handler)

create_dispatcher(protocol)

Creates a dispatcher which wraps a protocol instance and it's methods. Implements a call chain for each of the protocol interface's methods. The class returned from this function can be used to add functions to the call chain of those wrapped methods. The original method implementation is not lost.

Is a protocol that was already wrapped is passed as an argument, return the wrapper for that protocol.

Beware that this module uses monkey patching and may result in broken protocols if someone else tries to tamper with the protocol's methods.

If you want to implement an plugin or some other behaviour that requires overriding protocol's methods you should use this function

Parameters:

Name Type Description Default
protocol IProtocol

Protocol being wrapped

 required

Returns:

Type Description
ProtocolWrapper

ProtocolWrapper instance that allows methods to be added to the call chain
Source code in gradysim/protocol/plugin/dispatcher.py 
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
def create_dispatcher(protocol: IProtocol) -> ProtocolWrapper:
    """
    Creates a dispatcher which wraps a protocol instance and it's methods. Implements a call chain for each of the
    protocol interface's methods. The class returned from this function can be used to add functions to the call chain
    of those wrapped methods. The original method implementation is not lost.

    Is a protocol that was already wrapped is passed as an argument, return the wrapper for that protocol.

    Beware that this module uses monkey patching and may result in broken protocols if someone else tries to tamper with
    the protocol's methods.

    If you want to implement an plugin or some other behaviour that requires overriding protocol's
    methods you should use this function

    Args:
        protocol: Protocol being wrapped

    Returns:
        ProtocolWrapper instance that allows methods to be added to the call chain
    """
    global _protocol_wrappers
    if protocol not in _protocol_wrappers:
        _protocol_wrappers[protocol] = ProtocolWrapper(protocol)

    return _protocol_wrappers[protocol]