init

provider.py

@@ -0,0 +1,50 @@
+from collections import deque
+import threading
+import logging
+
+from utiles import *
+
+logging.basicConfig(level=logging.DEBUG)
+
+
+class ServiceConnection:
+    def __init__(self, conn_id: str, provider_sock: socket.socket, forward_sock: socket.socket, connector_addr: tuple):
+        self.conn_id = conn_id
+        self.provider_sock = provider_sock
+        self.forward_sock = forward_sock
+        self.connector_addr = connector_addr
+        self.running = False
+
+    def forward_thread(self):
+        while self.running:
+            try:
+                data = self.provider_sock.recv(1024)
+                pack = Package('data', {'conn_id': self.conn_id}, data)
+                self.forward_sock.sendto(pack.pack(), self.connector_addr)
+            except Exception as e:
+                logging.error(f"收到提供端数据，转发失败: {e}")
+                break
+        self.running = False
+        if self.provider_sock:
+            self.provider_sock.close()
+            self.provider_sock = None
+        return
+
+    def backward_data(self, data):
+        try:
+            self.provider_sock.sendall(data)
+        except Exception as e:
+            logging.error(f"收到客户端数据，转发失败: {e}")
+            return
+
+    def run(self):
+        self.running = True
+        threading.Thread(target=self.forward_thread, daemon=True).start()
+
+
+class ServiceProvider:
+    def __init__(self, service_name: str, host: str, port: int, sock: socket.socket):
+        self.service_name = service_name
+        self.host = host
+        self.port = port
+        self.sock = sock

utiles.py

@@ -0,0 +1,44 @@
+import json
+import socket
+import struct
+
+
+class Package:
+    def __init__(self, action: str, message: dict = None, data: bytes = None):
+        self.action = action
+        self.message = message if message else {}
+        self.data = data if data else b''
+
+    def pack(self) -> bytes:
+        action_encode = self.action.encode('utf-8')
+        message_encode = json.dumps(self.message).encode('utf-8')
+        packed = struct.pack('>I', len(action_encode)) + action_encode
+        packed += struct.pack('>I', len(message_encode)) + message_encode
+        packed += self.data
+        return packed
+
+    @classmethod
+    def unpack(cls, packed):
+        action_len = struct.unpack('>I', packed[:4])[0]
+        action = packed[4:4 + action_len].decode('utf-8')
+        message_len = struct.unpack('>I', packed[4 + action_len:8 + action_len])[0]
+        message = json.loads(packed[8 + action_len:8 + action_len + message_len].decode('utf-8'))
+        data = packed[8 + action_len + message_len:]
+        return cls(action, message, data)
+
+    def get_data(self) -> bytes:
+        return self.data
+
+    def get_message(self) -> dict:
+        return self.message
+
+    def get_action(self) -> str:
+        return self.action
+
+    def get_all(self) -> (str, dict, bytes):
+        return self.action, self.message, self.data
+
+
+def recv_package(sock) -> (tuple, Package):
+    addr, pack = sock.recvfrom(1300)
+    return addr, Package.unpack(pack)