由于在asyncio
模块中缺少recvmsg()
函数,因此我尝试以与BaseEventLoop.sock_recv()
相同的方式重新实现它:
import asyncio, socket
def _sock_recvmsg(loop, fut, registered, sock, bufsize, ancbufsize):
self = loop
fd = sock.fileno()
if registered: self.remove_reader(fd)
if fut.cancelled(): return
try: data = sock.recvmsg(bufsize, ancbufsize)
except (BlockingIOError, InterruptedError): self.add_reader(fd, self._sock_recvmsg, fut, True, sock, bufsize, ancbufsize)
except Exception as exc: fut.set_exception(exc)
else: fut.set_result(data)
def sock_recvmsg(loop, sock, bufsize, ancbufsize=0):
self = loop
if self._debug and sock.gettimeout() != 0: raise ValueError('the socket must be non-blocking')
fut = asyncio.futures.Future(loop=self)
self._sock_recvmsg(fut, False, sock, bufsize, ancbufsize)
return fut
asyncio.unix_events._UnixSelectorEventLoop._sock_recvmsg = _sock_recvmsg
asyncio.unix_events._UnixSelectorEventLoop.sock_recvmsg = sock_recvmsg
但是这个简单的测试失败了,只有第一个值被接收到,之后测试就会挂起:
async def produce():
for i in range(5):
end_out.sendmsg([bytes([i])])
await asyncio.sleep(1)
end_out.close()
async def consume():
while True:
v, a, b, c = await asyncio.get_event_loop().sock_recvmsg(end_in, 1)
if v == b'': return
print(v)
if __name__ == '__main__':
end_in, end_out = socket.socketpair()
asyncio.ensure_future(produce())
asyncio.get_event_loop().run_until_complete(consume())
我漏掉了什么?