logdb_client

Elixir client for the LogDB event bus: publish events, consume them with a durable consumer and confirm explicitly (ack, nack, term, progress, or defer and confirm later). Transports: WebSocket, HTTP, Erlang distribution, Hybrid. Reconnects on its own.

Installation

def deps do
  [
    {:logdb_client, "~> 0.3"}
  ]
end

A consumer

defmodule MyApp.Jobs do
  use LogDB.Client.Consumer

  @impl true
  def handle_event(_type, payload, _meta, state) do
    case MyApp.run(payload) do
      :ok -> {:ack, state}
      {:error, :busy} -> {:nack, state, delay_ms: 5_000}
      {:error, :bad_input} -> {:term, state, "bad input"}
    end
  end
end

# in your supervision tree; LOGDB_URL=logdb://<token>@bus.local/default/jobs?transport=ws
{MyApp.Jobs, name: :jobs, connection: System.fetch_env!("LOGDB_URL"), parser: LogDB.Client.Parser.Json}

A client that only publishes: {LogDB.Client.Supervisor, name: :bus, connection: "logdb://<token>@bus.local"}. Several clients can sit in one supervisor: the child id is the client's name.

Connection string

logdb[s]://[token@]host[:port][/stream/consumer][?transport=…&node=…&client=…]

logdbs — TLS (wss/https). Port: 4000, or 443 with TLS.
token — the service token; or pass it as token: (if both are given they must match).
/stream/consumer — the consumer to read, as consumer: "stream/name".
transport — ws, http, hybrid or erlangd (default: http).
node — the bus node for erlangd: logdb (→ logdb@host) or a full logdb@host; node_name is the old name of the same parameter.
client — this instance's name (client_name:).

An unknown parameter is an error. The same format is used by every LogDB client (docs/connection-string.md). The token is hidden whenever the config or the string is printed.

Instead of a string: host:, port:, secure:. Anything set both in the string and as an option must agree.

Options

:name (required), :connection or :host/:port/:secure, :token, :transport, :parser, :consumer (or the old :stream + :consumer_id), :node, :client_name, :req_options (Req, for HTTP), :init_arg (your data for the consumer's init/1).

Options are checked at start: a missing token or host, a wrong scheme, an unknown option (with a hint: :consumer_ld (did you mean :consumer_id?)) — the client refuses to start and says why. The client connects asynchronously; LogDB.await_connected(name, timeout) waits for it when you need to.

Calls and replies

LogDB.publish(:bus, "rkp://jobs/run", %{"id" => 1}, stream: "default")
LogDB.publish_batch(:bus, [%{"type" => "…", "payload" => …, "stream" => "default"}])
LogDB.ack(:jobs, ids)  # also nack/3, term/3, progress/2
LogDB.ping(:bus)

Every transport replies the same way: {:ok, result} or {:error, %LogDB.Error{}} (subscribe and ping: :ok or {:error, %LogDB.Error{}}). LogDB.Error has a code (:disconnected, :timeout, :not_found, :invalid, :unauthorized, …), a message and details, and can be raised. A client that is not running or lost its connection gives {:error, %LogDB.Error{code: :disconnected}}; the caller never crashes.

If handle_event/4 raises, throws, exits or returns something else, the event is not confirmed and handle_error/2 gets the reason; the consumer keeps its state.

Choosing the consumer

consumer: "default/rkp_core"          # stream/name; the service is known from the token (LogDB with /v1)
stream: "default", consumer_id: "…"   # the old pair, works with any LogDB

Without either the client only publishes. transport: LogDB.Client.Transport.HTTP with a pull consumer receives events by long polling /v1/events/poll.

Transports

LogDB.Client.Transport.WebSocket — delivery and every command (publish, subscribe, ack/nack/term/progress, ping) over one socket; each command waits for the bus's reply.
LogDB.Client.Transport.Erlangd — Erlang distribution; payload encoded by the parser.
LogDB.Client.Transport.HTTP — publish and confirm; with a pull consumer, delivery by long polling.
LogDB.Client.Transport.Hybrid — delivery over WS; writes over the same socket when LogDB speaks client API 1, over HTTP otherwise (kept for existing setups).

LogDB.subscribe(name, "stream/name") subscribes a running client to a consumer (LogDB with /v1).

Client name

client_name: "rkp_beam@node-2"   # default: the node name, or the host name

The client introduces itself to LogDB on every connection (hello over WS and erlangd, X-LogDB-Client and the standard User-Agent header over HTTP), so the LogDB console shows which instance of a service is connected.

Tests

mix test

Unit tests need nothing running. The live erlangd test (test/live_erlangd_test.exs) needs a bus node started with a name and the same cookie:

LOGDB_LIVE_CONNECTION="logdb://<host>:4000?node=<bus node name>" \
LOGDB_LIVE_TOKEN=<service token> LOGDB_LIVE_CONSUMER_ID=<push consumer on <protocol>://**> \
elixir --sname live_test -S mix test --only live

Next Page → Changelog