test_multiple_threads.py

import asyncio
import threading

from lahja import (
    BaseEvent,
    BaseRequestResponseEvent,
    BroadcastConfig,
    ConnectionConfig,
    Endpoint,
)


class Response(BaseEvent):
    pass


class Request(BaseRequestResponseEvent[Response]):
    @staticmethod
    def expected_response_type():
        return Response


async def run_request(endpoint):
    return await endpoint.request(Request())


async def main():
    endpoint = Endpoint()
    await endpoint.start_serving(ConnectionConfig.from_name('main'))
    await endpoint.connect_to_endpoint(ConnectionConfig.from_name('main'))

    endpoint.subscribe(Request, lambda event: endpoint.broadcast_nowait(
        Response(), event.broadcast_config()
    ))

    done_event = asyncio.Event()

    loop = asyncio.get_running_loop()
    thread = threading.Thread(target=background, args=(loop, endpoint, done_event))
    thread.start()

    await done_event.wait()


def background(event_loop, endpoint, done_event):
    fut = asyncio.run_coroutine_threadsafe(
        run_request(endpoint),
        loop=event_loop
    )
    data = fut.result(timeout=3)
    print(f'received response: {data}')

    event_loop.call_soon_threadsafe(done_event.set)


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())