CDC

Подписка на потоки Change Data Capture из источников db.cdc.postgres и db.cdc.sqlite. Позволяет перечислить настроенные источники, открыть поток и получать события изменений на уровне строк через канал. API нейтрален к драйверу: оба вида возвращают одну и ту же информацию об источнике и одни и те же события изменений, различаясь только возможностями, которые они публикуют.

Загрузка

local cdc = require("cdc")

list_sources

Перечислить настроенные источники CDC, которые вызывающей стороне разрешено видеть:

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

Источники, на которые у вызывающей стороны нет cdc.source, пропускаются, а не приводят к ошибке.

Возвращает: table, error

source

Получить один источник по имени (его ID записи):

local info, err = cdc.source("app:pg_cdc")
if info == nil then
    -- такого источника нет
end

Возвращает: table, error (информация об источнике либо nil, если не найден)

stream

Открыть поток изменений на источнике. Возвращает cdc.Stream, чей канал доставляет события изменений:

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

Неизвестные ключи опций отклоняются с errors.INVALID. Имена таблиц сопоставляются без учёта регистра как с полным именем отношения, так и с голым именем таблицы. Строки снимка фильтруются только по tables; ops применяется к живым изменениям.

Поток получает снимок, если либо opts.snapshot равно true, либо в записи источника задано поле snapshot; строки снимка приходят первыми с op = "snapshot", затем поток без разрыва продолжается живыми изменениями. opts.after зарезервирован для драйверов, возобновляющих работу с курсора — все поставляемые сегодня драйверы возвращают на него errors.INVALID ("cdc operation is not supported by this source"), включая db.cdc.postgres, когда он сообщает о capture_resume.

Фильтры лишь сужают доставку. Доступ к источнику даёт разрешение cdc.subscribe, а не фильтр.

Возвращает: Stream, error

Методы Stream

channel

Вернуть канал, принимающий события изменений. Первый вызов оформляет подписку на источник (уступает управление); последующие вызовы возвращают тот же канал. :receive() блокируется до прихода следующего изменения либо возвращает nil, когда поток завершается:

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

while true do
    local change = ch:receive()
    if change == nil then break end   -- поток закрыт

    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

Поток ленивый: сначала создайте его, затем вызовите channel() до генерации записей, которые он должен наблюдать. Это живое наблюдение, а не воспроизведение изменений, сделанных до подписки.

Когда источник завершает поток с ошибкой, канал доставляет значение ошибки перед закрытием. receive — псевдоним для channel.

close

Остановить подписку и освободить поток. Идемпотентно; также закрывается автоматически по завершении задачи. release — псевдоним для close.

stream:close()

Событие изменения

Каждое сообщение, полученное в канале, — это таблица изменения:

before и after — это отображения строки с ключами по именам колонок.

Информация об источнике

cdc.source и каждый элемент cdc.list_sources возвращают одну и ту же запись:

Ветвитесь по capabilities, а не по kind:

local info = cdc.source("app:changes")
if not info.capabilities.before_images then
    -- before не гарантирует полный образ строки; храните собственное последнее известное состояние
end

Семантику полей см. в источниках CDC.

Разрешения

Запрещённое действие возвращает errors.PERMISSION_DENIED.

Ошибки

О работе с ошибками см. Обработка ошибок.

См. также

• Change Data Capture - Настройка источников и возможности
• Channel - Семантика каналов
• Database - Сервисы SQL-баз данных