ETS-based partitioned queue buffer for high-throughput data processing.

PartitionedBuffer.Queue buffers arbitrary data in insertion order and periodically processes it using a configurable processor function. It implements partitioning to reduce lock contention during high-throughput writes, and uses double-buffering to ensure zero-downtime processing.

Examples

Standalone Usage

# Start a queue buffer with a custom processor
iex> {:ok, _sup_pid} =
...>   PartitionedBuffer.Queue.start_link(
...>     name: :my_buffer,
...>     processor: fn batch -> IO.inspect(batch) end
...>   )

# Push a single item into the buffer
iex> PartitionedBuffer.Queue.push(:my_buffer, "item1")
:ok

# Push a batch of items
iex> PartitionedBuffer.Queue.push(:my_buffer, ["item2", "item3"])
:ok

# Check buffer size
iex> PartitionedBuffer.Queue.size(:my_buffer)
3

# Stop the buffer gracefully (processes remaining items)
iex> PartitionedBuffer.Queue.stop(:my_buffer)
:ok

Adding to a Supervision Tree

children = [
  {PartitionedBuffer.Queue,
   name: :my_buffer,
   processor: &MyApp.EventProcessor.process_batch/1}
]

Supervisor.start_link(children, strategy: :one_for_one)

Processor

The processor function receives a list of values (the items pushed to the buffer):

fn batch ->
  # batch is [value1, value2, ...]
  Enum.each(batch, fn value -> process(value) end)
end

Options

See PartitionedBuffer for start and runtime options.