phylax

package module

Go to main page

Versions in this module

v0

v0.3.3

Sep 22, 2026

Changes in this version

type Config

+ ChangeBufferSize int

v0.3.2

Aug 18, 2026

v0.3.1

Aug 15, 2026

Changes in this version

type MetricsSnapshot

+ OutboxDelivered int64

+ OutboxFailed int64

+ OutboxInflight int64

type OutboxConsumer

+ func (oc *OutboxConsumer) Stats() (delivered, inflight, failed int64)

type ReplicationStream

+ func (s *ReplicationStream) OutboxStats() (delivered, inflight, failed int64)

v0.3.0

Aug 15, 2026

Changes in this version

type CDC

+ func (c *CDC) OnOutboxDelivery(fn func(context.Context, *OutboxRow) error)

type Config

+ OutboxTable string

+ type DeliveryFunc func(ctx context.Context, row *OutboxRow) error

+ type OutboxConsumer struct

+ func NewOutboxConsumer(db *pgx.Conn, deliver DeliveryFunc, tableName string) *OutboxConsumer

+ func (oc *OutboxConsumer) Handle(ctx context.Context, c *Change) bool

+ type OutboxRow struct

+ ID int64

+ Payload map[string]any

+ Topic string

+ func ToOutboxRow(c *Change, tableName string) (row *OutboxRow, ok bool, err error)

v0.2.0

Aug 7, 2026

v0.1.0

Aug 7, 2026

Changes in this version

+ func EnsurePublication(ctx context.Context, adminConn *pgx.Conn, publicationName string, ...) (bool, error)

+ func EnsureReplicationSlot(ctx context.Context, replConn *pgconn.PgConn, adminConn *pgx.Conn, ...) (bool, error)

+ func IdentifySystem(ctx context.Context, conn *pgconn.PgConn) (pglogrepl.IdentifySystemResult, error)

+ func OpenAdminConnection(ctx context.Context, adminURL string) (*pgx.Conn, error)

+ func OpenReplicationConnection(ctx context.Context, databaseURL string) (*pgconn.PgConn, error)

+ func PublicationExists(ctx context.Context, conn *pgx.Conn, publicationName string) (bool, error)

+ func SlotConfirmedFlushLSN(ctx context.Context, conn *pgx.Conn, slotName string) (pglogrepl.LSN, bool, error)

+ func SlotExists(ctx context.Context, conn *pgx.Conn, slotName string) (bool, error)

+ type Broadcaster struct

+ func NewBroadcaster() *Broadcaster

+ func (b *Broadcaster) ChangesDropped() int64

+ func (b *Broadcaster) Publish(change *Change)

+ func (b *Broadcaster) Subscribe(id string, bufferSize int) chan *Change

+ func (b *Broadcaster) SubscriberCount() int

+ func (b *Broadcaster) Unsubscribe(id string)

+ type CDC struct

+ func New(cfg Config) (*CDC, error)

+ func (c *CDC) Broadcaster() *Broadcaster

+ func (c *CDC) MetricsSnapshot() MetricsSnapshot

+ func (c *CDC) OnChange(fn func(*Change))

+ func (c *CDC) Server() *Server

+ func (c *CDC) Start(ctx context.Context) error

+ type Change struct

+ NewRow map[string]any

+ OldRow map[string]any

+ Operation string

+ Table string

+ func Decode(walData []byte, relations map[uint32]*pglogrepl.RelationMessage, ...) (*Change, error)

+ type ChangeHandler func(*Change) error

+ type ClientConfig struct

+ AdminURL string

+ DatabaseURL string

+ HeartbeatInterval time.Duration

+ PublicationName string

+ SlotName string

+ Tables []string

+ func DefaultClientConfig() ClientConfig

+ type Config struct

+ DSN string

+ PublicationName string

+ SlotName string

+ Tables []string

+ type Metrics struct

+ ChangesProcessed atomic.Int64

+ type MetricsProvider interface

+ MetricsSnapshot func() MetricsSnapshot

+ type MetricsSnapshot struct

+ ChangesDropped int64

+ ChangesProcessed int64

+ ReplicationLag uint64

+ Subscribers int

+ type ReplicationStream struct

+ func NewReplicationStream(ctx context.Context, conn *pgconn.PgConn, cfg ClientConfig, ...) (*ReplicationStream, error)

+ func (s *ReplicationStream) Broadcaster() *Broadcaster

+ func (s *ReplicationStream) ReplicationLag() uint64

+ func (s *ReplicationStream) Run(ctx context.Context) error

+ type Server struct

+ func NewServer(broadcaster *Broadcaster, metrics MetricsProvider) *Server

+ func (s *Server) Handler() http.Handler

+ func (s *Server) ListenAndServe(addr string) error

+ func (s *Server) NewMetricsHandler(w http.ResponseWriter, r *http.Request)

+ func (s *Server) NewSSEHandler(w http.ResponseWriter, r *http.Request)

+ func (s *Server) Serve(ln net.Listener) error

+ func (s *Server) Shutdown(ctx context.Context) error

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL