barrel_changes_stream (barrel_docdb v1.2.0)

View Source

Changes stream for barrel_docdb

A gen_statem process that provides a streaming interface to the changes feed. Supports two modes: - iterate: Pull-based, client calls next/1 to get changes - push: Push-based, changes are sent to subscriber

Summary

Types

att_info/0
change/0
db_config/0
db_name/0
db_ref/0
doc/0
doc_info/0
docid/0
endpoint/0
rep_options/0
rev_info/0
revid/0
revtree/0
seq/0
seq_string/0
stream_mode/0
stream_opts/0
view_name/0
view_result/0

Functions

ack(Pid, ReqId)

Acknowledge receipt of changes (push mode)

await(Pid)

Wait for changes (push mode)

await(Pid, Timeout)

Wait for changes with timeout (push mode). Distinguishes a genuine timeout (timeout) from the stream process dying (down), so a caller polling on a bounded timeout can tell "nothing new yet" from "this stream is gone."

callback_mode()
init(_)
iterate(EventType, Event, State)
next(Pid)

Get the next change (iterate mode)

push(EventType, Event, State)
start_link(StoreRef, DbName, Opts)

Start a changes stream

stop(Pid)

Stop the stream

terminate(Reason, State, Data)
wait_pending(EventType, Event, State)

Types

att_info/0

-type att_info() ::
          #{name := binary(),
            content_type := binary(),
            length := non_neg_integer(),
            digest := binary(),
            chunked => boolean(),
            chunk_size => pos_integer(),
            chunk_count => pos_integer()}.

change/0

-type change() :: map().

db_config/0

-type db_config() :: #{path => string(), store => module(), atom() => term()}.

db_name/0

-type db_name() :: binary().

db_ref/0

-type db_ref() :: pid() | atom().

doc/0

-type doc() :: #{binary() => term()}.

doc_info/0

-type doc_info() :: #{id := docid(), rev := revid(), deleted := boolean(), revtree := revtree()}.

docid/0

-type docid() :: binary().

endpoint/0

-type endpoint() :: db_name() | {node(), db_name()} | {module(), term()}.

rep_options/0

-type rep_options() ::
          #{continuous => boolean(),
            since => seq_string(),
            filter => fun((doc()) -> boolean()),
            atom() => term()}.

rev_info/0

-type rev_info() ::
          #{id := revid(),
            parent := revid() | undefined,
            deleted := boolean(),
            attachments => #{binary() => att_info()}}.

revid/0

-type revid() :: binary().

revtree/0

-type revtree() :: #{revid() => rev_info()}.

seq/0

-type seq() :: barrel_hlc:timestamp().

seq_string/0

-type seq_string() :: binary().

stream_mode/0

-type stream_mode() :: iterate | push.

stream_opts/0

-type stream_opts() ::
          #{since => seq() | first,
            mode => stream_mode(),
            include_docs => boolean(),
            interval => pos_integer(),
            batch_size => pos_integer(),
            owner => pid()}.

view_name/0

-type view_name() :: binary().

view_result/0

-type view_result() :: #{key := term(), value := term(), id := docid()}.

Functions

ack(Pid, ReqId)

-spec ack(pid(), reference()) -> ok.

Acknowledge receipt of changes (push mode)

await(Pid)

-spec await(pid()) -> {reference(), [barrel_changes:change()]} | down.

Wait for changes (push mode)

await(Pid, Timeout)

-spec await(pid(), timeout()) -> {reference(), [barrel_changes:change()]} | timeout | down.

Wait for changes with timeout (push mode). Distinguishes a genuine timeout (timeout) from the stream process dying (down), so a caller polling on a bounded timeout can tell "nothing new yet" from "this stream is gone."

callback_mode()

init(_)

iterate(EventType, Event, State)

next(Pid)

-spec next(pid()) -> {ok, barrel_changes:change()} | done | {error, term()}.

Get the next change (iterate mode)

push(EventType, Event, State)

start_link(StoreRef, DbName, Opts)

-spec start_link(barrel_store_rocksdb:db_ref(), db_name(), stream_opts()) ->
                    {ok, pid()} | {error, term()}.

Start a changes stream

stop(Pid)

-spec stop(pid()) -> ok.

Stop the stream

terminate(Reason, State, Data)

wait_pending(EventType, Event, State)