forwarder

Documentation

Index

• Constants
• Variables
• type DBForwarder
  • func New(db db.DB, bus bus.Bus, options *Options) (*DBForwarder, error)
  • func (fw *DBForwarder) Start(ctx context.Context) error
  • func (fw *DBForwarder) Stop() error
• type DebeziumForwarder
  • func NewDebeziumForwarder(db db.DB, bus bus.Bus, options *DebeziumOptions) (*DebeziumForwarder, error)
  • func (fw *DebeziumForwarder) Start(ctx context.Context) error
  • func (fw *DebeziumForwarder) Stop() error
• type DebeziumMessage
• type DebeziumOptions
• type Forwarder
• type Options

Constants

const DefaultPollingInterval = 100 * time.Millisecond

Variables

var DefaultOptions = &Options{
	PollingInterval: DefaultPollingInterval,
	Serializer:      serialization.NewProtobufSerializer(),
	OutboxTableName: "event_outbox",
}

Types

type DBForwarder

type DBForwarder struct {
	// contains filtered or unexported fields
}

func New

func New(db db.DB, bus bus.Bus, options *Options) (*DBForwarder, error)

func (*DBForwarder) Start

func (fw *DBForwarder) Start(ctx context.Context) error

func (*DBForwarder) Stop

func (fw *DBForwarder) Stop() error

type DebeziumForwarder

type DebeziumForwarder struct {
	// contains filtered or unexported fields
}

func NewDebeziumForwarder

func NewDebeziumForwarder(db db.DB, bus bus.Bus, options *DebeziumOptions) (*DebeziumForwarder, error)

func (*DebeziumForwarder) Start

func (fw *DebeziumForwarder) Start(ctx context.Context) error

func (*DebeziumForwarder) Stop

func (fw *DebeziumForwarder) Stop() error

type DebeziumMessage

type DebeziumMessage struct {
	Payload struct {
		Before interface{} `json:"before"`
		After  struct {
			ID        string `json:"id"`
			Topic     string `json:"topic"`
			Payload   []byte `json:"payload"`
			CreatedAt int64  `json:"created_at"`
		} `json:"after"`
		Source struct {
			Version   string      `json:"version"`
			Connector string      `json:"connector"`
			Name      string      `json:"name"`
			TsMs      int64       `json:"ts_ms"`
			Snapshot  string      `json:"snapshot"`
			Db        string      `json:"db"`
			Sequence  interface{} `json:"sequence"`
			TsUs      int64       `json:"ts_us"`
			TsNs      int64       `json:"ts_ns"`
			Table     string      `json:"table"`
			ServerID  int         `json:"server_id"`
			Gtid      interface{} `json:"gtid"`
			File      string      `json:"file"`
			Pos       int         `json:"pos"`
			Row       int         `json:"row"`
			Thread    int         `json:"thread"`
			Query     interface{} `json:"query"`
		} `json:"source"`
		Transaction interface{} `json:"transaction"`
		Op          string      `json:"op"`
		TsMs        int64       `json:"ts_ms"`
		TsUs        int64       `json:"ts_us"`
		TsNs        int64       `json:"ts_ns"`
	} `json:"payload"`
}

type DebeziumOptions

type DebeziumOptions struct {
	Serializer      serialization.Serializer
	DebeziumStream  string
	DebeziumSubject string
	SubscriberName  string
	Logger          *zap.Logger
}

type Forwarder

type Forwarder interface {
	Stop() error
	Start(ctx context.Context) error
}

type Options

type Options struct {
	PollingInterval time.Duration
	Serializer      serialization.Serializer
	OutboxTableName string
	Logger          *zap.Logger
}