我想在app.py上创建两个具有asyncio的协议(TcpClient和UdpServer),其中TcpClient将与服务器.py和作为UDP服务器的UdpServer建立持久连接:
我需要什么:
a) 这两个协议进行通信:互相调用方法。这只适用于第一个连接。如果TcpClient重新连接,它将无法再次发送来自UdpServer的字符串"send-to-tcp."。我用print(self)
和TcpClient进行了检查,创建了一个新实例,旧实例仍然存在,但没有连接,但我不知道如何重构它。我认为我使用异步的方式不对
b) 当TcpClient与server.py断开连接时,等待5秒,然后再次尝试重新连接,以此类推。我曾尝试使用asyncio的call_later()
来实现这一点,但我认为有一种本地方法可以做到这一点而不是技巧
c) 当我启动app.py时,如果TcpClient无法连接,我想在5秒后重新连接,以此类推。我不知道该怎么做。
这里是我的例子app.py-server.py的测试。server.py只是用于测试——这将是另一种语言。
只是想说我尝试了什么:
1) 当我启动app.py并且server.py关闭时,app.py不会重试
2) 当app.py连接到server.py,并且服务器关闭并快速启动时,TcpClient会重新连接,但我无法再连接新实例上的其他方法,并向server.py发送字符串"send-to-tcp.",只是旧的,那里没有更多的连接
3) 如果我使用asyncio.async()
而不是run_until_complete()
,我就不能从彼此的协议中调用方法。
我把app.py和server.py放在这里,这样你就可以复制并运行测试了。
我用ncat localhost 9000 -u -v
发送字符串"send to tcp."。这个字符串需要打印在UdpServer Class上,并传递到TcpClient Class上的方法send_data_to_tcp,这个方法会将字符串发送到server.py。<-首次重新连接tcpClient后,此操作不起作用。
我使用的是Python 3.4.0。
非常感谢。
app.py:
import asyncio
#TCP client
class TcpClient(asyncio.Protocol):
message = 'Testing'
def connection_made(self, transport):
self.transport = transport
self.transport.write(self.message.encode())
print('data sent: {}'.format(self.message))
server_udp[1].tcp_client_connected()
def data_received(self, data):
self.data = format(data.decode())
print('data received: {}'.format(data.decode()))
if self.data == 'Testing':
server_udp[1].send_data_to_udp(self.data)
def send_data_to_tcp(self, data):
self.transport.write(data.encode())
def connection_lost(self, exc):
msg = 'Connection lost with the server...'
info = self.transport.get_extra_info('peername')
server_udp[1].tcp_client_disconnected(msg, info)
#UDP Server
class UdpServer(asyncio.DatagramProtocol):
CLIENT_TCP_TIMEOUT = 5.0
def __init__(self):
self.client_tcp_timeout = None
def connection_made(self, transport):
print('start', transport)
self.transport = transport
def datagram_received(self, data, addr):
self.data = data.strip()
self.data = self.data.decode()
print('Data received:', self.data, addr)
if self.data == 'send to tcp.':
client_tcp[1].send_data_to_tcp(self.data)
def connection_lost(self, exc):
print('stop', exc)
def send_data_to_udp(self, data):
print('Receiving on UDPServer Class: ', (data))
def connect_client_tcp(self):
coro = loop.create_connection(TcpClient, 'localhost', 8000)
#client_tcp = loop.run_until_complete(coro)
client_tcp = asyncio.async(coro)
def tcp_client_disconnected(self, data, info):
print(data)
self.client_tcp_info = info
self.client_tcp_timeout = asyncio.get_event_loop().call_later(self.CLIENT_TCP_TIMEOUT, self.connect_client_tcp)
def tcp_client_connected(self):
if self.client_tcp_timeout:
self.client_tcp_timeout.cancel()
print('call_later cancel.')
loop = asyncio.get_event_loop()
#UDP Server
coro = loop.create_datagram_endpoint(UdpServer, local_addr=('localhost', 9000))
#server_udp = asyncio.Task(coro)
server_udp = loop.run_until_complete(coro)
#TCP client
coro = loop.create_connection(TcpClient, 'localhost', 8000)
#client_tcp = asyncio.async(coro)
client_tcp = loop.run_until_complete(coro)
loop.run_forever()
server.py:
import asyncio
class EchoServer(asyncio.Protocol):
def connection_made(self, transport):
peername = transport.get_extra_info('peername')
print('connection from {}'.format(peername))
self.transport = transport
def data_received(self, data):
print('data received: {}'.format(data.decode()))
self.transport.write(data)
# close the socket
#self.transport.close()
#def connection_lost(self):
# print('server closed the connection')
loop = asyncio.get_event_loop()
coro = loop.create_server(EchoServer, 'localhost', 8000)
server = loop.run_until_complete(coro)
print(server)
print(dir(server))
print(dir(server.sockets))
print('serving on {}'.format(server.sockets[0].getsockname()))
try:
loop.run_forever()
except KeyboardInterrupt:
print("exit")
finally:
server.close()
loop.close()
您实际上只需要一些小的修复。首先,我编写了一个协同程序来处理连接重试:
@asyncio.coroutine
def do_connect():
global tcp_server # Make sure we use the global tcp_server
while True:
try:
tcp_server = yield from loop.create_connection(TcpClient,
'localhost', 8000)
except OSError:
print("Server not up retrying in 5 seconds...")
yield from asyncio.sleep(5)
else:
break
然后我们用这个来启动一切:
loop = asyncio.get_event_loop()
#UDP Server
coro = loop.create_datagram_endpoint(UdpServer, local_addr=('localhost', 9000))
server_udp = loop.run_until_complete(coro)
#TCP client
loop.run_until_complete(do_connect())
loop.run_forever()
下一部分是处理app.py启动后服务器的故障/恢复。我们需要修复tcp_client_disconnected
和connect_client_tcp
才能正确处理:
def connect_client_tcp(self):
global client_tcp
task = asyncio.async(do_connect())
def cb(result):
client_tcp = result
task.add_done_callback(cb)
def tcp_client_disconnected(self, data, info):
print(data)
self.client_tcp_info = info
self.client_tcp_timeout = loop.call_later(self.CLIENT_TCP_TIMEOUT, self.connect_client_tcp)
有趣的是connect_client_tcp
。你的原始版本有两个问题:
您将
client_tcp
直接分配给asyncio.async(coro)
的结果,这意味着client_tcp
被分配给了asyncio.Task
。那不是你想要的;您希望将client_tcp
分配给已完成的asyncio.Task
的结果。我们通过使用task.add_done_callback
在Task
的结果完成后将client_tcp
分配给它来实现这一点。您忘记了方法顶部的
global client_tcp
。如果没有这一点,您只需要创建一个名为client_tcp
的局部变量,该变量在connect_client_tcp
结束时被丢弃。
一旦这些问题得到解决,我就可以运行app.py
,随时启动/停止serv.py
,但当所有三个组件一起运行时,我总能看到从ncat
到serv.py
的所有消息都正确传递。
这是完整的app.py
,便于复制/粘贴:
import asyncio
#TCP client
class TcpClient(asyncio.Protocol):
message = 'Testing'
def connection_made(self, transport):
self.transport = transport
self.transport.write(self.message.encode())
print('data sent: {}'.format(self.message))
server_udp[1].tcp_client_connected()
def data_received(self, data):
self.data = format(data.decode())
print('data received: {}'.format(data.decode()))
if self.data == 'Testing':
server_udp[1].send_data_to_udp(self.data)
def send_data_to_tcp(self, data):
self.transport.write(data.encode())
def connection_lost(self, exc):
msg = 'Connection lost with the server...'
info = self.transport.get_extra_info('peername')
server_udp[1].tcp_client_disconnected(msg, info)
#UDP Server
class UdpServer(asyncio.DatagramProtocol):
CLIENT_TCP_TIMEOUT = 5.0
def __init__(self):
self.client_tcp_timeout = None
def connection_made(self, transport):
print('start', transport)
self.transport = transport
def datagram_received(self, data, addr):
self.data = data.strip()
self.data = self.data.decode()
print('Data received:', self.data, addr)
if self.data == 'send to tcp.':
client_tcp[1].send_data_to_tcp(self.data)
def connection_lost(self, exc):
print('stop', exc)
def send_data_to_udp(self, data):
print('Receiving on UDPServer Class: ', (data))
def connect_client_tcp(self):
global client_tcp
coro = loop.create_connection(TcpClient, 'localhost', 8000)
task = asyncio.async(do_connect())
def cb(result):
client_tcp = result
task.add_done_callback(cb)
def tcp_client_disconnected(self, data, info):
print(data)
self.client_tcp_info = info
self.client_tcp_timeout = loop.call_later(self.CLIENT_TCP_TIMEOUT, self.connect_client_tcp)
def tcp_client_connected(self):
if self.client_tcp_timeout:
self.client_tcp_timeout.cancel()
print('call_later cancel.')
@asyncio.coroutine
def do_connect():
global client_tcp
while True:
try:
client_tcp = yield from loop.create_connection(TcpClient, 'localhost', 8000)
except OSError:
print("Server not up retrying in 5 seconds...")
yield from asyncio.sleep(1)
else:
break
loop = asyncio.get_event_loop()
#UDP Server
coro = loop.create_datagram_endpoint(UdpServer, local_addr=('localhost', 9000))
server_udp = loop.run_until_complete(coro)
#TCP client
loop.run_until_complete(do_connect())
loop.run_forever()