Open replication connections and stream pgoutput messages.
pgjdbc exposes the walsender protocol on a connection opened with
replication=database: unwrap to org.postgresql.PGConnection,
build a logical stream against a slot, and receive the raw pgoutput
payload of each WAL record. The driver owns the protocol mechanics
this library is happy not to reimplement: XLogData framing,
keepalive replies, and periodic standby status updates reporting
our flushed LSN.
start! returns anything implementing LogicalStream -- by
default a thin adapter over pgjdbc's stream, but the consumer loop
only talks to the protocol, so tests drive it with fakes.
Open replication connections and stream pgoutput messages. pgjdbc exposes the walsender protocol on a connection opened with ``replication=database``: unwrap to ``org.postgresql.PGConnection``, build a logical stream against a slot, and receive the raw pgoutput payload of each WAL record. The driver owns the protocol mechanics this library is happy not to reimplement: XLogData framing, keepalive replies, and periodic standby status updates reporting our flushed LSN. [[start!]] returns anything implementing [[LogicalStream]] -- by default a thin adapter over pgjdbc's stream, but the consumer loop only talks to the protocol, so tests drive it with fakes.
(connect spec)Open a dedicated replication connection (replication=database)
from a spec map or JDBC URL string. The caller owns closing it.
Open a dedicated replication connection (``replication=database``) from a spec map or JDBC URL string. The caller owns closing it.
One logical replication stream.
One logical replication stream.
(ack! stream lsn)Report lsn as the flushed-and-applied position and push a
status update to the server immediately. Call only after the
work belonging to everything up to lsn has succeeded.
Report `lsn` as the flushed-and-applied position and push a status update to the server immediately. Call only after the work belonging to everything up to `lsn` has succeeded.
(close! stream)End the stream. The slot keeps its last confirmed position.
End the stream. The slot keeps its last confirmed position.
(read-pending stream)The next pending pgoutput message's bytes, or nil when nothing is available right now. Never blocks.
The next pending pgoutput message's bytes, or nil when nothing is available right now. Never blocks.
(start! conn
{:keys [slot-name publications proto-version status-interval-ms binary?
messages?]})Start a logical stream on an open replication conn.
Opts: :slot-name (required) the slot to consume :publications (required) publication name(s), string or coll :proto-version pgoutput protocol version, default 2 :status-interval-ms standby status cadence, default 5000 :binary? request binary tuple values (default true; pass false for text values — see ebifriedrice.postlog.decode for both layouts) :messages? forward pg_logical_emit_message payloads
Binary is the default: scalars arrive in their typsend wire layouts instead of text renderings, floats keep their bit patterns, and timestamps skip a text round-trip. Starts from the slot's own confirmed position; the server decides where that is, which is what makes resume-on-restart work.
Start a logical stream on an open replication `conn`.
Opts:
:slot-name (required) the slot to consume
:publications (required) publication name(s), string or coll
:proto-version pgoutput protocol version, default 2
:status-interval-ms standby status cadence, default 5000
:binary? request binary tuple values (default true; pass
false for text values — see
ebifriedrice.postlog.decode for both layouts)
:messages? forward pg_logical_emit_message payloads
Binary is the default: scalars arrive in their typsend wire
layouts instead of text renderings, floats keep their bit patterns,
and timestamps skip a text round-trip. Starts from the slot's own
confirmed position; the server decides where that is, which is what
makes resume-on-restart work.cljdoc builds & hosts documentation for Clojure/Script libraries
| Ctrl+k | Jump to recent docs |
| ← | Move to previous article |
| → | Move to next article |
| Ctrl+/ | Jump to the search field |