pulsario

Documentation

Index

• func Ack(ctx context.Context, consumers []pulsar.Consumer, ...)
• func Receive(ctx context.Context, consumer pulsar.Consumer, consumerId int, bufferSize int, ...) chan *model.ConsumerMessage

Functions

func Ack

func Ack(ctx context.Context, consumers []pulsar.Consumer, msgs chan []*model.ConsumerMessageId, wg *sync.WaitGroup)

Ack will ack all pulsar messages coming in on the msgs channel. The incoming messages contain a consumer id which corresponds to the index of the consumer that should be used to perform the ack. In theory, the acks could be done in parallel, however its unlikely that they will be a performance bottleneck

func Receive

func Receive(ctx context.Context, consumer pulsar.Consumer, consumerId int, bufferSize int, receiveTimeout time.Duration, backoffTime time.Duration) chan *model.ConsumerMessage

Receive returns a channel containing messages received from pulsar. This channel will remain open until the supplied context is closed. consumerId: Internal Id of the consumer. We use this so that when messages from different consumers are multiplexed, we know which messages originated form which consumers bufferSize: sets the size of the buffer in the returned channel receiveTimeout: sets how long the pulsar consumer will wait for a message before retrying backoffTime: sets how long the consumer will wait before retrying if the pulsar consumer indicates an error receiving from pulsar.