python - 为什么不调用 data_received()？

Question

我正在使用 Python 3.4 制作一个练习应用程序来学习 Python 的 asyncio 模块。

我的应用程序中有几个模块:

app.py

import asyncio
import logging


class Client(asyncio.Protocol):

    def __init__(self, loop):
        self.loop = loop

    def connection_made(self, transport):
        print('Connection with server established')
        self.transport = transport

    def data_received(self, data):
        data = data.decode()
        print('Received: ', data)
        if not int(data):
            print('Stopping loop')
            self.loop.stop()
        else:
            message = 'remove one'.encode()
            self.transport.write(message)
            print('Sent:', message)

    def connection_lost(self, exc):
        print('Connection with server lost.')
        self.loop.stop()


loop = asyncio.get_event_loop()

fn = loop.create_connection(
    lambda: Client(loop), '127.0.0.1', 9999
)

logging.basicConfig(level=logging.DEBUG)

client = loop.run_until_complete(fn)

try:
    loop.run_forever()
except KeyboardInterrupt:
    pass

print('Loop ended')
loop.close()

计数器.py

import asyncio


class Counter:

    def __init__(self, maxcount):
        self.maxcount = maxcount

    @asyncio.coroutine
    def count(self):
        yield from asyncio.sleep(self.maxcount)
        print('Done counting')

serie.py

import asyncio


class Serie:

    def __init__(self, loop, items=None):
        self.loop = loop
        self.items = items or []

    @asyncio.coroutine
    def remove_one(self, counter):
        if len(self.items) is not 0:
            yield from counter.count()
            item = self.items.pop(0)
            print('Removed', item)
        else:
            print('Serie is empty')

    @asyncio.coroutine
    def start_removing(self, counter):
        while self.items:
            yield from self.remove_one(counter)

        print('Serie.start_removing() has finished')

mission_control.py

import asyncio
import logging

from counter import Counter
from serie import Serie


class MissionControl(asyncio.Protocol):

    def __init__(self, loop, counter, serie):
        self.loop = loop
        self.counter = counter
        self.serie = serie

    def connection_made(self, transport):
        print('Connection established with', transport.get_extra_info('peername'))
        self.transport = transport
        self.transport.write(str(len(self.serie.items)).encode())

    def data_received(self, data):
        data = data.decode()
        print('Received:', data)
        if data == 'remove one':
            yield from self.serie.remove_one()
            print('Removed one: {}'.format(self.serie.items))
            self.transport.write(str(len(self.serie.items)).encode())
        else:
            print('Done')

    def connection_lost(self, exc):
        print('Connection with {} ended'.format(self.transport.get_extra_info('peername')))


logging.basicConfig(level=logging.DEBUG)

loop = asyncio.get_event_loop()

counter = Counter(2)
planets = Serie(loop, ['Mercúrio', 'Vênus', 'Terra', 'Marte',
                       'Júpiter', 'Saturno', 'Urano', 'Netuno'])

fn = loop.create_server(
    lambda: MissionControl(loop, counter, planets), '127.0.0.1', 9999
)

server = loop.run_until_complete(fn)

print('Server started')

try:
    loop.run_forever()
except KeyboardInterrupt:
    pass

server.close()
loop.run_until_complete(server.wait_closed())
loop.stop()
loop.close()

您还可以在此 Github gist 找到源代码.

在 mission_control.py 中，当客户端 (app.py) 通过它的 self.transport 属性。

执行错误在哪里，我该如何解决？

Answer 1

问题是 data_received 不是 ( and cannot be ) 协程，但您在其中使用 yield from。在内部 asyncio 只是调用 self.data_received(data) 而没有任何 yield from 调用，这意味着方法的主体没有被执行完全没有——而是立即返回一个生成器对象。您需要重构您的实现以不使用 yield from，这需要改用回调:

class MissionControl(asyncio.Protocol):

    def __init__(self, loop, counter, serie):
        self.loop = loop
        self.counter = counter
        self.serie = serie

    def connection_made(self, transport):
        print('Connection established with', transport.get_extra_info('peername'))
        self.transport = transport
        self.transport.write(str(len(self.serie.items)).encode())

    def data_received(self, data):
        data = data.decode()
        print('Received:', data)
        if data == 'remove one':
            fut = asyncio.async(self.serie.remove_one(self.counter))
            fut.add_done_callback(self.on_removed_one)
        else:
            print('Done')

    def on_removed_one(self, result):
        print('Removed one: {}'.format(self.serie.items))
        self.transport.write(str(len(self.serie.items)).encode())

    def connection_lost(self, exc):
        print('Connection with {} ended'.format(self.transport.get_extra_info('peername')))

另一种选择是使用asyncio Streams API而不是 asyncio.Protocol，这将允许您使用协程:

import asyncio
import logging

from counter import Counter
from serie import Serie

@asyncio.coroutine
def mission_control(reader, writer):
    counter = Counter(2)
    serie = Serie(loop, ['Mercúrio', 'Vênus', 'Terra', 'Marte',
                           'Júpiter', 'Saturno', 'Urano', 'Netuno'])
    writer.write(str(len(serie.items)).encode())
    while True:
        data = (yield from reader.read(100)).decode()
        print('Received:', data)
        if data == 'remove one':
            result = yield from serie.remove_one(counter)
        else:
            print('Done')
            return
        print('Removed one: {}'.format(serie.items))
        writer.write(str(len(serie.items)).encode())


logging.basicConfig(level=logging.DEBUG)
loop = asyncio.get_event_loop()
coro = asyncio.start_server(mission_control, '127.0.0.1', 9999, loop=loop)
server = loop.run_until_complete(coro)
# The rest is the same