subscribers

Aug 16, 2022

Changes in this version

+ type Base struct

+ func NewBase(ctx context.Context, buf int, ack bool) *Base

+ func (b *Base) Ack() bool

+ func (b *Base) C() chan<- []events.Event

+ func (b *Base) Closed() <-chan struct{}

+ func (b *Base) Halt()

+ func (b *Base) ID() int

+ func (b *Base) Pause()

+ func (b *Base) Resume()

+ func (b *Base) SetID(id int)

+ func (b *Base) Skip() <-chan struct{}

+ type Broker interface

+ Subscribe func(s broker.Subscriber) int

+ Unsubscribe func(id int)

+ type EventFilter func(events.Event) bool

+ type MarketStreamEvent interface

+ StreamMarketMessage func() *eventspb.BusEvent

+ type Service struct

+ func NewService(broker Broker) *Service

+ func (s *Service) ObserveEvents(ctx context.Context, retries int, eTypes []events.Type, batchSize int, ...) (<-chan []*eventspb.BusEvent, chan<- int)

+ type StreamEvent interface

+ StreamMessage func() *eventspb.BusEvent

+ type StreamSub struct

+ func NewStreamSub(ctx context.Context, types []events.Type, batchSize int, ...) *StreamSub

+ func (s *StreamSub) GetData(ctx context.Context) []*eventspb.BusEvent

+ func (s *StreamSub) Halt()

+ func (s *StreamSub) Push(evts ...events.Event)

+ func (s *StreamSub) UpdateBatchSize(ctx context.Context, size int) []*eventspb.BusEvent

+ func (s StreamSub) Types() []events.Type