Twisted - как сделать много кода Python неблокирующим

Question

Я пытался заставить этот скрипт выполнить код в hub() в письменном порядке.

hub() содержит смесь стандартного кода Python и запросов на выполнение операций ввода-вывода с использованием Twisted и Crossbar.

Однако, поскольку код Python блокируется, у реактора нет никаких шансов выполнить эти задачи "публикации". Мой интерфейс получает все опубликованные сообщения в конце.

1. Этот код является чрезвычайно упрощенной версией того, с чем я на самом деле имею дело. Настоящий сценарий (hub() и другие методы, которые он вызывает) длиной более 1500 строк. Модификация всех этих функций, чтобы сделать их неблокирующими, не идеальна. Я бы предпочел выделить изменения в несколько методов, таких как publish() если это возможно, чтобы решить эту проблему.
2. Я играл с такими терминами, как async, await, deferLater, loopingCall, и другие. Я не нашел пример, который помог бы в моей ситуации.

Есть ли способ изменить publish() (или же hub() ) так они рассылают сообщения по порядку?

from autobahn.twisted.component import Component, run
from twisted.internet.defer import inlineCallbacks, returnValue
from twisted.internet import reactor, defer

component = Component(
    transports=[
        {
            u"type": u"websocket",
            u"url": u"ws://127.0.0.1:8080/ws",
            u"endpoint": {
                u"type": u"tcp",
                u"host": u"localhost",
                u"port": 8080,
            },
            u"options": {
                u"open_handshake_timeout": 100,
            }
        },
    ],
    realm=u"realm1",
)

@component.on_join
@inlineCallbacks
def join(session, details):
    print("joined {}: {}".format(session, details))

    def publish(context='output', value='default'):
        """ Publish a message. """
        print('publish', value)
        session.publish(u'com.myapp.universal_feedback', {"id": context, "value": value})

    def hub(thing):
        """ Main script. """
        do_things
        publish('output', 'some data for you')
        do_more_things
        publish('status', 'a progress message')
        do_even_more_things
        publish('status', 'some more data')
        do_all_the_things
        publish('other', 'something else')

    try:
        yield session.register(hub, u'com.myapp.hello')
        print("procedure registered")
    except Exception as e:
        print("could not register procedure: {0}".format(e))


if __name__ == "__main__":
    run([component])
    reactor.run()

Answer 1

Ваш join() функция асинхронна (украшена @inlineCallbacks и содержит по крайней мере один yield в теле).

Внутренне это регистрирует функцию hub() как WAMP RPC; hub() однако не асинхронный.

Также звонки session.publish() не выдаются, как должны быть асинхронные вызовы.

Результат: вы добавляете кучу событий в цикл событий, но не ожидаете их, пока не сбросите цикл событий при завершении работы приложения.

Answer 2

Вы должны сделать свою функцию концентратором и опубликовать асинхронный.

@inlineCallbacks
def publish(context='output', value='default'):
    """ Publish a message. """
    print('publish', value)
    yield session.publish(u'com.myapp.universal_feedback', {"id": context, "value": value})

@inlineCallbacks
def hub(thing):
    """ Main script. """
    do_things
    yield publish('output', 'some data for you')
    do_more_things
    yield publish('status', 'a progress message')
    do_even_more_things
    yield publish('status', 'some more data')
    do_all_the_things
    yield publish('other', 'something else')