From 0a19c2dedcaaa177b420eac99149515d84508204 Mon Sep 17 00:00:00 2001
From: jwijenbergh <jeroenwijenbergh@protonmail.com>
Date: Mon, 26 Sep 2022 16:47:35 +0200
Subject: Python: Move from src/ to eduvpn_common/ and absolufy imports

---
 wrappers/python/eduvpn_common/event.py | 109 +++++++++++++++++++++++++++++++++
 1 file changed, 109 insertions(+)
 create mode 100644 wrappers/python/eduvpn_common/event.py

(limited to 'wrappers/python/eduvpn_common/event.py')

diff --git a/wrappers/python/eduvpn_common/event.py b/wrappers/python/eduvpn_common/event.py
new file mode 100644
index 0000000..4532bef
--- /dev/null
+++ b/wrappers/python/eduvpn_common/event.py
@@ -0,0 +1,109 @@
+from eduvpn_common import VPNStateChange, get_ptr_string
+from enum import Enum
+from typing import Callable
+from eduvpn_common.state import State, StateType
+from eduvpn_common.server import (
+    get_locations,
+    get_transition_profiles,
+    get_transition_server,
+    get_servers,
+)
+
+
+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:
+    def wrapper(func):
+        setattr(func, EDUVPN_CALLBACK_PROPERTY, (state, state_type))
+        return func
+
+    return wrapper
+
+
+def convert_data(state: State, data):
+    if not data:
+        return None
+    if state is State.NO_SERVER:
+        return get_servers(data)
+    if state is State.OAUTH_STARTED:
+        return get_ptr_string(data)
+    if state is State.ASK_LOCATION:
+        return get_locations(data)
+    if state is State.ASK_PROFILE:
+        return get_transition_profiles(data)
+    if state in [
+        State.DISCONNECTED,
+        State.DISCONNECTING,
+        State.CONNECTING,
+        State.CONNECTED,
+    ]:
+        return get_transition_server(data)
+
+
+class EventHandler(object):
+    def __init__(self):
+        self.handlers = {}
+
+    def change_class_callbacks(self, cls, add=True) -> None:
+        # 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):
+        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):
+        if (state, state_type) not in self.handlers:
+            self.handlers[(state, state_type)] = []
+        self.handlers[(state, state_type)].append(func)
+
+    # A decorator for standalone functions
+    def on(self, state: int, state_type: StateType) -> Callable:
+        def wrapped_f(func):
+            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
+    ) -> None:
+        if (state, state_type) not in self.handlers:
+            return
+        for func in self.handlers[(state, state_type)]:
+            func(other_state, data)
+
+    def run(
+        self, old_state: int, new_state: int, data: str, convert: bool = True
+    ) -> None:
+        # First run leave transitions, then enter
+        # The state is done when the wait event finishes
+        converted = data
+        if convert:
+            converted = convert_data(new_state, data)
+        self.run_state(old_state, new_state, StateType.Leave, converted)
+        self.run_state(new_state, old_state, StateType.Enter, converted)
+        self.run_state(new_state, old_state, StateType.Wait, converted)
-- 
cgit v1.2.3