PgOutput2Json

.NET library · PostgreSQL logical replication

Every committed row change, as JSON.

PgOutput2Json reads PostgreSQL's replication stream and hands each insert, update and delete to your .NET code, or publishes it to Kafka, RabbitMQ, Redis and other outputs, shortly after the transaction commits.

Get startedSource on GitHub

dotnet add package PgOutput2Json

PostgreSQL, committed

INSERT INTO orders (id, status, total)
VALUES (42, 'new', 19.90);

Your handler, or a broker

{
  "c": "I",change type
  "w": 2485645760,transaction final LSN
  "n": 1,message number
  "r": { "id": 42, "status": "new", "total": 19.90 }
}

Outputs

One stream in, eleven places to send it

Handle the messages in your own code, or add a package and let it publish them. Each output has its own page in the docs.

How it behaves

Built for the cases where a change must not be lost

  1. Nothing to install in PostgreSQL

    It reads pgoutput, the logical decoding plugin that ships with PostgreSQL 10 and later. You need wal_level = logical, a publication and a replication user.

    Getting started

  2. At-least-once, with a key to deduplicate

    A slot replays unconfirmed changes after a crash or a reconnect, and they are published again. Every message carries a key, the transaction LSN and a counter, so consumers can skip what they already have.

    Delivery and deduplication

  3. Confirmed only after the output has it

    Changes are published in batches. A batch is confirmed to PostgreSQL after the output acknowledged it, so a failure replays the batch.

    Configuration

  4. Changes survive a restart

    With a permanent replication slot, PostgreSQL keeps the changes made while your worker is stopped and sends them when it connects again.

    Replication slots

  5. Stable routing for partitioned outputs

    Choose the columns to partition by. Inserts, updates and deletes of a row go to the same Kafka partition, super stream partition or Event Hubs partition.

    Message key

  6. Compact mode

    Rows as arrays of values, with the column names sent once per table. The SQLite, MongoDB and DynamoDB outputs use it to keep a copy of your tables.

    Change events

Quick start

A publication, a package, a handler

Set wal_level = logical in postgresql.conf, then:

  1. Create a replication user and a publication for the tables you want.
  2. Add the PgOutput2Json package to a .NET worker service.
  3. Build the listener with a handler, or with one of the outputs, and start it.
The full walkthrough
CREATE USER pgoutput2json WITH
    PASSWORD '_your_password_here_'
    REPLICATION;

CREATE PUBLICATION my_publication
    FOR TABLE my_table1, my_table2
    WITH (publish = 'insert, update, delete');
using var pgOutput2Json = PgOutput2JsonBuilder.Create()
    .WithLoggerFactory(_loggerFactory)
    .WithPgConnectionString("server=localhost;database=my_database;username=pgoutput2json;password=_your_password_here_")
    .WithPgPublications("my_publication")
    .WithMessageHandler((json, table, key, partition) =>
    {
        Console.WriteLine($"{table}: {json}");
        return Task.FromResult(true);
    })
    .Build();

await pgOutput2Json.StartAsync(stoppingToken);

Use it for

Reacting to data, not polling for it

  • Background tasks triggered by inserts and updates
  • Application caches updated with low latency
  • Asynchronous logging to a file or another database
  • Real-time aggregation
  • Lightweight ETL into a data warehouse

Start with one table

The library is under active development and usable for testing and early integration. It targets .NET 8, 9 and 10.

Read the docsStar it on GitHub