Streaming data. From server to a client.

[4l1fe]

Is it possible to read data from a stream on a server side by chunks and send them to a client?

I'm making a simple app on top of LLM and it is going to be a long running task which i would like to run in a subprocess and read ready data from it by chunks sending them to a client on the go.
Another example is Redis Publish/Subscribe channels. I don't know when data will come from a channel, but would like to rerender a component when it is received.

I see that internally Websocket is used to send data to a client.

[Archmonger]

It will be possible soon. You'll need to use our upcoming use_messenger hook which provides bidirectional communication for any arbitrary data.

Once this is released released, your scenario will require attaching a receiver/listener on the client/server respectively.

[rmorshea]

I'll see if I can reproduce this.

[rmorshea]

I modified your code by implementing a function pick_random_label instead of your labeling function but was unable to reproduce the error you were seeing. Can you create a reproducible example?

Here's the code I ran with:

import asyncio
import random
from concurrent.futures import ProcessPoolExecutor

from starlette.applications import Starlette

from reactpy import component, event, html, use_effect, use_state
from reactpy.backend.starlette import Options, configure

labels = ["Baseball", "Football", "Hockey", "Basketball"]


def pick_random_label(sentence):
    print("my_func", sentence)
    return [[labels.index(random.choice(labels))]]


async def run_labeling(sentence):
    print("run labeling", sentence)
    loop = asyncio.get_running_loop()

    with ProcessPoolExecutor() as pool:
        try:
            print("run in pool")
            result = await asyncio.wait_for(
                loop.run_in_executor(pool, pick_random_label, sentence), timeout=5
            )
            label_idx = result[0][
                0
            ]  # 0 - highest probability row, 0 - tag index column
            print(labels[label_idx])
            return labels[label_idx]
        except asyncio.TimeoutError:
            print("Waited too long")
            return "error"


@component
def Labeling():
    sentence, set_sentence = use_state("")
    label, set_label = use_state("")

    @use_effect
    async def update_label():
        set_label(await run_labeling(sentence))

    @event(prevent_default=True)
    def handle_submit(event):
        set_sentence(event["target"]["elements"][0]["value"])

    return html.div(
        {"class_name": "flex flex-col gap-4 w-1/2 ms-12"},
        html.h1({"class_name": "text-info"}, "label text"),
        html.form(
            {"on_submit": handle_submit},
            html.input({"class_name": "input", "type": "text", "value": sentence}),
            html.button({"type": "submit", "class_name": "btn btn-primary"}, "Run"),
        ),
        html.div({"class_name": "badge badge-info"}, label),
    )


app = Starlette()
configure(
    app,
    Labeling,
    options=Options(
        head=html.head(
            html.link(
                {
                    "href": "https://cdn.jsdelivr.net/npm/[email protected]/dist/full.min.css",
                    "rel": "stylesheet",
                    "type": "text/css",
                }
            ),
            html.script({"src": "https://cdn.tailwindcss.com"}),
        )
    ),
)

[4l1fe]

Sry for the late reply. Well, the core difference with your code and mine is that u must run txtai instance in a subprocess, not pure python function.

Internally, txtai work with multiprocessing and , morelikely, spawn a new process in in the line

            result = await asyncio.wait_for(loop.run_in_executor(pool, labeling, sentence),
                                            timeout=5)

Here you see the variable labeling, it's an instance of class Labels(), this class spwan a process due to its internal mechanic of working with ML Models when we call it with the second variable labeling(sentence).

I guess the error could be related to inability of reactpy to deal with parallel processes properly, because with other asyncio libraries it doesn't occur. Just correct me if i'm wrong, i only suppose that.

After all, i switched to Litestar to implement logic that i need , so if you don't find this topic valuable to discuss we can close it.

[rmorshea]

I'm still spawning a subprocess in the example I gave so I don't think that's the issue. The only thing I can think of is that, somehow, you were creating too many processes. From the example you gave I wasn't able to trigger a situation where that happened. I guess we can revisit this if it comes up again.

[4l1fe]

I'm still spawning a subprocess in the example I gave so I don't think that's the issue.

Sounds good, then at the next try i'll pay more attention my local enviroment.