From 14cc8f11a79fabec83f3c5754af45583fb4edcd0 Mon Sep 17 00:00:00 2001
From: wu_xinjun <bent20140204@outlook.com>
Date: 星期四, 16 六月 2022 17:11:21 +0800
Subject: [PATCH] udp_get rebuild
---
src/microwave/servers.py | 127 ++++++++++++++++++++++++++++++++++-------
1 files changed, 104 insertions(+), 23 deletions(-)
diff --git a/src/microwave/servers.py b/src/microwave/servers.py
index 8686b4a..fec7edd 100644
--- a/src/microwave/servers.py
+++ b/src/microwave/servers.py
@@ -1,10 +1,13 @@
import os, sys
from ctypes import sizeof, string_at
-from socketserver import BaseRequestHandler, UDPServer
+from socketserver import _AddressType, _RequestType, BaseRequestHandler, BaseServer, UDPServer
+import threading
+from typing import Union
import warnings
+import time
+import queue
from .components import *
from .constants import *
-from ..protos import aerial_pb2 as pb
@@ -141,32 +144,110 @@
"""
micro wave udp request handler
"""
+ def __init__(self,
+ request: _RequestType,
+ client_address: _AddressType,
+ server: BaseServer) -> None:
+ super().__init__(request, client_address, server)
+
def handle(self):
# Get message and client socket
- buf, _ = self.request
- receiver = self.server.Receiver
- # add buffer to receiver
- receiver.AddBuf(buf)
- if receiver.DataFrameReady:
- # dataframe ready, add to udp_queue
- msg = pb.RequestCommand()
- msg.sys_command = pb.SysCommand.MICRO_WAVE_COMMAND
- msg.sub_command = pb.SubCommand.MICRO_WAVE_SUB_COMMAND
- data1 = pb.MicroWaveMessage()
- data1.length = len(receiver.DataFrame)
- data1.data = bytes(receiver.DataFrame)
- data1_str = data1.SerializeToString()
- msg.data1.length = len(data1_str)
- msg.data1.data = data1_str
- self.server.msg_queue.put(msg)
+ while not self.server.udp_get_stop["value"]:
+ buf, _ = self.request
+ packed_count = self.server.packed_count
+ print(f"{time.asctime()} | Got connection from {self.client_address}, {packed_count}")
+ self.server.udp_queue.put(buf)
+ self.server.packed_count += 1
+ else:
+ UDPServer.server_close(self.server)
+ self.server.udp_queue.put(False)
+ print(f"{time.asctime()} | Server is stopped by thresholds or keyborad interrupt.")
+
class MicroWaveUDPServer(UDPServer):
"""
micro wave udp server
"""
- def __init__(self, server_address, msg_queue):
- super().__init__(server_address, MicroWaveRequestHandler)
- # create receiver instance
- self.Receiver = MicroWaveReceiver()
- self.msg_queue = msg_queue
\ No newline at end of file
+ def __init__(self,
+ server_address: tuple[str,int],
+ RequestHandlerClass: MicroWaveRequestHandler,
+ udp_queue:queue.Queue,
+ udp_get_stop:dict,
+ packet_threshold:Union[int, None],
+ timeusage_threshold:Union[int, None]
+ ) -> None:
+
+ self.server_address = server_address
+ self.RequestHandlerClass = MicroWaveRequestHandler
+ self.udp_queue = udp_queue
+ self.udp_get_stop = udp_get_stop
+ self.packet_threshold = packet_threshold
+ self.timeusage_threshold = timeusage_threshold # mins
+ self.packed_count = 0
+
+ super().__init__(server_address, RequestHandlerClass)
+
+ # record start time of the server
+ while self.timeusage_threshold is not None:
+ self.start_time = time.time()
+
+ def close_request(self, request: _RequestType) -> None:
+
+ # check to stop the server via muti-thresdhold
+ while self.packet_threshold is not None and \
+ self.packed_count >= self.packet_threshold:
+
+ self.udp_get_stop["value"] = True
+
+ while self.timeusage_threshold is not None:
+ cur_time = time.time() # seconds
+ time_usage = int(cur_time - self.start_time)
+
+ if time_usage > 60*self.timeusage_threshold:
+ self.udp_get_stop["value"] = True
+
+ return super().close_request(request)
+
+
+
+class MicroWaveUDPThread(threading.Thread):
+ def __init__(self,
+ MicrowaveHost:str,
+ MicrowavePort:int,
+ udp_queue:queue.Queue,
+ udp_get_stop:dict,
+ packet_threshold:Union[int, None],
+ timeusage_threshold:Union[int, None]
+ ):
+ super().__init__()
+
+ self.MicrowaveHost = MicrowaveHost
+ self.MicrowavePort = MicrowavePort
+ self.udp_queue = udp_queue
+ self.udp_get_stop = udp_get_stop
+ self.packet_threshold = packet_threshold
+ self.timeusage_threshold = timeusage_threshold
+
+ def stop(self):
+ """stop the server to stop the """
+ self.udp_get_stop["value"] = True
+
+ def run(self):
+ server = MicroWaveUDPServer(server_address=(self.MicrowaveHost,self.MicrowavePort),
+ RequestHandlerClass=MicroWaveRequestHandler,
+ udp_queue=self.udp_queue,
+ udp_get_stop=self.udp_get_stop,
+ packet_threshold=self.packet_threshold,
+ timeusage_threshold=self.timeusage_threshold
+ )
+ try:
+ server.serve_forever()
+ except:
+ server.server_close()
+ print(time.asctime(),"UDP Getting Server Stopped with Unknown Reason!")
+
+
+
+
+
--
Gitblit v1.9.1