教你如何优雅的管理UDP接口—

教你如何优雅的管理UDP接口——Python

news2024/11/25 0:44:51

设计思路主要参考了Flask框架，通过 route() 装饰器来告诉 UDE触发函数的 URL，通过蓝图接口进行分类管理。
在这里插入图片描述

ude.py 代码内容

import socket
import msgpack


class UDE:
    def __init__(self):
        self.urls = {}

    # 回调函数
    def Callback(self, packet):
        try:
            data = msgpack.unpackb(packet[0])  # 解析请求体的内容
            data["address"] = packet[1][0]# 将发送方的IP添加到请求体内
            # 判断该函数是否注册
            if data['url'] in self.urls:
                self.urls[data['url']](data)
            else:
                print(f"参数错误：{data['url']}")
        except Exception as e:
            print(f" {e}")

    def route(self, url):
        def blueprint(func):
            self.urls[url] = func

            def inner(*args, **kwargs):
                res = func(*args, **kwargs)
                return res

            return inner

        return blueprint

    def register_blueprint(self, module):
        # 寄存器蓝图，加载模块
        self.urls.update(module.urls)

    def run(self, port):
        print(f"--------------UDE-{port}--------------")
        udp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        local_addr = ('', port)
        udp_socket.bind(local_addr)
        while True:
            recv_data = udp_socket.recvfrom(1024)
            self.Callback(recv_data)

demo.py 代码内容

from ude import UDE

demo = UDE()
response = {}


@demo.route("scan")
def api_scan(request_value):
    print(f"出发scan接口，入参为：{request_value}")

main.py 代码块

from demo import demo
from ude import UDE

app = UDE()
app.register_blueprint(demo)

if __name__ == '__main__':
    app.run(8080)

sendto.py代码内容

import socket

import msgpack


def SendTo(value):
    service = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
    if type(value) == dict:
        service.sendto(msgpack.packb(value), ('192.168.3.5', 8080))


if __name__ == "__main__":
    SendTo({"url": "scan"})
    print(msgpack.packb({"url": "scan"}))