Server-py/sri-pysdk/xserver/xudp.py

"""
python3 /home/server/repositories/repositories/sri-project.demo-py/3rdparty/xserver/xudp.py
"""
import asyncio
import struct


# from concurrent.futures import ThreadPoolExecutor


class CustomProtocol:
    clients = {}

    # executor = ThreadPoolExecutor()

    def connection_made(self, transport):
        self.transport = transport
        self.peername = None  # UDP没有真正的连接，因此peername为空
        print(f"UDP Server started", flush=True)

    def datagram_received(self, data, addr):
        asyncio.create_task(self.handle_data(data, addr))

    def error_received(self, exc):
        print(f"Error received: {exc}", flush=True)

    async def handle_data(self, data, addr):
        if len(data) < 4:
            return  # Ensure complete data packet is received
        loop = asyncio.get_running_loop()
        values = await loop.run_in_executor(None, self.process_data, data)
        # values = await loop.run_in_executor(CustomProtocol.executor, self.process_data, data)
        print(f"Received values: {values} from {addr}", flush=True)

        response = struct.pack('!hh', values[0] + 1, values[1] + 1)
        self.transport.sendto(response, addr)

    @staticmethod
    def process_data(data):
        values = struct.unpack('!hh', data)
        print(f"Processing data: {values}", flush=True)
        # Simulate some processing
        return (values[0] + 1, values[1] + 1)


async def main():
    loop = asyncio.get_running_loop()
    transport, protocol = await loop.create_datagram_endpoint(
        lambda: CustomProtocol(),
        local_addr=('127.0.0.1', 20917)
    )

    print("UDP Server listening on 127.0.0.1:20917", flush=True)
    try:
        await asyncio.Future()  # Keep running indefinitely
    finally:
        transport.close()  # Close the transport


if __name__ == "__main__":
    asyncio.run(main())