gstream

Documentation

Index

• func NewFixedSizeObserver[T interface{}](config FixedSizeObserverConfiguration) (types.FixedSizeObserver[T], error)
• func NewGrpcClient[T interface{}](config GrpcClientConfiguration[T]) (types.GrpcClient[T], error)
• func NewGrpcStreamDecorator[I interface{}, O interface{}](config GrpcStreamDecoratorConfiguration[I, O]) (types.GrpcStreamDecorator[I, O], error)
• func NewObserver[T interface{}](config ObserverConfiguration) (types.Observer[T], error)
• type FixedSizeObserverConfiguration
• type GrpcClientConfiguration
• type GrpcStreamDecoratorConfiguration
• type ObserverConfiguration

Functions

func NewFixedSizeObserver

func NewFixedSizeObserver[T interface{}](config FixedSizeObserverConfiguration) (types.FixedSizeObserver[T], error)

func NewGrpcClient

func NewGrpcClient[T interface{}](config GrpcClientConfiguration[T]) (types.GrpcClient[T], error)

func NewGrpcStreamDecorator

func NewGrpcStreamDecorator[I interface{}, O interface{}](config GrpcStreamDecoratorConfiguration[I, O]) (types.GrpcStreamDecorator[I, O], error)

func NewObserver

func NewObserver[T interface{}](config ObserverConfiguration) (types.Observer[T], error)

Types

type FixedSizeObserverConfiguration

type FixedSizeObserverConfiguration struct {
	Ctx                      context.Context
	SubscribersChannelLength *int
	// SkipOnFail skip not delivered messages or retry delivery them unlimited
	SkipOnFail bool
	// SkipPublishWithoutSubscribers skip element publishing without subscribers
	SkipPublishWithoutSubscribers bool
	// ElementsCheckInterval sleep time between checking of elements existence
	ElementsCheckInterval *time.Duration
	// SubscribersCheckInterval sleep time between checking of subscribers
	SubscribersCheckInterval *time.Duration
	// SkipAfterDeliveryRetriesCount drop messages after retries count
	SkipAfterDeliveryRetriesCount *int
}

type GrpcClientConfiguration

type GrpcClientConfiguration[T interface{}] struct {
	Ctx                           context.Context
	ServerStream                  grpc.ServerStream
	ClientStreamProvider          func() (grpc.ClientStream, error)
	MessagesCallback              func(ctx context.Context, grpcClient types.GrpcClient[T], msg *T) error
	ErrorsCallback                func(grpcClient types.GrpcClient[T], err error) error
	SkipMessagesIfClientWithoutId bool
	MessagesChannelSize           *int
	GenerateId                    bool
}

type GrpcStreamDecoratorConfiguration

type GrpcStreamDecoratorConfiguration[I interface{}, O interface{}] struct {
	Ctx                  context.Context
	ServerStream         grpc.ServerStream
	ClientStreamProvider func() (grpc.ClientStream, error)
	ChannelSize          *int
	MappingFunc          func(msg *I) O
	ErrorCallback        func(err error) error
}

type ObserverConfiguration

type ObserverConfiguration struct {
	WaitingRepeatInterval time.Duration
}