udsource

Documentation

Overview

Package udsource implements user-defined source, enabling the platform to communicate with user-defined data sources through gRPC.

Index

• func ConvertToUserDefinedSourceOffset(offset isb.Offset) *sourcepb.Offset
• func New(vertexInstance *dfv1.VertexInstance, writers map[string][]isb.BufferWriter, ...) (sourcer.Sourcer, error)
• func NewUserDefinedSourceOffset(offset *sourcepb.Offset) isb.Offset
• type GRPCBasedUDSource
  • func NewUDSgRPCBasedUDSource(vertexInstance *dfv1.VertexInstance, c sourceclient.Client) (*GRPCBasedUDSource, error)
  • func (u *GRPCBasedUDSource) ApplyAckFn(ctx context.Context, offsets []isb.Offset) error
  • func (u *GRPCBasedUDSource) ApplyPartitionFn(ctx context.Context) ([]int32, error)
  • func (u *GRPCBasedUDSource) ApplyPendingFn(ctx context.Context) (int64, error)
  • func (u *GRPCBasedUDSource) ApplyReadFn(ctx context.Context, count int64, timeout time.Duration) ([]*isb.ReadMessage, error)
  • func (u *GRPCBasedUDSource) CloseConn(ctx context.Context) error
  • func (u *GRPCBasedUDSource) IsHealthy(ctx context.Context) error
  • func (u *GRPCBasedUDSource) WaitUntilReady(ctx context.Context) error
• type Option
  • func WithLogger(l *zap.SugaredLogger) Option
  • func WithReadTimeout(t time.Duration) Option

Functions

func ConvertToUserDefinedSourceOffset

func ConvertToUserDefinedSourceOffset(offset isb.Offset) *sourcepb.Offset

func New

func New(
	vertexInstance *dfv1.VertexInstance,
	writers map[string][]isb.BufferWriter,
	fsd forwarder.ToWhichStepDecider,
	transformer applier.SourceTransformApplier,
	sourceApplier *GRPCBasedUDSource,
	fetchWM fetch.SourceFetcher,
	toVertexPublisherStores map[string]store.WatermarkStore,
	publishWMStores store.WatermarkStore,
	idleManager wmb.IdleManager,
	opts ...Option) (sourcer.Sourcer, error)

func NewUserDefinedSourceOffset

func NewUserDefinedSourceOffset(offset *sourcepb.Offset) isb.Offset

Types

type GRPCBasedUDSource

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

GRPCBasedUDSource applies a user-defined source over gRPC connection where server is the UDSource.

func NewUDSgRPCBasedUDSource

func NewUDSgRPCBasedUDSource(vertexInstance *dfv1.VertexInstance, c sourceclient.Client) (*GRPCBasedUDSource, error)

NewUDSgRPCBasedUDSource accepts a vertex instance, gRPC client and returns a new GRPCBasedUDSource.

func (*GRPCBasedUDSource) ApplyAckFn

func (u *GRPCBasedUDSource) ApplyAckFn(ctx context.Context, offsets []isb.Offset) error

ApplyAckFn acknowledges messages in the source.

func (*GRPCBasedUDSource) ApplyPartitionFn

func (u *GRPCBasedUDSource) ApplyPartitionFn(ctx context.Context) ([]int32, error)

ApplyPartitionFn returns the partitions associated with the source.

func (*GRPCBasedUDSource) ApplyPendingFn

func (u *GRPCBasedUDSource) ApplyPendingFn(ctx context.Context) (int64, error)

ApplyPendingFn returns the number of pending messages in the source.

func (*GRPCBasedUDSource) ApplyReadFn

func (u *GRPCBasedUDSource) ApplyReadFn(ctx context.Context, count int64, timeout time.Duration) ([]*isb.ReadMessage, error)

ApplyReadFn reads messages from the source.

func (*GRPCBasedUDSource) CloseConn

func (u *GRPCBasedUDSource) CloseConn(ctx context.Context) error

CloseConn closes the gRPC client connection.

func (*GRPCBasedUDSource) IsHealthy

func (u *GRPCBasedUDSource) IsHealthy(ctx context.Context) error

IsHealthy checks if the udsource is healthy.

func (*GRPCBasedUDSource) WaitUntilReady

func (u *GRPCBasedUDSource) WaitUntilReady(ctx context.Context) error

WaitUntilReady waits until the udsource is connected.

type Option

type Option func(*userDefinedSource) error

func WithLogger

func WithLogger(l *zap.SugaredLogger) Option

WithLogger is used to return logger information

func WithReadTimeout

func WithReadTimeout(t time.Duration) Option

WithReadTimeout sets the read timeout