ebifriedrice.postlog.consumer — net.clojars.ebifriedrice/postlog 0.1.0

The replication consumer loop: open a stream, assemble decoded messages into transactions, deliver them to handlers, and ack the flushed LSN only after delivery succeeds.

Delivery is ordered and at-least-once. The slot's confirmed position advances only at transaction boundaries after the handler returns, so a handler failure or a crash redelivers from the last acked transaction; handlers must tolerate (or dedupe) duplicates.

(def c (consumer/start!
        {:db           {:host "localhost" :database "app"
                        :user "repl" :password "..."}
         :slot-name    "my_slot"
         :publication-name "my_pub"
         :tables       ["public.users"]
         :on-change    (fn [change] ...)       ; per change
         ;; and/or
         :on-transaction (fn [tx] ...)}))      ; per commit
(consumer/stop! c)

Connections (exactly one source is required):

:db spec map; postlog opens and owns both the replication and admin connections, closing them on stop. :conn-fn (fn [] Connection); called for every (re)connect and must return a fresh, dedicated, un-pooled replication connection (replication=database) which postlog never closes. A bare connection instance is not accepted on purpose: a replication connection that dies mid-stream cannot be revived -- resuming the slot means a fresh connection, which only a factory can produce. :admin-conn optional Connection for slot/publication bootstrap; never closed by postlog. Without :db or :admin-conn, the slot and publication must already exist.

:stream-fn test escape hatch (fn [] LogicalStream); bypasses connection handling entirely.

:binary? stream values in the binary wire format (default true); false selects text. decode/decode-value decodes both, so the choice matters mainly for custom value-decoders and for unknown types, which pass through as raw bytes (binary) or raw text (text).

Handlers run on the consumer's single thread, one at a time -- that is what keeps ordering and provides backpressure. Handler exceptions retry with exponential backoff (see :retry) and fail the consumer when exhausted; a failed consumer acks nothing past the failure. Connection loss reconnects with the same backoff, and the slot resumes from its confirmed position.

Transactions with no delivered changes (nothing published, or only filtered operations) still ack -- the position must advance -- but do not invoke :on-transaction.

start! clj

(start! opts)

Start a consumer for opts (see the namespace docs for the full option reference). Returns a handle:

{:stop! (fn []) :await (fn [timeout-ms?]) :status (fn [])}

stop! is idempotent and asynchronous; await blocks until the consumer stops (or the timeout passes, returning nil) and yields the final status map {:state :stopped/:failed, :error?, :txns, :confirmed-lsn, ...}. stop! then await is the graceful-shutdown pair.

stop! clj

(stop! consumer)

Request a graceful stop: the loop finishes its current poll cycle, closes its stream and any owned connections, and acks nothing past what it already delivered. Idempotent.