LogDB.Client.Transport behaviour (logdb_client v0.3.0)

Copy Markdown View Source

Как клиент говорит с шиной. Реализации: Erlangd (дистрибуция Erlang), WebSocket (доставка и подтверждения), HTTP (публикация и подтверждения), Hybrid (WS для доставки, HTTP для остального).

Ответы — одной формы: команды `{:ok, результат}}`,
subscribe и ping — `:ok}`. Чего транспорт не

умеет, он не делает вид, что сделал, — отвечает ошибкой с кодом :not_supported. Фасад (LogDB) приводит к этой форме и то, что ответил транспорт не по контракту, но транспорт должен держать её сам.

Входящие события транспорт отдаёт консьюмеру сам — LogDB.Client.Worker.deliver/5 (имя — opts[:name]) — или шлёт процессу- владельцу (Connection) {:logdb_event, type, payload, meta}, если формат задаёт не он (erlangd: события шлёт шина). В каком виде payload — :bytes или :json — говорит wire/0. О состоянии соединения — владельцу: {:transport_status, :connected | :ready | :disconnected}.

Противодавление: push-доставку (WS, erlangd) ограничивает шина — max_ack_pending консьюмера (неподтверждённых не больше стольких). Придерживать чтение сокета нельзя: ответы на подтверждения идут по нему же. Кто забирает события сам (HTTP long-poll), берёт следующую пачку, когда консьюмер разобрал эту (LogDB.Client.Worker.drained/2).

Summary

Types

ids()
opts()
result()
state()
status()

Callbacks

ack(state, ids)
connect(opts)
disconnect(state)
nack(state, ids, list)
ping(state)
progress(state, ids)
publish(state, type, payload, opts)
publish_batch(state, events)
server_info(state)
subscribe(state, consumer)

Подписка по адресу "stream/имя" (сервис шина знает по токену; LogDB с /v1).

subscribe(state, stream, consumer_id)
term(state, ids, reason)
wire()

В каком виде транспорт отдаёт payload событий; по умолчанию :json.

Types

ids()

@type ids() :: [String.t()]

opts()

@type opts() :: keyword()

result()

@type result() :: {:ok, term()} | {:error, LogDB.Error.t()}

state()

@type state() :: term()

status()

@type status() :: :ok | {:error, LogDB.Error.t()}

Callbacks

ack(state, ids)

@callback ack(state(), ids()) :: result()

connect(opts)

@callback connect(opts()) :: {:ok, state()} | {:error, term()}

disconnect(state)

@callback disconnect(state()) :: :ok | term()

nack(state, ids, list)

@callback nack(state(), ids(), [{:delay_ms, non_neg_integer()}]) :: result()

ping(state)

@callback ping(state()) :: status()

progress(state, ids)

@callback progress(state(), ids()) :: result()

publish(state, type, payload, opts)

@callback publish(state(), type :: String.t(), payload :: term(), opts()) :: result()

publish_batch(state, events)

@callback publish_batch(state(), events :: [map()]) :: result()

server_info(state)

@callback server_info(state()) :: map()

subscribe(state, consumer)

@callback subscribe(state(), consumer :: String.t()) :: status()

Подписка по адресу "stream/имя" (сервис шина знает по токену; LogDB с /v1).

subscribe(state, stream, consumer_id)

@callback subscribe(state(), stream :: String.t(), consumer_id :: String.t()) :: status()

term(state, ids, reason)

@callback term(state(), ids(), reason :: String.t() | nil) :: result()

wire()

(optional) 
@callback wire() :: :bytes | :json

В каком виде транспорт отдаёт payload событий; по умолчанию :json.