UmaDbClient (uma_db_client v0.6.4)

Copy Markdown

Elixir gRPC client for UmaDB's DCB (Dynamic Consistency Boundary) service.

Usage

{:ok, channel} = UmaDbClient.connect("localhost:50051")

# Append events
event = UmaDbClient.Builder.event("OrderPlaced", ["order:1"], Jason.encode!(%{id: 1}))
{:ok, position} = UmaDbClient.append(channel, [event])

# Read events as a lazy stream
{:ok, stream} = UmaDbClient.read(channel)
Enum.each(stream, fn %UmaDb.V1.SequencedEvent{position: pos, event: e} ->
  IO.inspect({pos, e.event_type})
end)

TLS

Pass gRPC credentials via opts to connect/2:

cred = GRPC.Credential.new(ssl: [cacertfile: "/path/to/ca.pem"])
{:ok, channel} = UmaDbClient.connect("myserver:443", cred: cred)

Summary

Types

channel()
position()

Functions

append(channel, events, opts \\ [])

Appends one or more events to the log.

connect(target, opts \\ [])

Opens a gRPC channel to a UmaDB server.

get_tracking_info(channel, source)

Returns the last tracked position for the given source identifier.

head(channel)

Returns the current head position of the event log.

read(channel, opts \\ [])

Returns a lazy Enumerable of UmaDb.V1.SequencedEvent structs from the log.

subscribe(channel, opts \\ [])

Returns a lazy Enumerable of UmaDb.V1.SequencedEvent structs, starting from the given position and continuing indefinitely as new events arrive.

Types

channel()

@type channel() :: GRPC.Channel.t()

position()

@type position() :: non_neg_integer()

Functions

append(channel, events, opts \\ [])

@spec append(channel(), [UmaDb.V1.Event.t()], keyword()) ::
  {:ok, position()} | {:error, term()}

Appends one or more events to the log.

Returns {:ok, position} where position is the log position after the append.

Options

connect(target, opts \\ [])

@spec connect(
  String.t(),
  keyword()
) :: {:ok, channel()} | {:error, term()}

Opens a gRPC channel to a UmaDB server.

target is a "host:port" string. opts are passed directly to GRPC.Stub.connect/2 (e.g. cred: for TLS, interceptors: etc.).

get_tracking_info(channel, source)

@spec get_tracking_info(channel(), String.t()) ::
  {:ok, position() | nil} | {:error, term()}

Returns the last tracked position for the given source identifier.

Returns {:ok, nil} when no tracking record exists for source.

head(channel)

@spec head(channel()) :: {:ok, position() | nil} | {:error, term()}

Returns the current head position of the event log.

Returns {:ok, nil} when the log is empty.

read(channel, opts \\ [])

@spec read(
  channel(),
  keyword()
) :: {:ok, Enumerable.t()} | {:error, term()}

Returns a lazy Enumerable of UmaDb.V1.SequencedEvent structs from the log.

The stream is server-driven: each element is received from the server as it is produced. Iterate with Enum.to_list/1, Stream.each/2, etc.

Mid-stream gRPC errors raise GRPC.RPCError. Wrap iteration in try/rescue if you need to recover from them.

Options

  • :query — UmaDb.V1.Query to filter by event type and/or tags
  • :start — starting position (inclusive)
  • :backwards — if true, read in reverse order
  • :limit — maximum number of events to return
  • :subscribe — if true, keep the stream open for new events after catching up
  • :batch_size — number of events per server response batch

subscribe(channel, opts \\ [])

@spec subscribe(
  channel(),
  keyword()
) :: {:ok, Enumerable.t()} | {:error, term()}

Returns a lazy Enumerable of UmaDb.V1.SequencedEvent structs, starting from the given position and continuing indefinitely as new events arrive.

This is a long-running server-push stream. Blocking iteration (e.g. via Enum.each/2) will not return until the server closes the stream or an error occurs. Consider running it in a dedicated process.

Options

  • :query — UmaDb.V1.Query to filter by event type and/or tags
  • :after — only receive events after this position
  • :batch_size — number of events per server response batch