subscribers

func (*Base) Ack ¶

func (b *Base) Ack() bool

Ack returns whether or not this is a synchronous/async subscriber.

func (*Base) C ¶

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

C returns the event channel for optional subscribers.

func (*Base) Closed ¶

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

Closed indicates to the broker that the subscriber is closed for business.

func (*Base) Halt ¶

func (b *Base) Halt()

Halt is called by the broker on shutdown, this closes the open channels.

func (*Base) ID ¶

func (b *Base) ID() int

ID returns the subscriber ID.

func (*Base) Pause ¶

func (b *Base) Pause()

Pause the current subscriber will not receive events from the channel.

func (*Base) Resume ¶

func (b *Base) Resume()

Resume unpauzes the subscriber.

func (*Base) SetID ¶

func (b *Base) SetID(id int)

SetID set the ID (exposed only to broker).

func (*Base) Skip ¶

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

Skip lets the broker know that the subscriber is not receiving events.

func (*Service) ObserveEvents ¶

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

func (*Service) ObserveEventsOnStream ¶

func (s *Service) ObserveEventsOnStream(ctx context.Context, retries int,
	sub StreamSubscription,
) (<-chan []*eventspb.BusEvent, chan<- int)

func (*StreamSub) GetData ¶

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

GetData returns events from buffer, all if bufSize == 0, or max buffer size (rest are kept in data slice).

func (*StreamSub) Halt ¶

func (s *StreamSub) Halt()

func (*StreamSub) Push ¶

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

func (StreamSub) Types ¶

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

func (*StreamSub) UpdateBatchSize ¶

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

UpdateBatchSize changes the batch size, and returns whatever the current buffer contains it's effectively a poll of current events ignoring requested batch size.