✨ Final touches

patterns/network_event_pattern.py

@@ -1,10 +1,13 @@
 import sys
 import socket
+import threading
+import tempfile
+from time import time
 
 from typing import Any, Dict, List
 
 from meow_base.functionality.validation import valid_string, valid_dict
-from meow_base.core.vars import VALID_RECIPE_NAME_CHARS, VALID_VARIABLE_NAME_CHARS
+from meow_base.core.vars import VALID_RECIPE_NAME_CHARS, VALID_VARIABLE_NAME_CHARS, DEBUG_INFO
 from meow_base.core.base_recipe import BaseRecipe
 from meow_base.core.base_monitor import BaseMonitor
 from meow_base.core.base_pattern import BasePattern
@@ -12,19 +15,17 @@ from meow_base.functionality.meow import create_event
 from meow_base.functionality.debug import setup_debugging, print_debug
 from meow_base.core.meow import EVENT_KEYS
 
-# watchdog events
+# network events
 EVENT_TYPE_NETWORK = "network"
-NETWORK_BASE = "monitor_base"
-NETWORK_HASH = "file_hash"
+TRIGGERING_PORT = "triggering port"
 
 NETWORK_EVENT_KEYS = {
-    NETWORK_BASE: str,
-    NETWORK_HASH: str,
+    TRIGGERING_PORT: int,
     **EVENT_KEYS
 }
 
-def create_network_event(temp_path:str, rule:Any, base:str, time:float,
-            hash:str, extras:Dict[Any,Any]={})->Dict[Any,Any]:
+def create_network_event(temp_path:str, rule:Any, time:float,
+                         port: int, extras:Dict[Any,Any]={})->Dict[Any,Any]:
     """Function to create a MEOW event dictionary."""
     return create_event(
         EVENT_TYPE_NETWORK,
@@ -32,39 +33,11 @@ def create_network_event(temp_path:str, rule:Any, base:str, time:float,
         rule,
         time,
         extras={
-            **extras,
-            **{
-                NETWORK_HASH: hash,
-                NETWORK_BASE: base
-            }
+            TRIGGERING_PORT: port,
+            **extras
         }
     )
 
-class Connector():
-    def __init__(self, host: int, port: int) -> None:
-        self.host = host
-        self.port = port
-        self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-
-    # Only used for testing
-    def send(self, message: bytes) -> None:
-        self.socket.connect((self.host, self.port))
-        self.socket.sendall(message)
-        self.socket.close()
-
-    def receive(self, path:str, buff_size:int = 2048):
-        self.socket.bind((self.host, self.port))
-        self.socket.listen()
-        conn, _ = self.socket.accept()
-
-        with conn:
-            with open(path, "wb") as file_pointer:
-                while True:
-                    data = conn.recv(buff_size)
-                    if not data:
-                        break
-                    file_pointer.write(data)
-
 class NetworkEventPattern(BasePattern):
     # The port to monitor
     triggering_port:int
@@ -109,9 +82,6 @@ class NetworkMonitor(BaseMonitor):
             name:str="", print:Any=sys.stdout, logging:int=0) -> None:
         super().__init__(patterns, recipes, name=name)
         self._print_target, self.debug_level = setup_debugging(print, logging)
-        self.ports = set(
-            pattern.triggering_port for pattern in patterns.values()
-        )
         if autostart:
             self.start()
 
@@ -120,12 +90,48 @@ class NetworkMonitor(BaseMonitor):
         implemented by any child process. Depending on the nature of the
         monitor, this may wish to directly call apply_retroactive_rules before
         starting."""
-        pass
+        self.ports = set(
+            pattern.triggering_port for pattern in self._patterns.values()
+        )
+        self.listeners = [Listener("127.0.0.1",i,2048,self) for i in self.ports]
+
+        for listener in self.listeners:
+            listener.start()
+
+    def match(self, event)->None:
+        """Function to determine if a given event matches the current rules."""
+
+        self._rules_lock.acquire()
+        try:
+            for rule in self._rules.values():
+                # Match event port against rule ports
+                hit = event["triggering port"]
+
+                # If matched, the create a watchdog event
+                if hit:
+                    meow_event = create_network_event(
+                        event["tmp file"],
+                        rule,
+                        event["time stamp"],
+                        event["triggering port"]
+                    )
+                    print_debug(self._print_target, self.debug_level,
+                        f"Event at {event['triggering port']} hit rule {rule.name}",
+                        DEBUG_INFO)
+                    # Send the event to the runner
+                    self.send_event_to_runner(meow_event)
+
+        except Exception as e:
+            self._rules_lock.release()
+            raise e
+
+        self._rules_lock.release()
 
     def stop(self)->None:
         """Function to stop the monitor as an ongoing process/thread. Must be
         implemented by any child process"""
-        pass
+        for listener in self.listeners:
+            listener.stop()
 
     def _is_valid_recipes(self, recipes:Dict[str,BaseRecipe])->None:
         """Validation check for 'recipes' variable from main constructor. Is
@@ -137,3 +143,59 @@ class NetworkMonitor(BaseMonitor):
 
     def _get_valid_recipe_types(self)->List[type]:
         return [BaseRecipe]
+
+class Listener():
+    def __init__(self, host: int, port: int, buff_size: int,
+                 monitor:NetworkMonitor) -> None:
+        self.host = host
+        self.port = port
+        self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        self.socket.settimeout(0.5)
+        self._stopped = False
+        self.buff_size = buff_size
+        self.monitor = monitor
+
+    def start(self):
+        self._handle_thread = threading.Thread(
+            target=self.main_loop
+        )
+        self._handle_thread.start()
+
+    def main_loop(self):
+        self.socket.bind((self.host, self.port))
+        self.socket.listen(1)
+        while not self._stopped:
+            try:
+                conn, _ = self.socket.accept()
+            except socket.timeout:
+                pass
+            except:
+                raise
+            else:
+                threading.Thread(
+                    target=self.handle_event,
+                    args=(conn,time(),)
+                ).start()
+
+
+    def handle_event(self, conn, time_stamp):
+        with conn:
+            with tempfile.NamedTemporaryFile("wb", delete=False) as tmp:
+                while True:
+                    data = conn.recv(self.buff_size)
+                    if not data:
+                        break
+                    tmp.write(data)
+
+                tmp_name = tmp.name
+
+        event = {
+            "triggering port": self.port,
+            "tmp file": tmp_name,
+            "time stamp": time_stamp
+        }
+        self.monitor.match(event)
+
+    def stop(self):
+        self._stopped = True