我有一个要在 CPython 和 MicroPython 中运行的TCP 套接字服务器类。它确实可以运行，没有问题。我只是想用 IP 过滤器来扩展它的功能。在早期版本中，它仍然是一个 http 服务器，我有一个正则表达式模式来匹配请求的 IP 地址。但是我应该在哪里介入并检查远程地址以接受或拒绝连接呢？
注意：newConnection 中的读取器对象在 microPython 中没有属性“_transport”。
def log(*args, prio = 0):
if prio >= 0:
print(*args)
class TcpServer(SocketBase):
def __init__(self, port=8888, requestHandler = None, ipMask = r"192.168.(\d+).(\d+)|localhost|127.0.0.1"):
'''
:param requestHandler: callable with parameters request, response
'''
super().__init__(host="0.0.0.0", port = port)
self.requestHandler = requestHandler
self.ipRange = re.compile(ipMask)
self.timeout = TimeOut()
async def start(self):
self.server = await asyncio.start_server(self.newConnection, host=self.host, port=self.port)
log("server started at ", (self.host, self.port), prio=1)
while True:
await asyncio.sleep(10)
async def newConnection(self, reader, writer):
log("Connected")
self.connection = (reader, writer)
self.timeout.extend(30)
while not self.timeout.expired():
if not await self.pollOpenConnection(reader, writer):
# not data, then doze a while:
await asyncio.sleep(1./25.)
writer.close()
log("Disconnected")