asyncio in proxy manager

backend/proxy/__init__.py

@@ -1,5 +1,4 @@
-import subprocess, re, os
-from threading import Lock
+import subprocess, re, os, asyncio
 
 #c++ -o proxy proxy.cpp
 
@@ -25,11 +24,11 @@ class Filter:
             yield case_sensitive + "S" + self.regex.hex() if self.is_blacklist else case_sensitive + "s"+ self.regex.hex()
 
 class Proxy:
-    def __init__(self, internal_port, public_port, callback_blocked_update=None, filters=None, public_host="0.0.0.0", internal_host="127.0.0.1"):
+    def __init__(self, internal_port=0, public_port=0, callback_blocked_update=None, filters=None, public_host="0.0.0.0", internal_host="127.0.0.1"):
         self.filter_map = {}
-        self.filter_map_lock = Lock()
-        self.update_config_lock = Lock()
-        self.status_change = Lock()
+        self.filter_map_lock = asyncio.Lock()
+        self.update_config_lock = asyncio.Lock()
+        self.status_change = asyncio.Lock()
         self.public_host = public_host
         self.public_port = public_port
         self.internal_host = internal_host
@@ -38,75 +37,72 @@ class Proxy:
         self.process = None
         self.callback_blocked_update = callback_blocked_update
     
-    def start(self, in_pause=False):
-        self.status_change.acquire()
+    async def start(self, in_pause=False):
+        await self.status_change.acquire()
         if not self.isactive():
             try:
                 self.filter_map = self.compile_filters()
                 filters_codes = list(self.filter_map.keys()) if not in_pause else []
                 proxy_binary_path = os.path.join(os.path.dirname(os.path.abspath(__file__)),"./proxy")
 
-                self.process = subprocess.Popen(
-                    [ proxy_binary_path, str(self.public_host), str(self.public_port), str(self.internal_host), str(self.internal_port)],
-                    stdout=subprocess.PIPE, stdin=subprocess.PIPE, universal_newlines=True
+                self.process = await asyncio.create_subprocess_exec(
+                    proxy_binary_path, str(self.public_host), str(self.public_port), str(self.internal_host), str(self.internal_port),
+                    stdout=asyncio.subprocess.PIPE, stdin=asyncio.subprocess.PIPE
                 )
-                self.update_config(filters_codes)
+                await self.update_config(filters_codes)
             finally:
                 self.status_change.release()
-
-            for stdout_line in iter(self.process.stdout.readline, ""):
-                if stdout_line.startswith("BLOCKED"):
-                    regex_id = stdout_line.split()[1]
-                    with self.filter_map_lock:
-                        self.filter_map[regex_id].blocked+=1
-                        if self.callback_blocked_update: self.callback_blocked_update(self.filter_map[regex_id])
-            self.process.stdout.close()
-            return self.process.wait()
+            try:
+                while True:
+                    buff = await self.process.stdout.readuntil()
+                    stdout_line = buff.decode()
+                    if stdout_line.startswith("BLOCKED"):
+                        regex_id = stdout_line.split()[1]
+                        async with self.filter_map_lock:
+                            self.filter_map[regex_id].blocked+=1
+                            if self.callback_blocked_update: await self.callback_blocked_update(self.filter_map[regex_id])
+            except Exception:
+                return await self.process.wait()
         else:
             self.status_change.release()
                  
 
-    def stop(self):
-        with self.status_change:
+    async def stop(self):
+        async with self.status_change:
             if self.isactive():
-                self.process.terminate()
-                try:
-                    self.process.wait(timeout=3)
-                except Exception:
-                    self.process.kill()
-                    return False
-                finally:
-                    self.process = None
+                self.process.kill()
+                self.process = None
+                return False
             return True
 
-    def restart(self, in_pause=False):
-        status = self.stop()
-        self.start(in_pause=in_pause)
+    async def restart(self, in_pause=False):
+        status = await self.stop()
+        await self.start(in_pause=in_pause)
         return status
     
-    def update_config(self, filters_codes):
-        with self.update_config_lock:
+    async def update_config(self, filters_codes):
+        async with self.update_config_lock:
             if (self.isactive()):
-                self.process.stdin.write(" ".join(filters_codes)+"\n")
-                self.process.stdin.flush()
+                self.process.stdin.write((" ".join(filters_codes)+"\n").encode())
+                await self.process.stdin.drain()
 
-    def reload(self):
+    async def reload(self):
         if self.isactive():
-            with self.filter_map_lock:
+            async with self.filter_map_lock:
                 self.filter_map = self.compile_filters()
                 filters_codes = list(self.filter_map.keys())
-                self.update_config(filters_codes)
+                await self.update_config(filters_codes)
 
     def isactive(self):
-        if self.process and not self.process.poll() is None:
+        if self.process and not self.process.returncode is None:
             self.process = None
         return True if self.process else False
 
-    def pause(self):
+    async def pause(self):
         if self.isactive():
-            self.update_config([])
+            await self.update_config([])
         else:
-            self.start(in_pause=True)
+            await self.start(in_pause=True)
 
     def compile_filters(self):
         res = {}