OtlpShipper.Buffer (otlp_shipper v0.1.1)

Copy Markdown View Source

A supervised, fixed-capacity ingress ring and single batch worker.

Producers obtain a handle once with handle/1, then call enqueue/2 without sending a message per item. Ring slots replace older entries under overload. Only one wakeup is pending at a time; export runs outside the GenServer.

One queued ring plus one in-flight batch bounds retained work. Items must fit max_item_bytes measured using Erlang external size. Batches also respect max_batch_bytes. Overflow drops old queued records, never in-flight records. Concurrent producer order is best effort. A handle is invalid after restart; adapters must acquire the replacement handle. There is no disk persistence.

The export callback receives a list and returns :ok, {:ok, :partial, count}, or a tagged error. Transport owns normal export/drop diagnostics; this module records ingress drops and crashed/timed-out callbacks. A callback is never allowed to run longer than the configured export timeout.

Summary

Functions

Enqueues an item without waiting for export.

Requests a batch flush without blocking the caller or promising delivery.

Obtains the producer handle for this process incarnation.

Returns the current queued item count; in-flight items are excluded.

Starts a buffer under the consumer's supervisor; requires :config and :export.

Functions

enqueue(handle, item)

@spec enqueue(OtlpShipper.Buffer.Handle.t(), term()) ::
  :ok | {:error, :closed | :item_too_large}

Enqueues an item without waiting for export.

:ok means queued, not delivered. Oversized items and handles whose owner has stopped return errors. Overflow replaces the oldest slot and emits drop counts.

flush(handle)

@spec flush(OtlpShipper.Buffer.Handle.t()) :: :ok

Requests a batch flush without blocking the caller or promising delivery.

handle(server)

Obtains the producer handle for this process incarnation.

size(handle)

Returns the current queued item count; in-flight items are excluded.

start_link(opts)

@spec start_link(keyword()) :: GenServer.on_start()

Starts a buffer under the consumer's supervisor; requires :config and :export.