CDC

Abonnieren Sie Change-Data-Capture-Streams aus db.cdc.postgres- und db.cdc.sqlite-Quellen. Konfigurierte Quellen auflisten, einen Stream öffnen und zeilenbezogene Änderungsereignisse über einen Channel empfangen. Die API ist treiberneutral: Beide Kinds liefern dieselben Quellinformationen und dieselben Änderungsereignisse und unterscheiden sich nur in den Capabilities, die sie veröffentlichen.

Laden

local cdc = require("cdc")

list_sources

Listet die konfigurierten CDC-Quellen auf, die der Aufrufer sehen darf:

local sources, err = cdc.list_sources()
for _, s in ipairs(sources) do
    print(s.id, s.kind, s.state, s.capabilities.before_images)
end

Quellen, für die dem Aufrufer cdc.source fehlt, werden ausgelassen statt als Fehler gemeldet.

Rückgabe: table, error

source

Eine einzelne Quelle per Name (ihrer Entry-ID) abrufen:

local info, err = cdc.source("app:pg_cdc")
if info == nil then
    -- keine solche Quelle
end

Rückgabe: table, error (Quellinformationen oder nil, wenn nicht gefunden)

stream

Einen Änderungs-Stream auf einer Quelle öffnen. Gibt einen cdc.Stream zurück, dessen Channel Änderungsereignisse liefert:

local stream, err = cdc.stream("app:pg_cdc", {
    tables = { "public.users", "public.orders" },
    ops    = { "insert", "update" },
    buffer = 128,
})

Unbekannte Optionsschlüssel werden mit errors.INVALID abgelehnt. Tabellennamen werden ohne Beachtung der Groß-/Kleinschreibung sowohl gegen die qualifizierte Relation als auch gegen den bloßen Tabellennamen gematcht. Snapshot-Zeilen werden nur über tables gefiltert; ops gilt für Live-Änderungen.

Ein Stream erhält einen Snapshot, wenn entweder opts.snapshot wahr ist oder das Feld snapshot des Quelleintrags gesetzt ist; Snapshot-Zeilen treffen zuerst mit op = "snapshot" ein, danach setzt der Stream ohne Lücke mit Live-Änderungen fort. opts.after ist für Treiber reserviert, die ab einem Cursor fortsetzen — jeder heute ausgelieferte Treiber gibt dafür errors.INVALID zurück ("cdc operation is not supported by this source"), auch db.cdc.postgres, wenn er capture_resume meldet.

Filter schränken nur die Zustellung ein. Zugriff auf eine Quelle wird durch die Berechtigung cdc.subscribe gewährt, niemals durch einen Filter.

Rückgabe: Stream, error

Stream-Methoden

channel

Gibt den Channel zurück, der Änderungsereignisse empfängt. Der erste Aufruf abonniert die Quelle (gibt ab); nachfolgende Aufrufe geben denselben Channel zurück. :receive() blockiert, bis die nächste Änderung eintrifft, oder gibt nil zurück, wenn der Stream endet:

local stream = cdc.stream("app:pg_cdc")
local ch = stream:channel()

while true do
    local change = ch:receive()
    if change == nil then break end   -- Stream geschlossen

    if change.op == "snapshot" then
        seed_row(change.table, change.after)
    elseif change.op == "insert" then
        handle_new_user(change.table, change.after)
    elseif change.op == "update" then
        handle_update(change.table, change.before, change.after)
    elseif change.op == "delete" then
        handle_delete(change.table, change.before)
    end
end

Der Stream ist lazy: Erstellen Sie ihn und rufen Sie dann channel() auf, bevor Sie die Schreibvorgänge erzeugen, die er beobachten soll. Das ist Live-Beobachtung, kein Nachspielen von Änderungen, die vor dem Abonnement erfolgt sind.

Wenn eine Quelle einen Stream mit einem Fehler beendet, liefert der Channel einen Fehlerwert, bevor er schließt. receive ist ein Alias für channel.

close

Beendet das Abonnement und gibt den Stream frei. Idempotent; wird auch am Ende des Task-Scopes automatisch geschlossen. release ist ein Alias für close.

stream:close()

Änderungsereignis

Jede auf dem Channel empfangene Nachricht ist eine Änderungstabelle:

before und after sind Zeilen-Maps mit Spaltennamen als Schlüssel.

Quellinformationen

cdc.source und jeder Eintrag von cdc.list_sources geben denselben Datensatz zurück:

Verzweigen Sie über capabilities, nicht über kind:

local info = cdc.source("app:changes")
if not info.capabilities.before_images then
    -- before ist kein garantiert vollständiges Zeilenabbild; halten Sie Ihren eigenen zuletzt bekannten Zustand
end

Siehe CDC-Quellen für die Feldsemantik.

Berechtigungen

Eine verweigerte Aktion gibt errors.PERMISSION_DENIED zurück.

Fehler

Siehe Fehlerbehandlung für den Umgang mit Fehlern.

Siehe auch

• Change Data Capture - Quellkonfiguration und Capabilities
• Channel - Channel-Semantik
• Datenbank - SQL-Datenbankdienste