source

Documentation

Index

• Constants
• type BaseSource
  • func (b *BaseSource) Capture(cp cursor.Checkpoint) (changes chan Change, err error)
  • func (b *BaseSource) Commit(cp cursor.Checkpoint)
  • func (b *BaseSource) Error() error
  • func (b *BaseSource) Stop() error
• type CaptureFn
• type Change
• type FlushFn
• type PGXSource
  • func (p *PGXSource) Capture(cp cursor.Checkpoint) (changes chan Change, err error)
  • func (p *PGXSource) Commit(cp cursor.Checkpoint)
  • func (p *PGXSource) Requeue(cp cursor.Checkpoint, reason string)
  • func (p *PGXSource) TxCounter() uint64
• type PulsarConsumerSource
  • func (p *PulsarConsumerSource) Capture(cp cursor.Checkpoint) (changes chan Change, err error)
  • func (p *PulsarConsumerSource) Commit(cp cursor.Checkpoint)
  • func (p *PulsarConsumerSource) Requeue(cp cursor.Checkpoint, reason string)
• type PulsarReaderSource
  • func (p *PulsarReaderSource) Capture(cp cursor.Checkpoint) (changes chan Change, err error)
  • func (p *PulsarReaderSource) Commit(cp cursor.Checkpoint)
• type ReadFn
• type RequeueSource
• type Source

Constants

const (
	ReceiverQueueSize = 5000
	AckTrackerSize    = 1000
)

Types

type BaseSource

type BaseSource struct {
	ReadTimeout time.Duration
	// contains filtered or unexported fields
}

func (*BaseSource) Capture

func (b *BaseSource) Capture(cp cursor.Checkpoint) (changes chan Change, err error)

func (*BaseSource) Commit

func (b *BaseSource) Commit(cp cursor.Checkpoint)

func (*BaseSource) Error

func (b *BaseSource) Error() error

func (*BaseSource) Stop

func (b *BaseSource) Stop() error

type CaptureFn

type CaptureFn func(changes chan Change) error

type Change

type Change struct {
	Checkpoint cursor.Checkpoint
	Message    *pb.Message
}

type FlushFn

type FlushFn func()

type PGXSource

type PGXSource struct {
	BaseSource

	SetupConnStr      string
	ReplConnStr       string
	ReplSlot          string
	CreateSlot        bool
	CreatePublication bool
	StartLSN          string
	DecodePlugin      string
	// contains filtered or unexported fields
}

func (*PGXSource) Capture

func (p *PGXSource) Capture(cp cursor.Checkpoint) (changes chan Change, err error)

func (*PGXSource) Commit

func (p *PGXSource) Commit(cp cursor.Checkpoint)

func (*PGXSource) Requeue

func (p *PGXSource) Requeue(cp cursor.Checkpoint, reason string)

func (*PGXSource) TxCounter

func (p *PGXSource) TxCounter() uint64

type PulsarConsumerSource

type PulsarConsumerSource struct {
	BaseSource

	PulsarOption         pulsar.ClientOptions
	PulsarTopic          string
	PulsarSubscription   string
	PulsarReplicateState bool
	PulsarMaxReconnect   *uint
	// contains filtered or unexported fields
}

func (*PulsarConsumerSource) Capture

func (p *PulsarConsumerSource) Capture(cp cursor.Checkpoint) (changes chan Change, err error)

func (*PulsarConsumerSource) Commit

func (p *PulsarConsumerSource) Commit(cp cursor.Checkpoint)

func (*PulsarConsumerSource) Requeue

func (p *PulsarConsumerSource) Requeue(cp cursor.Checkpoint, reason string)

type PulsarReaderSource

type PulsarReaderSource struct {
	BaseSource

	PulsarOption pulsar.ClientOptions
	PulsarTopic  string
	// contains filtered or unexported fields
}

func (*PulsarReaderSource) Capture

func (p *PulsarReaderSource) Capture(cp cursor.Checkpoint) (changes chan Change, err error)

func (*PulsarReaderSource) Commit

func (p *PulsarReaderSource) Commit(cp cursor.Checkpoint)

type ReadFn

type ReadFn func(ctx context.Context) (Change, error)

type RequeueSource

type RequeueSource interface {
	Source
	Requeue(cp cursor.Checkpoint, reason string)
}

type Source

type Source interface {
	Capture(cp cursor.Checkpoint) (changes chan Change, err error)
	Commit(cp cursor.Checkpoint)
	Error() error
	Stop() error
}