使用asyncio实现的UDP服务器

Question

我正在尝试在Python的asyncio循环中接收UDP数据包。由于对asyncio非常新，所以我可能做错了什么，因为回调从未被调用：

import asyncio

class DiscoveryProtocol(asyncio.DatagramProtocol):
    def __init__(self):
        super().__init__()
    def connection_made(self, transport):
        self.transport = transport
    def datagram_received(self, data, addr):
        print(data)

def start_discovery():
    loop = asyncio.get_event_loop()
    t = loop.create_datagram_endpoint(DiscoveryProtocol,local_addr=('',5006))
    loop.run_until_complete(t)
    loop.run_forever()

我可以使用普通套接字（不使用asyncio）来接收数据包。

我做错了什么？

- chaos.ct

---

你的代码在我的机器上运行良好，我只需要将 '' 替换为 '0.0.0.0'。我使用 netcat 客户端运行了这个测试：nc -u localhost 5006。 - Vincent

哦，也许是因为我正在尝试接收广播UDP？我会再次测试一下... - chaos.ct

与UPP广播相关：asyncio问题＃480和cpython PR＃423。 - Vincent

Answer 1

没有被接受的答案，所以这个问题似乎已经萎缩了，但它会在搜索中出现。如果有人来到这里想要一个最终解决方案，下面的代码片段展示了一个完全功能的UDP服务器。 write_messages() 函数只是一个测试方法。它读取一个日志文件，内容可以自定义，并将每一行作为Syslog消息发布到UDP端口514上。将其作为脚本运行可展示服务器监听并打印从Syslog中读取的内容。根据您的格式/处理需求更新 SyslogProtocol。

import socket
import asyncio
import os, random

HOST, PORT = 'localhost', 514


def send_test_message(message: 'Message to send to UDP port 514') -> None:
    sock = socket.socket(socket.AF_INET,  # Internet
                         socket.SOCK_DGRAM)  # UDP
    sock.sendto(message.encode(), (HOST, PORT))


async def write_messages() -> "Continuously write messages to UDP port 514":
    dir_path = os.path.dirname(os.path.realpath(__file__))
    fp = open(os.path.join(dir_path, "tests/example.log"))
    print("writing")
    for line in fp.readlines():
        await asyncio.sleep(random.uniform(0.1, 3.0))
        send_test_message(line)


class SyslogProtocol(asyncio.DatagramProtocol):
    def __init__(self):
        super().__init__()

    def connection_made(self, transport) -> "Used by asyncio":
        self.transport = transport

    def datagram_received(self, data, addr) -> "Main entrypoint for processing message":
        # Here is where you would push message to whatever methods/classes you want.
        print(f"Received Syslog message: {data}")


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    t = loop.create_datagram_endpoint(SyslogProtocol, local_addr=('0.0.0.0', PORT))
    loop.run_until_complete(t) # Server starts listening
    loop.run_until_complete(write_messages()) # Start writing messages (or running tests)
    loop.run_forever()

- JohnMetta