LogDB (logdb_client v0.3.0)

Copy Markdown View Source

Работа с шиной из приложения. name — имя клиента (name: в его опциях).

Ответы одной формы для любого транспорта:

  • команды — {:ok, результат} или {:error, %LogDB.Error{}};
  • subscribe/2,3 и ping/1 — :ok или {:error, %LogDB.Error{}}.

Клиент не запущен или соединение потеряно — {:error, %LogDB.Error{code: :disconnected}}: вызывающий процесс не падает.

Summary

Functions

Подтвердить обработку: шина больше не доставит эти события.

Дождаться, пока клиент подключится: :ok или {:error, %LogDB.Error{code: :timeout}}. Клиент подключается сам и асинхронно — приложение не ждёт шину при старте; это для тех, кому нужно (тесты, скрипты, миграции).

Отказ от событий — шина доставит их снова. delay_ms: — не раньше чем через столько миллисекунд (нужна шина с поддержкой задержки; иначе — сразу).

Шина отвечает и принимает токен клиента.

«Ещё обрабатываю»: шина заново отсчитает ack_wait, чтобы не доставить событие повторно, пока долгая обработка не закончилась. Вызывать из handle_event/4 периодически.

Опубликовать событие. stream: — стрим (по умолчанию "default"), meta: — метаданные, id: — id события для дедупликации.

Опубликовать пачку событий: %{"type", "payload", "stream", "meta"?}.

Каким транспортом подключён клиент.

Подписаться на консьюмер по адресу "stream/имя" — имя без префикса сервиса, сервис шина знает по токену. Нужен LogDB с клиентским API /v1.

Подписаться прежней парой стрим + id консьюмера.

Окончательный отказ: шина больше никогда не доставит эти события — для «ядовитых», которые не обработать сколько ни пытайся.

Types

ids()

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

name()

@type name() :: atom()

result()

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

status()

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

Functions

ack(name, ids)

@spec ack(name(), ids()) :: result()

Подтвердить обработку: шина больше не доставит эти события.

await_connected(name, timeout \\ 5000)

@spec await_connected(name(), timeout()) :: status()

Дождаться, пока клиент подключится: :ok или {:error, %LogDB.Error{code: :timeout}}. Клиент подключается сам и асинхронно — приложение не ждёт шину при старте; это для тех, кому нужно (тесты, скрипты, миграции).

nack(name, ids, opts \\ [])

@spec nack(name(), ids(), keyword()) :: result()

Отказ от событий — шина доставит их снова. delay_ms: — не раньше чем через столько миллисекунд (нужна шина с поддержкой задержки; иначе — сразу).

ping(name)

@spec ping(name()) :: status()

Шина отвечает и принимает токен клиента.

progress(name, ids)

@spec progress(name(), ids()) :: result()

«Ещё обрабатываю»: шина заново отсчитает ack_wait, чтобы не доставить событие повторно, пока долгая обработка не закончилась. Вызывать из handle_event/4 периодически.

publish(name, type, payload, opts \\ [])

@spec publish(name(), String.t(), term(), keyword()) :: result()

Опубликовать событие. stream: — стрим (по умолчанию "default"), meta: — метаданные, id: — id события для дедупликации.

publish_batch(name, events)

@spec publish_batch(name(), [map()]) :: result()

Опубликовать пачку событий: %{"type", "payload", "stream", "meta"?}.

server_info(name)

@spec server_info(name()) :: {:ok, map()} | {:error, LogDB.Error.t()}

Каким транспортом подключён клиент.

subscribe(name, consumer)

@spec subscribe(name(), String.t()) :: status()

Подписаться на консьюмер по адресу "stream/имя" — имя без префикса сервиса, сервис шина знает по токену. Нужен LogDB с клиентским API /v1.

subscribe(name, stream, consumer_id)

@spec subscribe(name(), String.t(), String.t()) :: status()

Подписаться прежней парой стрим + id консьюмера.

term(name, ids, reason \\ nil)

@spec term(name(), ids(), String.t() | nil) :: result()

Окончательный отказ: шина больше никогда не доставит эти события — для «ядовитых», которые не обработать сколько ни пытайся.