Вызов сопрограммы из asyncio.Protocol.data_received

Это похоже на вызов сопрограммы в asyncio.Protocol.data_received, но я думаю, что это требует нового вопроса.

У меня есть простой сервер, настроенный как

loop.create_unix_server(lambda: protocol, path=serverSocket)

Он отлично работает, если я это делаю

 def data_received(self, data):
    data = b'data reply'
    self.send(data)

мой клиент получает ответ. Но я не могу заставить его работать с любым вызовом asyncio. Я пробовал все из следующих, и никто из них не работал.

@asyncio.coroutine
def go(self):
    yield from asyncio.sleep(1, result = b'data reply')

def data_received(self, data):
    print('Data Received', flush=True)

    task = asyncio.get_event_loop().create_task(self.go())
    data = yield from asyncio.wait_for(task,10)
    self.send(data)

что кто-то висел и ничего не печатал (если я украсил data_received с помощью @asyncio.coroutine, я понял, что это не получается) OK, я понимаю, что использование yield в data_received неверно.

Если я попробую новый цикл событий, как показано ниже, который находится в run_until_complete

    loop = asyncio.new_event_loop()
    task = loop.create_task(self.go())
    loop.run_until_complete(task)
    data = task.result()
    self.send(data)

Если я использую Future, который также зависает в run_until_complete

@asyncio.coroutine
def go(self, future):
    yield from asyncio.sleep(1)
    future.set_result(b'data reply')

def data_received(self, data):
    print('Data Received', flush=True)

    loop = asyncio.new_event_loop()
    future = asyncio.Future(loop=loop)
    asyncio.async(self.go(future))
    loop.run_until_complete(future)
    data = future.result()
    self.send(data)

Следующее приближается, но оно возвращается немедленно, а результат имеет тип asyncio.coroutines.CoroWrapper, подразумевая, что строка wait_for немедленно возвращается с незавершенной задачей?

@asyncio.coroutine
def go(self):
    return(yield from asyncio.sleep(3, result = b'data reply'))

@asyncio.coroutine
def go2(self):
    task = asyncio.get_event_loop().create_task(self.go())
    res = yield from asyncio.wait_for(task, 10)
    return result

def data_received(self, data):
    print('Data Received', flush=True)

    data = self.go2()
    self.send(data)

Я немного застрял на самом деле, и буду благодарен некоторым указаниям о том, на что смотреть.

Ответ 1

Вам нужно добавить свою сопрограмму в цикл событий, а затем использовать Future.add_done_callback для обработки результата при завершении сопрограммы:

@asyncio.coroutine
def go(self):
    return(yield from asyncio.sleep(3, result = b'data reply'))

def data_received(self, data):
    print('Data Received', flush=True)

    task = asyncio.async(self.go()) # or asyncio.get_event_loop().create_task()
    task.add_done_callback(self.handle_go_result)

def handle_go_result(self, task):
    data = task.result()
    self.send(data)

Вызов coroutine непосредственно в data_received просто не разрешен, так как вызывающий не собирается пытаться yield from его, а создание/запуск нового цикла событий внутри data_received всегда будет заканчиваться блокируя цикл основного события, пока цикл внутреннего события не завершит свою работу.

Вы просто хотите запланировать некоторую работу с контуром основного события (asyncio.async/loop.create_task()) и запланировать обратный вызов для выполнения, когда работа выполнена (add_done_callback).