postgresworker

README

Example

This is example for create Postgres Event Listener (CDC), inspired by Hasura Event Triggers

Create delivery handler

package workerhandler

import (
	"context"
	"encoding/json"
	
	"example.service/internal/modules/examplemodule/delivery/workerhandler"

	"github.com/golangid/candi/candishared"
	"github.com/golangid/candi/codebase/factory/types"
	"github.com/golangid/candi/tracer"
)

// PostgresListenerHandler struct
type PostgresListenerHandler struct {
	uc        usecase.Usecase
	validator interfaces.Validator
}

// NewPostgresListenerHandler constructor
func NewPostgresListenerHandler(uc usecase.Usecase, validator interfaces.Validator) *PostgresListenerHandler {
	return &PostgresListenerHandler{
		uc:        uc,
		validator: validator,
	}
}

// MountHandlers mount handler group
func (h *PostgresListenerHandler) MountHandlers(group *types.WorkerHandlerGroup) {
	group.Add("table-names", h.handleDataChange) // listen change data capture on table "table-names"
}

func (h *PostgresListenerHandler) handleDataChange(eventContext *candishared.EventContext) error {
	trace := tracer.StartTrace(eventContext.Context(), "DeliveryPostgresListener:HandleDataChange")
	defer trace.Finish()

	fmt.Printf("data change on table 'table-names' detected: %s\n", eventContext.Message())
	// call usecase
	return nil
}

Register in module

package examplemodule

import (
	"example.service/internal/modules/examplemodule/delivery/workerhandler"

	"github.com/golangid/candi/codebase/factory/dependency"
	"github.com/golangid/candi/codebase/factory/types"
	"github.com/golangid/candi/codebase/interfaces"
)

type Module struct {
	// ...another delivery handler
	workerHandlers map[types.Worker]interfaces.WorkerHandler
}

func NewModules(deps dependency.Dependency) *Module {
	return &Module{
		workerHandlers: map[types.Worker]interfaces.WorkerHandler{
			// ...another worker handler
			// ...
			types.PostgresListener: workerhandler.NewPostgresListenerHandler(usecaseUOW.User(), deps.GetValidator()),
		},
	}
}

// ...another method

Message in JSON Payload

{
	"event_id": "<md5-hash-unique-event-id>",
	"table": "<table-name>",
	"action": "<operation-name>", // INSERT, UPDATE, or DELETE
	"data": {
		"old": <old-column-values-object>,
		"new": <new-column-values-object>
	}
}

Documentation

Index

• func NewWorker(service factory.ServiceFactory, opts ...OptionFunc) factory.AppServerFactory
• type EventPayload
• type EventPayloadData
• type OptionFunc
  • func SetConsul(consul *candiutils.Consul) OptionFunc
  • func SetDebugMode(debugMode bool) OptionFunc
  • func SetLocker(locker candiutils.Locker) OptionFunc
  • func SetMaxGoroutines(maxGoroutines int) OptionFunc
  • func SetPostgresDSN(dsn string) OptionFunc

Functions

func NewWorker

func NewWorker(service factory.ServiceFactory, opts ...OptionFunc) factory.AppServerFactory

NewWorker create new postgres event listener

Types

type EventPayload

type EventPayload struct {
	EventID string           `json:"event_id"`
	Table   string           `json:"table"`
	Action  string           `json:"action"`
	Data    EventPayloadData `json:"data"`
}

EventPayload event model

type EventPayloadData

type EventPayloadData struct {
	Old interface{} `json:"old"`
	New interface{} `json:"new"`
}

EventPayloadData event data

type OptionFunc

type OptionFunc func(*option)

OptionFunc type

func SetConsul

func SetConsul(consul *candiutils.Consul) OptionFunc

SetConsul option func

func SetDebugMode

func SetDebugMode(debugMode bool) OptionFunc

SetDebugMode option func

func SetLocker

func SetLocker(locker candiutils.Locker) OptionFunc

SetLocker option func

func SetMaxGoroutines

func SetMaxGoroutines(maxGoroutines int) OptionFunc

SetMaxGoroutines option func

func SetPostgresDSN

func SetPostgresDSN(dsn string) OptionFunc

SetPostgresDSN option func