stream

Oct 9, 2024

Changes in this version

+ var ErrUnexpectedMessageStatus = cerrors.New("unexpected message status")

+ func LoggerWithComponent(logger log.CtxLogger, v Node) log.CtxLogger

+ func LoggerWithNodeID(logger log.CtxLogger, n Node) log.CtxLogger

+ func SetLogger(n Node, logger log.CtxLogger, ...)

+ type AckHandler func(*Message) error

+ type ControlMessageType string

+ const ControlMessageStopSourceNode

+ type DLQHandler interface

+ Close func(context.Context) error

+ Open func(context.Context) error

+ Write func(context.Context, opencdc.Record) error

+ type DLQHandlerNode struct

+ Handler DLQHandler

+ Histogram metrics.RecordBytesHistogram

+ Name string

+ Timer metrics.Timer

+ WindowNackThreshold int

+ WindowSize int

+ func (n *DLQHandlerNode) Ack(msg *Message)

+ func (n *DLQHandlerNode) Add(delta int)

+ func (n *DLQHandlerNode) Done()

+ func (n *DLQHandlerNode) ForceStop(ctx context.Context)

+ func (n *DLQHandlerNode) ID() string

+ func (n *DLQHandlerNode) Nack(msg *Message, nackMetadata NackMetadata) error

+ func (n *DLQHandlerNode) Run(ctx context.Context) error

+ func (n *DLQHandlerNode) SetLogger(logger log.CtxLogger)

+ type Destination interface

+ Ack func(context.Context) ([]connector.DestinationAck, error)

+ Errors func() <-chan error

+ ID func() string

+ Open func(context.Context) error

+ Stop func(context.Context, opencdc.Position) error

+ Teardown func(context.Context) error

+ Write func(context.Context, []opencdc.Record) error

+ type DestinationAckerNode struct

+ Destination Destination

+ Name string

+ func (n *DestinationAckerNode) ForceStop(ctx context.Context)

+ func (n *DestinationAckerNode) ID() string

+ func (n *DestinationAckerNode) Run(ctx context.Context) (err error)

+ func (n *DestinationAckerNode) SetLogger(logger log.CtxLogger)

+ func (n *DestinationAckerNode) Sub(in <-chan *Message)

+ type DestinationNode struct

+ ConnectorTimer metrics.Timer

+ Destination Destination

+ Name string

+ func (n *DestinationNode) ForceStop(ctx context.Context)

+ func (n *DestinationNode) ID() string

+ func (n *DestinationNode) Pub() <-chan *Message

+ func (n *DestinationNode) Run(ctx context.Context) (err error)

+ func (n *DestinationNode) SetLogger(logger log.CtxLogger)

+ func (n *DestinationNode) Sub(in <-chan *Message)

+ type FaninNode struct

+ Name string

+ func (n *FaninNode) ID() string

+ func (n *FaninNode) Pub() <-chan *Message

+ func (n *FaninNode) Run(ctx context.Context) error

+ func (n *FaninNode) Sub(in <-chan *Message)

+ type FanoutNode struct

+ Name string

+ func (n *FanoutNode) ID() string

+ func (n *FanoutNode) Pub() <-chan *Message

+ func (n *FanoutNode) Run(ctx context.Context) error

+ func (n *FanoutNode) Sub(in <-chan *Message)

+ type ForceStoppableNode interface

+ ForceStop func(ctx context.Context)

+ type LoggingNode interface

+ SetLogger func(log.CtxLogger)

+ type Message struct

+ Ctx context.Context

+ Record opencdc.Record

+ SourceID string

+ func (m *Message) Ack() error

+ func (m *Message) Acked() <-chan struct{}

+ func (m *Message) Clone() *Message

+ func (m *Message) ControlMessageType() ControlMessageType

+ func (m *Message) ID() string

+ func (m *Message) Nack(reason error, nodeID string) error

+ func (m *Message) Nacked() <-chan struct{}

+ func (m *Message) RegisterAckHandler(h AckHandler)

+ func (m *Message) RegisterNackHandler(h NackHandler)

+ func (m *Message) RegisterStatusHandler(h StatusChangeHandler)

+ func (m *Message) Status() MessageStatus

+ func (m *Message) StatusError() error

+ type MessageStatus int

+ const MessageStatusAcked

+ const MessageStatusNacked

+ const MessageStatusOpen

+ func (i MessageStatus) String() string

+ type MetricsNode struct

+ Histogram metrics.RecordBytesHistogram

+ Name string

+ func (n *MetricsNode) ID() string

+ func (n *MetricsNode) Pub() <-chan *Message

+ func (n *MetricsNode) Run(ctx context.Context) error

+ func (n *MetricsNode) SetLogger(logger log.CtxLogger)

+ func (n *MetricsNode) Sub(in <-chan *Message)

+ type NackHandler func(*Message, NackMetadata) error

+ type NackMetadata struct

+ NodeID string

+ Reason error

+ type Node interface

+ ID func() string

+ Run func(ctx context.Context) error

+ type OpenMessagesTracker sync.WaitGroup

+ func (t *OpenMessagesTracker) Add(msg *Message)

+ func (t *OpenMessagesTracker) Wait()

+ type ParallelNode struct

+ Name string

+ NewNode func(i int) PubSubNode

+ Workers int

+ func (n *ParallelNode) ID() string

+ func (n *ParallelNode) Pub() <-chan *Message

+ func (n *ParallelNode) Run(ctx context.Context) error

+ func (n *ParallelNode) SetLogger(logger log.CtxLogger)

+ func (n *ParallelNode) Sub(in <-chan *Message)

+ type Processor interface

+ Open func(ctx context.Context) error

+ Process func(context.Context, []opencdc.Record) []sdk.ProcessedRecord

+ Teardown func(context.Context) error

+ type ProcessorNode struct

+ Name string

+ Processor Processor

+ ProcessorTimer metrics.Timer

+ func (n *ProcessorNode) ID() string

+ func (n *ProcessorNode) Pub() <-chan *Message

+ func (n *ProcessorNode) Run(ctx context.Context) error

+ func (n *ProcessorNode) SetLogger(logger log.CtxLogger)

+ func (n *ProcessorNode) Sub(in <-chan *Message)

+ type PubNode interface

+ Pub func() <-chan *Message

+ type PubSubNode interface

+ type Source interface

+ Ack func(context.Context, []opencdc.Position) error

+ Errors func() <-chan error

+ ID func() string

+ Open func(context.Context) error

+ Read func(context.Context) ([]opencdc.Record, error)

+ Stop func(context.Context) (opencdc.Position, error)

+ Teardown func(context.Context) error

+ type SourceAckerNode struct

+ DLQHandlerNode *DLQHandlerNode

+ Name string

+ Source Source

+ func (n *SourceAckerNode) ID() string

+ func (n *SourceAckerNode) Pub() <-chan *Message

+ func (n *SourceAckerNode) Run(ctx context.Context) error

+ func (n *SourceAckerNode) SetLogger(logger log.CtxLogger)

+ func (n *SourceAckerNode) Sub(in <-chan *Message)

+ type SourceNode struct

+ Name string

+ PipelineTimer metrics.Timer

+ Source Source

+ func (n *SourceNode) ForceStop(ctx context.Context)

+ func (n *SourceNode) ID() string

+ func (n *SourceNode) Pub() <-chan *Message

+ func (n *SourceNode) Run(ctx context.Context) (err error)

+ func (n *SourceNode) SetLogger(logger log.CtxLogger)

+ func (n *SourceNode) Stop(ctx context.Context, reason error) error

+ type StatusChange struct

+ NackMetadata NackMetadata

+ New MessageStatus

+ Old MessageStatus

+ type StatusChangeHandler func(*Message, StatusChange) error

+ type StoppableNode interface

+ Stop func(ctx context.Context, reason error) error

+ type SubNode interface

+ Sub func(in <-chan *Message)