TermUI.Widgets.StreamWidget (TermUI v1.0.0)

View Source

StreamWidget for displaying bounded streaming data.

StreamWidget can receive data directly through add_item/2 or through the companion GenStage consumer, providing buffer controls and stream statistics.

Usage

StreamWidget.new(
  buffer_size: 1000,
  overflow_strategy: :drop_oldest
)

Features

  • Optional GenStage consumer adapter
  • Buffer management with configurable overflow strategies
  • Pause/resume stream controls
  • Render-rate metadata for custom hosts
  • Real-time stream statistics (items/sec)

Keyboard Controls

  • Space: Toggle pause/resume
  • c: Clear buffer
  • s: Toggle stats display
  • Up/Down: Scroll through buffer
  • PageUp/PageDown: Scroll by page
  • Home/End: Jump to first/last item

GenStage Integration

The widget provides a companion consumer module that can be started separately and sends {:stream_items, events} to a process:

{:ok, consumer} = StreamWidget.Consumer.start_link(widget_pid)
GenStage.sync_subscribe(consumer, to: producer)

For an embedded widget, widget_pid is normally the runtime/root process; its optional handle_info/2 must forward that message to this module and store the returned widget state. The adapter uses GenStage's subscription demand. The widget's :demand and :render_rate_ms fields are metadata in 1.0 and do not reconfigure that subscription or throttle root rendering.

The :block overflow name is retained for compatibility, but a full buffer rejects and counts new items just like :drop_newest; it does not block a producer in 1.0. Configure max_demand/min_demand when subscribing if the producer needs tighter flow control.

Summary

Types

overflow_strategy()
stats()
stream_item()
stream_state()

Functions

add_item(state, item)

Add a single item directly.

add_items(state, items)

Add items directly (for non-GenStage sources).

buffer_count(state)

Get current buffer count.

clear(state)

Clear the buffer.

get_items(state)

Get buffer items as a list.

get_stats(state)

Get current statistics.

new(opts)

Creates new StreamWidget props.

pause(state)

Pause receiving items.

paused?(state)

Check if stream is paused.

resume(state)

Resume receiving items.

set_buffer_size(state, new_size)

Set buffer size. Will drop oldest items if new size is smaller.

set_overflow_strategy(state, strategy)

Set overflow strategy.

stream_state(state)

Get stream state.

Types

overflow_strategy()

@type overflow_strategy() :: :drop_oldest | :drop_newest | :block | :sliding

stats()

@type stats() :: %{
  items_received: non_neg_integer(),
  items_dropped: non_neg_integer(),
  items_per_second: float(),
  buffer_size: non_neg_integer(),
  buffer_capacity: non_neg_integer(),
  last_update: DateTime.t() | nil
}

stream_item()

@type stream_item() :: %{
  id: non_neg_integer(),
  timestamp: DateTime.t(),
  data: any(),
  metadata: map()
}

stream_state()

@type stream_state() :: :idle | :running | :paused | :error

Functions

add_item(state, item)

@spec add_item(map(), any()) :: {:ok, map()}

Add a single item directly.

add_items(state, items)

@spec add_items(map(), [any()]) :: {:ok, map()}

Add items directly (for non-GenStage sources).

buffer_count(state)

@spec buffer_count(map()) :: non_neg_integer()

Get current buffer count.

clear(state)

@spec clear(map()) :: {:ok, map()}

Clear the buffer.

get_items(state)

@spec get_items(map()) :: [stream_item()]

Get buffer items as a list.

get_stats(state)

@spec get_stats(map()) :: stats()

Get current statistics.

new(opts)

@spec new(keyword()) :: map()

Creates new StreamWidget props.

Options

  • :buffer_size - Maximum items in buffer (default: 1000)
  • :overflow_strategy - What to do when buffer is full (default: :drop_oldest)
  • :demand - Reserved demand metadata (default: 10); not applied to the companion consumer subscription in 1.0
  • :show_stats - Display statistics bar (default: true)
  • :render_rate_ms - Reserved render-rate metadata (default: 100); the runtime render loop does not consume it in 1.0
  • :item_renderer - Function to render each item (fn item -> String.t)
  • :on_item - Callback when item is received
  • :on_error - Callback when error occurs

pause(state)

@spec pause(map()) :: {:ok, map()}

Pause receiving items.

paused?(state)

@spec paused?(map()) :: boolean()

Check if stream is paused.

resume(state)

@spec resume(map()) :: {:ok, map()}

Resume receiving items.

set_buffer_size(state, new_size)

@spec set_buffer_size(map(), non_neg_integer()) :: {:ok, map()}

Set buffer size. Will drop oldest items if new size is smaller.

set_overflow_strategy(state, strategy)

@spec set_overflow_strategy(map(), overflow_strategy()) :: {:ok, map()}

Set overflow strategy.

stream_state(state)

@spec stream_state(map()) :: stream_state()

Get stream state.