Drafter.Widget.DataChannel (drafter v0.3.1)

Buffered data ingestion channel for a single widget.

Accepts data pushes into a bounded RingBuffer and throttles render notifications. Data accumulates between renders — the widget only re-renders when the throttle window opens and new data has arrived since the last render.

A push marks the channel dirty and, unless it is paused or a window is already open, schedules a :data_channel_tick message to the calling process. With a :on_demand throttle the message is sent immediately instead of after a delay. The owner reacts to the tick by calling flush/1, which clears the dirty flag and the pending window but leaves the buffer contents in place for Drafter.Widget.apply_data_buffer/3 to read.

Summary

Types

t()

Functions

buffer(data_channel)

The channel's ring buffer, holding everything pushed so far up to its capacity.

dirty?(data_channel)

Whether anything has been pushed since the last flush/1.

flush(ch)

Closes the throttle window and reports whether anything arrived while it was open.

new(buffer_size, throttle_ms)

Opens a channel over a ring buffer of buffer_size items.

pause(ch)

Stops tick notifications and cancels any pending window.

paused?(data_channel)

Whether the channel is paused.

push(ch, item)

Appends one item to the buffer, marks the channel dirty and opens a throttle window if none is open.

push_many(ch, items)

Appends every item of items in order, otherwise behaving exactly like push/2.

resize_buffer(ch, new_size)

Changes the buffer's capacity, keeping the newest items that still fit. Leaves the dirty, paused and throttle state alone.

resume(ch)

Clears the paused flag and opens a throttle window when the channel is dirty.

Types

t()

@type t() :: %Drafter.Widget.DataChannel{
  buffer: Drafter.RingBuffer.t(),
  dirty: boolean(),
  paused: boolean(),
  throttle_ms: pos_integer() | :on_demand,
  throttle_ref: reference() | :immediate | nil
}

Functions

buffer(data_channel)

@spec buffer(t()) :: Drafter.RingBuffer.t()

The channel's ring buffer, holding everything pushed so far up to its capacity.

dirty?(data_channel)

@spec dirty?(t()) :: boolean()

Whether anything has been pushed since the last flush/1.

iex> Drafter.Widget.DataChannel.new(4, 50) |> Drafter.Widget.DataChannel.dirty?()
false

flush(ch)

@spec flush(t()) :: {t(), boolean()}

Closes the throttle window and reports whether anything arrived while it was open.

Returns {channel, true} when the channel was dirty, having cleared both the dirty flag and the pending window, and {channel, false} unchanged otherwise. The buffer is left intact, so the caller still sees every item pushed since the buffer was created.

iex> ch = Drafter.Widget.DataChannel.new(4, 50) |> Drafter.Widget.DataChannel.push(:a)
iex> {ch, flushed?} = Drafter.Widget.DataChannel.flush(ch)
iex> {flushed?, Drafter.Widget.DataChannel.dirty?(ch), elem(Drafter.Widget.DataChannel.flush(ch), 1)}
{true, false, false}

new(buffer_size, throttle_ms)

@spec new(pos_integer(), pos_integer() | :on_demand) :: t()

Opens a channel over a ring buffer of buffer_size items.

throttle_ms is the minimum gap in milliseconds between tick notifications, or :on_demand to notify on the next message pass. The channel starts clean and unpaused with no window open.

iex> ch = Drafter.Widget.DataChannel.new(4, 50)
iex> {Drafter.Widget.DataChannel.dirty?(ch), Drafter.Widget.DataChannel.paused?(ch)}
{false, false}

pause(ch)

@spec pause(t()) :: t()

Stops tick notifications and cancels any pending window.

Pushes still land in the buffer and still mark the channel dirty while it is paused; resume/1 opens a window immediately if there is anything to report.

iex> ch = Drafter.Widget.DataChannel.new(4, 50) |> Drafter.Widget.DataChannel.pause()
iex> Drafter.Widget.DataChannel.paused?(ch)
true

paused?(data_channel)

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

Whether the channel is paused.

iex> Drafter.Widget.DataChannel.new(4, 50) |> Drafter.Widget.DataChannel.paused?()
false

push(ch, item)

@spec push(t(), term()) :: t()

Appends one item to the buffer, marks the channel dirty and opens a throttle window if none is open.

Pushing to a paused channel still buffers the item but schedules nothing.

iex> ch = Drafter.Widget.DataChannel.new(4, 50) |> Drafter.Widget.DataChannel.push(:a)
iex> {Drafter.Widget.DataChannel.dirty?(ch), Drafter.RingBuffer.to_list(Drafter.Widget.DataChannel.buffer(ch))}
{true, [:a]}

push_many(ch, items)

@spec push_many(t(), Enumerable.t()) :: t()

Appends every item of items in order, otherwise behaving exactly like push/2.

Only one throttle window is opened for the whole batch. Items beyond the buffer's capacity push the oldest ones out.

iex> ch = Drafter.Widget.DataChannel.new(2, :on_demand)
iex> ch = Drafter.Widget.DataChannel.push_many(ch, [1, 2, 3])
iex> Drafter.RingBuffer.to_list(Drafter.Widget.DataChannel.buffer(ch))
[2, 3]

resize_buffer(ch, new_size)

@spec resize_buffer(t(), pos_integer()) :: t()

Changes the buffer's capacity, keeping the newest items that still fit. Leaves the dirty, paused and throttle state alone.

iex> ch = Drafter.Widget.DataChannel.new(4, :on_demand)
iex> ch = Drafter.Widget.DataChannel.push_many(ch, [1, 2, 3, 4])
iex> ch = Drafter.Widget.DataChannel.resize_buffer(ch, 2)
iex> Drafter.RingBuffer.to_list(Drafter.Widget.DataChannel.buffer(ch))
[3, 4]

resume(ch)

@spec resume(t()) :: t()

Clears the paused flag and opens a throttle window when the channel is dirty.

iex> ch = Drafter.Widget.DataChannel.new(4, 50) |> Drafter.Widget.DataChannel.pause()
iex> Drafter.Widget.DataChannel.resume(ch) |> Drafter.Widget.DataChannel.paused?()
false