go-pq-cdc-rabbitmq

February 20, 2026 · View on GitHub

RabbitMQ connector for go-pq-cdc.

go-pq-cdc-rabbitmq streams PostgreSQL logical replication events to RabbitMQ with batching and publisher confirms. It follows the same architecture used by go-pq-cdc-kafka and go-pq-cdc-elasticsearch.

Features

PostgreSQL CDC via go-pq-cdc
RabbitMQ topic exchange publishing
Routing key mapping (schema.table.operation)
Batch publishing by size, bytes, and ticker
Publisher confirms before PostgreSQL ACK
Connection recovery with exponential backoff
Snapshot mode support (initial and snapshot_only)
Prometheus metrics

Install

go get github.com/ezeql/go-pq-cdc-rabbitmq

Usage

package main

import (
    "context"
    "encoding/json"
    "log/slog"
    "time"

    cdc "github.com/ezeql/go-pq-cdc-rabbitmq"
    "github.com/ezeql/go-pq-cdc-rabbitmq/config"
    "github.com/ezeql/go-pq-cdc-rabbitmq/rabbitmq"
    cdcconfig "github.com/Trendyol/go-pq-cdc/config"
    "github.com/Trendyol/go-pq-cdc/pq/publication"
    "github.com/Trendyol/go-pq-cdc/pq/slot"
)

func main() {
    cfg := config.Connector{
        CDC: cdcconfig.Config{
            Host:     "127.0.0.1",
            Username: "cdc_user",
            Password: "cdc_pass",
            Database: "cdc_db",
            Publication: publication.Config{
                CreateIfNotExists: true,
                Name:              "cdc_publication",
                Operations: publication.Operations{
                    publication.OperationInsert,
                    publication.OperationUpdate,
                    publication.OperationDelete,
                },
                Tables: publication.Tables{
                    {Name: "users", ReplicaIdentity: publication.ReplicaIdentityFull},
                },
            },
            Slot: slot.Config{
                CreateIfNotExists: true,
                Name:              "cdc_slot",
            },
        },
        RabbitMQ: config.RabbitMQ{
            URL: "amqp://guest:guest@localhost:5672/",
            Exchange: config.ExchangeConfig{
                Name:    "cdc.events",
                Type:    "topic",
                Durable: true,
            },
            TableRoutingKeyMapping: map[string]string{
                "public.users": "public.users",
            },
            PublisherBatchTickerDuration: 200 * time.Millisecond,
        },
    }

    connector, err := cdc.NewConnector(context.Background(), cfg, func(msg *cdc.Message) []rabbitmq.PublishMessage {
        if !msg.Type.IsInsert() && !msg.Type.IsUpdate() && !msg.Type.IsSnapshot() && !msg.Type.IsDelete() {
            return nil
        }
        data := msg.NewData
        if msg.Type.IsDelete() {
            data = msg.OldData
        }
        data["operation"] = string(msg.Type)
        payload, _ := json.Marshal(data)
        }
        return []rabbitmq.PublishMessage{{Body: payload, ContentType: "application/json"}}
    })
    if err != nil {
        slog.Error("new connector", "error", err)
        return
    }
    defer connector.Close()
    connector.Start(context.Background())
}