python - 使用 asyncio 将数据从任务获取到连接

Question

所以这应该相当简单，我有一个连接到服务器的客户端，并且它可以很好地从服务器接收消息。但是我需要能够向服务器发送消息。我正在使用 asyncio 来异步处理这些，但我有一个问题。我如何将用户输入发送到客户端，以便它可以使用其传输将数据发送到服务器。这是代码。

import asyncio, zen_utils, json
from struct import *

class ChatClient(asyncio.Protocol):
    def connection_made(self, transport):
        self.transport = transport
        self.address = transport.get_extra_info('peername')
        self.data = b''
        print('Accepted connection from {}'.format(self.address))

        self.username = b'jacksonm'
        json_name = b'{"USERNAME": "'+self.username+b'"}'
        length = len(json_name)
        code_len = pack(b'!I', length)
        message = code_len + json_name
        self.json_loaded = False
        self.next_length = -1
        self.transport.write(message)

    def data_received(self, data):
        self.data += data

        if (self.json_loaded == False):
            self.compile_server_data(data)
        elif(self.json_loaded):
            if (len(self.data) > 4 and self.next_length == -1):
                self.next_length = self.data[:4]
                self.next_length = unpack(b'!I', self.next_length)[0]
                print("LENGTH: ", self.next_length)

            elif (len(self.data) >= self.next_length):
                self.data = self.data[4:]
                print("MESSAGE: ", self.data)
                self.next_length = -1

    def compile_server_data(self, data):
        if (self.data.find(b'}') != -1):
            start_index = self.data.find(b'{')
            end_index = self.data.find(b'}')

            self.json_data = self.data[start_index:end_index + 1]
            self.data = self.data[end_index + 1:]

            self.json_data = self.json_data.decode('ascii')
            self.json_data = json.loads(self.json_data)
            self.json_loaded = True

            self.print_server_status()

    def send_message(self, message):
        message = message.encode('ascii')
        length = len(message)
        code_len = pack(b'!I', length)
        message = code_len + message

    def parse_message(self, raw_message):
        message = {}
        message['SRC'] = self.username
        message['DEST'] = b'ALL'
        message['TIMESTAMP'] = int(time.time())
        message['CONTENT'] = b'test_message'

        json_message = json.loads(message)
        print (json_message)

    def print_server_status(self):
        print ("USERS:")
        for user in self.json_data["USER_LIST"]:
            print(user)
        print()
        print("MESSAGES:")
        for message in self.json_data["MESSAGES"][-10:]:
            print ("From: ", message[0], "     ", "To: ", message[1])
            print ("Message: ", message[3])
            print()

    def get_inital_data(self):
        pass

    def connection_lost(self, exc):
        if exc:
            print('Client {} error: {}'.format(self.address, exc))
        elif self.data:
            print('Client {} sent {} but then closed'
                  .format(self.address, self.data))
        else:
            print('Client {} closed socket'.format(self.address))

@asyncio.coroutine
def handle_user_input(loop):
    """reads from stdin in separate thread

    if user inputs 'quit' stops the event loop
    otherwise just echos user input
    """
    while True:
        message = yield from loop.run_in_executor(None, input, "> ")
        if message == "quit":
            loop.stop()
            return
        print(message)

if __name__ == '__main__':
    address = zen_utils.parse_command_line('asyncio server using callbacks')
    loop = asyncio.get_event_loop()
    coro = loop.create_connection(ChatClient, *address)
    server = loop.run_until_complete(coro)

    # Start a task which reads from standard input
    asyncio.async(handle_user_input(loop))

    print('Listening at {}'.format(address))
    try:
        loop.run_forever()
    finally:
        server.close()
        loop.close()

Answer 1

这在我的测试中通过使用 Python 3.4 更改一行来工作:

更改 run_until_complete() 调用以接收 (transport,protocol) 对而不是单个值。

具体来说，将server = loop.run_until_complete(coro)更改为transport, protocol = loop.run_until_complete(coro)，因为调用执行协程并返回其值这是一对元素。第一项是服务器对象，第二项是协议(protocol)对象，它是 ChatClient 的实例。

在不同的命令窗口中运行服务器，然后运行客户端。客户端代码:

import asyncio, json #, zen_utils
from struct import *

class ChatClient(asyncio.Protocol):
    def connection_made(self, transport):
        self.transport = transport
        self.address = transport.get_extra_info('peername')
        self.data = b''
        print('Accepted connection from {}'.format(self.address))

    def data_received(self, data):
        pass

    def compile_server_data(self, data):
        pass

    def send_message(self, message):
        message = message.encode('ascii')
        self.transport.write(message)

    def parse_message(self, raw_message):
        pass

    def print_server_status(self):
        pass

    def get_inital_data(self):
        pass

    def connection_lost(self, exc):
        if exc:
            print('Client {} error: {}'.format(self.address, exc))
        elif self.data:
            print('Client {} sent {} but then closed'
                  .format(self.address, self.data))
        else:
            print('Client {} closed socket'.format(self.address))

@asyncio.coroutine
def handle_user_input(loop, protocol):
    """reads from stdin in separate thread

    if user inputs 'quit' stops the event loop
    otherwise just echos user input
    """
    while True:
        message = yield from loop.run_in_executor(None, input, "> ")
        if message == "quit":
            loop.stop()
            return
        print(message)
        protocol.send_message(message)

if __name__ == '__main__':
    address = "127.0.0.1"
    port = 82

    loop = asyncio.get_event_loop()
    coro = loop.create_connection(ChatClient, address, port) #*address
    transport, protocol = loop.run_until_complete(coro)

    # Start a task which reads from standard input
    asyncio.async(handle_user_input(loop,protocol))

    print('Listening at {}'.format(address))
    try:
        loop.run_forever()
    finally:
        transport.close()
        loop.close()

服务器代码，归功于 this site :

import socket

host = ''        # Symbolic name meaning all available interfaces
port = 82     # Arbitrary non-privileged port
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.bind((host, port))
s.listen(1)
conn, addr = s.accept()
print('Connected by', addr)
while True:
    data = conn.recv(1024)
    if not data: break

    print( 'Received ', repr(data) )
    if repr(data)=="stop": break

    conn.sendall(data)
conn.close()