Streaming list updates

Streaming list updates

This example shows updates that no client asked for: a background timer appends one number to rx[list[int]] every tick, entirely independent of any client action. Each tick sends one small update — the message size never grows with how many items the list already holds.

Backend

import asyncio

from rxdjango import ContextChannel, rx, action


class StreamingListChannel(ContextChannel):
    """Background-timer appends to `rx[list[int]]`: every tick sends one
    small insert, however long the list grows, never a full re-send.

    `reactive_model`'s background-work example (`ReactiveModelChannel`) uses
    a real OS `threading.Thread` because its work is a *blocking* Django ORM
    call. This channel's periodic work is a plain `asyncio.sleep` with no
    blocking I/O, so the safe and idiomatic Channels equivalent is an
    `asyncio.Task` cooperating on the consumer's own event loop instead of a
    second OS thread — mutating `self.items` from a genuinely different
    thread would race the consumer's own coroutine touching the same
    websocket.
    """

    TICK_SECONDS = 0.3

    items = rx[list[int]]([])
    ticking = rx[bool](True)

    async def on_connect(self):
        self._next_value = 1
        self._task = asyncio.create_task(self._tick_loop())

    async def on_disconnect(self):
        task = getattr(self, '_task', None)
        if task is not None:
            task.cancel()

    async def _tick_loop(self):
        try:
            while True:
                await asyncio.sleep(self.TICK_SECONDS)
                if not self.ticking:
                    continue
                self.items.append(self._next_value)
                self._next_value += 1
                # Outside an action, nothing flushes the queued update —
                # push it explicitly.
                await self._consumer._flush_rx()
        except asyncio.CancelledError:
            pass

    @action
    async def pause(self):
        self.ticking = False

    @action
    async def resume(self):
        self.ticking = True

    @action
    async def reset(self):
        self.items = []

Frontend

import React from 'react';
import { useChannel } from '@rxdjango/react';
import { StreamingListChannel } from '../../rx/streaming_list/streaming_list.channels';
import { Demo, Fields, Field, Button, Row, Note } from '../../components/demo';

export function StreamingListDemo() {
  const channel = useChannel(StreamingListChannel);

  return (
    <Demo>
      <Fields>
        <Field label="Streamed items">
          <ul className="flex flex-wrap gap-2">
            {channel.items.map((item, index) => (
              <li
                key={index}
                className="rounded-md border border-ink/40 px-2 py-1 text-sm tabular-nums"
              >
                {item}
              </li>
            ))}
          </ul>
        </Field>
      </Fields>
      <Note>
        A background timer appends one number every tick — no page reload,
        no full re-send, just a small insert op per item.
      </Note>
      <Row>
        {channel.ticking ? (
          <Button variant="secondary" onClick={() => channel.pause()}>
            Pause
          </Button>
        ) : (
          <Button onClick={() => channel.resume()}>
            Resume
          </Button>
        )}
        <Button variant="secondary" onClick={() => channel.reset()}>
          Reset
        </Button>
      </Row>
    </Demo>
  );
}

export default StreamingListDemo;