messenger pattern#1084

Draft

rmorshea wants to merge 1 commit into

reactive-python:main from

rmorshea:messenger-pattern

Draft

messenger pattern #1084
rmorshea wants to merge 1 commit into
reactive-python:main from
rmorshea:messenger-pattern

Conversation

@rmorshea

@rmorshea rmorshea commented Jul 3, 2023 •

edited by Archmonger

Loading

Copy link

Copy Markdown

Collaborator

_{By submitting this pull request you agree that all contributions to this project are made under the MIT license.}

Issues

Currently, while the client and server have the ability to specify different message types, the server has no way of broadcasting messages other than layout events. Doing so is necessary for a number of different use cases...

Solution

Implement a generic Messenger class that allows for users to send/receive arbitrary messages to/from the client. Users will gain access to a Messenger via the use_messenger hook. Usage could look like one of the following...

Using start_consumer and start_producer:

@component
def demo_producer_consumer():
 msgr = use_messenger()
 use_effect(lambda: msgr.start_consumer("my-message-type", my_message_consumer))
 use_effect(lambda: msgr.start_producer(my_message_producer))
async def my_message_consumer(msg):
 ...
async def my_message_producer():
 while True:
 yield ...

Using send and receive:

@component
def demo_send_receive():
 msgr = use_messenger()
 @use_effect
 async def start_consumer():
 async for msg in msgr.consume("my-message-type"):
 ...
 
 @use_effect
 async def start_producer():
 while True:
 await msgr.send("my-message-type", ...)

Ultimately, the start_consumer and start_producer methods are convenience methods that call send and receive under the hood.

Checklist

Tests have been included for all bug fixes or added functionality.
The changelog.rst has been updated with any significant changes.

@rmorshea


 initial work on messenger pattern

bde1205

rmorshea

rmorshea commented

Jul 4, 2023

View reviewed changes

src/py/reactpy/reactpy/backend/messenger.py

Comment on lines +22 to +40

def start_producer(self, producer: Callable[[], AsyncIterator[Message]]) -> None:

"""Add a message producer"""

async def _producer() -> None:

async for message in producer():

await self.send(message)

self._task_group.start_soon(_producer)

def start_consumer(

self, message_type: str, consumer: Callable[[Message], Awaitable[None]]

) -> None:

"""Add a message consumer"""

async def _consumer() -> None:

async for message in self.receive(message_type):

self._task_group.start_soon(consumer, message)

self._task_group.start_soon(_consumer)

@rmorshea rmorshea Jul 4, 2023

Copy link

Copy Markdown

Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

These need to return callbacks to stop the consumer/producer - this will work well with use_effect.

@rmorshea rmorshea requested a review from Archmonger

July 4, 2023 18:22

@rmorshea rmorshea changed the title ~~(削除) initial work on messenger pattern (削除ここまで)~~ (追記) messenger pattern (追記ここまで)

Jul 4, 2023

@Archmonger

Archmonger commented Jul 4, 2023 •

edited

Loading

Copy link

Copy Markdown

Contributor

Unfortunately, async for msg in msgr.consume("my-message-type"): is fundamentally flawed since it relies on our async use_effect.

We would need a stateful async paradigm, similar to Django Channels consumers.

@rmorshea

rmorshea commented Jul 4, 2023 •

edited

Loading

Copy link

Copy Markdown

Collaborator Author

At some point we need to figure out async effects. I don't think that should be a blocker here.

Edit: see #1090

Archmonger

Archmonger reviewed

Jul 7, 2023

View reviewed changes

@Archmonger Archmonger left a comment •

edited

Loading

Copy link

Copy Markdown

Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I dislike the start_consumer and start_producer from an "ease of learning" perspective. Following the ReactJS philsophy of diluting the core API to nothing but the most basic interface, we should simplify everything to just send and receive.

The "easiest to learn" way to advertise this functionality would be as such:

@component
def demo_send_receive():
 @use_messenger
 async def consumer(send, receive):
 while True:
 msg = await receive("my-message-type")
 await send("my-message-type", ...)

Just the bare concept of async cancellation, and lack of reliability for other hooks to work in a persistent way makes the user experience really awkward though.

For example, if we force the user to enable shield_cancel, we should consider whether state and set_state will always work as expected in the following scenario:

@component
def demo_send_receive():
 state, set_state = use_state(0)
 @use_messenger
 async def consumer(send, receive):
 while True:
 msg = await msgr.receive("my-message-type")
 print(state)
 set_state(msg + 1)

Or alternatively, if we do not enable shield_cancel is there potentially a gap in time where the previous use_effect has been de-registered, but the new one is not registered yet?

src/py/reactpy/reactpy/backend/messenger.py

"""Receive messages of a given type"""

send, recv = create_memory_object_stream()

self._streams.setdefault(message_type, []).append((send, recv))

async with recv:

@Archmonger Archmonger Jul 7, 2023

Copy link

Copy Markdown

Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Would async with send, receive: not work here?

src/py/reactpy/reactpy/backend/starlette.py

Comment on lines +148 to +149

MessengerContext(

ConnectionContext(

@Archmonger Archmonger Jul 7, 2023

Copy link

Copy Markdown

Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We can consider merging everything into a ReactPyContext.

src/py/reactpy/reactpy/core/types.py

Comment on lines +76 to +78

async def renders(self) -> AsyncIterator[_Render]:

"""Render a series of updates to a view"""

@Archmonger Archmonger Jul 7, 2023 •

edited

Loading

Copy link

Copy Markdown

Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why not have AsyncIterator be an accepted type on render, and then handle detection of this within layout.py?

Less APIs is generally more intuitive than more.

@Archmonger

Archmonger commented Jul 7, 2023 •

edited

Loading

Copy link

Copy Markdown

Contributor

After today's tag up, two theoretical interfaces could look like this

Notes

Use messenger will never restart a consumer that is already running
We should timeout the consumer if it takes too long to teardown
Async function interface needs a teardown parameter in the hook

Async Function Interface

from typing import Coroutine
import asyncio
from reactpy import use_state, component, use_effect
def use_messenger(consumer: Coroutine, timeout: int = 10, teardown: Coroutine | None = None):
 ...
@component
def example():
 state, set_state = use_state(0)
 async def consumer_teardown(send, receive):
 ...
 @use_messenger(timeout=20, teardown=consumer_teardown)
 async def consumer(receive, send, cancel: asyncio.Event):
 while True:
 msg = await receive("my-message-type")
 await send("my-message-type", ...)
 print(state)
 set_state(msg + 1)
 if cancel.is_set():
 break

Class Based Interface

from typing import Coroutine
import asyncio
from reactpy import use_state, component, use_effect
def use_messenger(consumer: Coroutine, timeout: int = 10):
 ...
@component
def example():
 state, set_state = use_state(0)
 @use_messenger(timeout=20)
 class Consumer:
 async def __call__(self, receive, send, cancel: asyncio.Event):
 while True:
 msg = await receive("my-message-type")
 await send("my-message-type", ...)
 print(state)
 set_state(msg + 1)
 if cancel.is_set():
 break
 async def teardown(self):
 ...