UmaDbClient (uma_db_client v0.7.5)

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

Usage

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

# Append events
event = UmaDbClient.Builder.event(type: "OrderPlaced", tags: ["order:1"], data: 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)

Encoding and decoding payloads

Event data is raw bytes, so callers normally encode and decode by hand. To configure that once, use this module to generate a client facade:

defmodule MyApp.DB do
  use UmaDbClient, encode: &Jason.encode!/1, decode: &Jason.decode!/1
end

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

event = MyApp.DB.event(type: "OrderPlaced", tags: ["order:1"], data: %{id: 1})
{:ok, position} = MyApp.DB.append(channel, [event])

{:ok, stream} = MyApp.DB.read(channel)
Enum.each(stream, fn %UmaDbClient.Event{data: data} -> IO.inspect(data) end)

See __using__/1 for the full list of generated functions.

Summary

Functions

__using__(opts)

Generates a client facade with payload encoding and decoding configured.

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

using(opts)

(macro)

Generates a client facade with payload encoding and decoding configured.

Options

:encode — applied to event/1's :data before it is sent
:decode — applied to each event's data when read

Both are optional and independent. Each accepts a one-argument function, a {module, function} tuple, or a module — exporting encode!/1 for :encode and decode!/1 for :decode, so encode: Jason, decode: Jason works.

Generated functions

event/1 — like UmaDbClient.Builder.event/1, but :data is a term run through :encode. Pass :raw_data instead to supply already-encoded bytes; passing both raises ArgumentError.
read/2 and subscribe/2 — return a lazy stream of UmaDbClient.Event structs with data decoded via :decode.
connect/2, append/3, head/1, get_tracking_info/2 — delegate to this module.
query/1, query_item/2, append_condition/1, tracking_info/2 — delegate to UmaDbClient.Builder.

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

:condition — an UmaDb.V1.AppendCondition struct for optimistic concurrency
:tracking_info — an UmaDb.V1.TrackingInfo struct to update a tracking cursor

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