wrapper

Documentation

Index

• func NewStreamWrapper(stream stream.Streamer, whereObj condition.Condition, metrics metrics.Metricer) stream.Streamer
• type StreamWrapper
  • func (s *StreamWrapper) Check(ctx context.Context, msg message.Message) bool
  • func (s *StreamWrapper) Close() error
  • func (s *StreamWrapper) ID() string
  • func (s *StreamWrapper) Put(ctx context.Context, msg message.Message) error
  • func (s *StreamWrapper) Run(ctx context.Context) error

Functions

func NewStreamWrapper

func NewStreamWrapper(stream stream.Streamer, whereObj condition.Condition, metrics metrics.Metricer) stream.Streamer

NewStreamWrapper with support condition

Types

type StreamWrapper

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

StreamWrapper extends any stream interface with conditional check support to any message

func (*StreamWrapper) Check

func (s *StreamWrapper) Check(ctx context.Context, msg message.Message) bool

Check if the message meets the conditions

func (*StreamWrapper) Close

func (s *StreamWrapper) Close() error

Close stream and shut down all process

func (*StreamWrapper) ID

func (s *StreamWrapper) ID() string

ID returns unical stream identificator

func (*StreamWrapper) Put

func (s *StreamWrapper) Put(ctx context.Context, msg message.Message) error

Put message to the stream to process information

func (*StreamWrapper) Run

func (s *StreamWrapper) Run(ctx context.Context) error

Run the stream reading loop