instrumented

Documentation

Index

• func NewInstrumentedSinkServer(factory proximo.AsyncSinkFactory, debug bool) proto.MessageSinkServer
• func NewInstrumentedSourceServer(factory proximo.AsyncSourceFactory, debug bool) proto.MessageSourceServer
• type AsyncSinkFactory
  • func (f AsyncSinkFactory) NewAsyncSink(ctx context.Context, req *proto.StartPublishRequest) (substrate.AsyncMessageSink, error)
• type AsyncSourceFactory
  • func (f AsyncSourceFactory) NewAsyncSource(ctx context.Context, req *proto.StartConsumeRequest) (substrate.AsyncMessageSource, error)
• type BackendStatusChecker
  • func NewBackendStatusChecker(maxFailedCount int) BackendStatusChecker

Functions

func NewInstrumentedSinkServer

func NewInstrumentedSinkServer(factory proximo.AsyncSinkFactory, debug bool) proto.MessageSinkServer

NewInstrumentedSinkServer returns a message sink server with error metrics.

func NewInstrumentedSourceServer

func NewInstrumentedSourceServer(factory proximo.AsyncSourceFactory, debug bool) proto.MessageSourceServer

NewInstrumentedSourceServer returns a message source server with error metrics.

Types

type AsyncSinkFactory

type AsyncSinkFactory struct {
	BackendStatusChecker BackendStatusChecker
	CounterOpts          prometheus.CounterOpts
	SinkFactory          proximo.AsyncSinkFactory
}

AsyncSinkFactory adds metrics and health check to every new sink.

func (AsyncSinkFactory) NewAsyncSink

func (f AsyncSinkFactory) NewAsyncSink(ctx context.Context, req *proto.StartPublishRequest) (substrate.AsyncMessageSink, error)

type AsyncSourceFactory

type AsyncSourceFactory struct {
	BackendStatusChecker BackendStatusChecker
	CounterOpts          prometheus.CounterOpts
	SourceFactory        proximo.AsyncSourceFactory
}

AsyncSourceFactory adds metrics and health check to every new sink.

func (AsyncSourceFactory) NewAsyncSource

func (f AsyncSourceFactory) NewAsyncSource(ctx context.Context, req *proto.StartConsumeRequest) (substrate.AsyncMessageSource, error)

type BackendStatusChecker

type BackendStatusChecker interface {
	CheckStatus(resp *op.CheckResponse)

	RemoveStatuser(id string)
	RegisterStatuser(id string, conn substrate.Statuser, cancel func())
}

BackendStatusChecker is an interface that check status of substrate connections.

func NewBackendStatusChecker

func NewBackendStatusChecker(maxFailedCount int) BackendStatusChecker

NewBackendStatusChecker returns a new instance of BackendStatusChecker.