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
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
187
188
189
|
from ctypes import CDLL
from typing import Any, Callable, Dict, List, Tuple
from eduvpn_common.server import (
get_locations,
get_servers,
get_transition_profiles,
get_transition_server,
)
from eduvpn_common.state import State, StateType
from eduvpn_common.types import get_ptr_string
# The attribute that callback functions get
EDUVPN_CALLBACK_PROPERTY = "_eduvpn_property_callback"
# A state transition decorator for classes
# To use this, make sure to register the class with `register_class_callbacks`
def class_state_transition(state: int, state_type: StateType) -> Callable:
"""A decorator to be internally by classes to register the event
:param state: int: The state of the transition
:param state_type: StateType: The type of transition
:meta private:
"""
def wrapper(func):
"""
:param func: The function to set the internal attribute for
"""
setattr(func, EDUVPN_CALLBACK_PROPERTY, (state, state_type))
return func
return wrapper
def convert_data(lib: CDLL, state: int, data: Any) -> None:
"""The function that converts the C structure from the Go library to a Python structure
:param lib: CDLL: The Go shared library
:param state: int: The state to convert the data for
:param data: Any: The data itself that has to be converted
:meta private:
"""
if not data:
return None
if state is State.NO_SERVER:
return get_servers(lib, data)
if state is State.OAUTH_STARTED:
return get_ptr_string(lib, data)
if state is State.ASK_LOCATION:
return get_locations(lib, data)
if state is State.ASK_PROFILE:
return get_transition_profiles(lib, data)
if state in [
State.DISCONNECTED,
State.DISCONNECTING,
State.CONNECTING,
State.CONNECTED,
]:
return get_transition_server(lib, data)
class EventHandler(object):
"""The class that neatly handles event callbacks from the internal Go FSM"""
def __init__(self, lib: CDLL):
self.handlers: Dict[Tuple[int, StateType], List[Callable]] = {}
self.lib = lib
def change_class_callbacks(self, cls: Any, add: bool = True) -> None:
"""The function that is used to change class callbacks
:param cls: Any: The class to change the callbacks for
:param add: bool: (Default value = True): Whether or not to add or remove the event. If true the event gets added
:meta private:
"""
# Loop over method names
for method_name in dir(cls):
try:
# Get the method
method = getattr(cls, method_name)
except:
# Unable to get a value, go to the next
continue
# If it has a callback defined, add it to the events
method_value = getattr(method, EDUVPN_CALLBACK_PROPERTY, None)
if method_value:
state, state_type = method_value
if add:
self.add_event(state, state_type, method)
else:
self.remove_event(state, state_type, method)
def remove_event(self, state: int, state_type: StateType, func: Callable) -> None:
"""Removes an event
:param state: int: The state to remove the event for
:param state_type: StateType: The state type to remove the event for
:param func: Callable: The function that needs to be removed from the event
:meta private:
"""
for key, values in self.handlers.copy().items():
if key == (state, state_type):
values.remove(func)
if not values:
del self.handlers[key]
else:
self.handlers[key] = values
def add_event(self, state: int, state_type: StateType, func: Callable) -> None:
"""Adds an event
:param state: int: The state to add the event for
:param state_type: StateType: The state type to add the event for
:param func: Callable: The function that needs to be added to the event
:meta private:
"""
if (state, state_type) not in self.handlers:
self.handlers[(state, state_type)] = []
self.handlers[(state, state_type)].append(func)
def on(self, state: int, state_type: StateType) -> Callable:
"""The decorator for standalone functions
:param state: int: The state of the event
:param state_type: StateType: The state type of the event
:meta private:
"""
def wrapped_f(func):
"""
:param func: The function to add the event for
"""
self.add_event(state, state_type, func)
return func
return wrapped_f
def run_state(
self, state: int, other_state: int, state_type: StateType, data: str
) -> bool:
"""The function that runs the callback for a specific event
:param state: int: The state of the event
:param other_state: int: The other state of the event
:param state_type: StateType: The state type of the event
:param data: str: The data that gets passed to the function callback when the event is ran
:meta private:
"""
if (state, state_type) not in self.handlers:
return False
for func in self.handlers[(state, state_type)]:
func(other_state, data)
return True
def run(
self, old_state: int, new_state: int, data: Any, convert: bool = True
) -> bool:
"""Run a specific event.
It converts the data and then runs the event for all state types
:param old_state: int: The previous state for running the event
:param new_state: int: The new state for running the event
:param data: Any: The data that gets passed to the event
:param convert: bool: (Default value = True): Whether or not to convert the data further
"""
# First run leave transitions, then enter
# The state is done when the wait event finishes
converted = data
if convert:
converted = convert_data(self.lib, new_state, data)
self.run_state(old_state, new_state, StateType.LEAVE, converted)
# We decide handled based on enter transitions
handled = self.run_state(new_state, old_state, StateType.ENTER, converted)
# Only run wait transitions if the enter transition is handled
if handled:
self.run_state(new_state, old_state, StateType.WAIT, converted)
return handled
|
