// Consume converts messages to event instances, and pass the event to the provided handler.
Consume converts messages to event instances, and pass the event to the provided handler.
// Consume converts messages to event instances and pass the event to the provided handler .
func Consume(ctx context.Context, messageConsumer kafka.IConsumerGroup, handler Handler, cfg *config.Config) {
	var consume = func(workerID int) {
		for {
			select {
			case message, ok := <-messageConsumer.Channels().Upstream:
				if !ok {
					log.Info(ctx, "closing event consumer loop because upstream channel is closed", log.Data{"worker_id": workerID})
					return
				}
				messageCtx := context.Background()
				processMessage(messageCtx, message, handler)
				message.Release()
			case <-messageConsumer.Channels().Closer:
				log.Info(ctx, "closing event consumer loop because closer channel is closed", log.Data{"worker_id": workerID})
				return
			}
		}
	}
	for w := 1; w <= cfg.KafkaConfig.NumWorkers; w++ {
		go consume(w)
	}
}
func Consume ( ctx context . Context , messageConsumer kafka . IConsumerGroup , handler Handler , cfg * config . Config ) { var consume = func ( workerID int ) { for { select { case message , ok := <- messageConsumer . Channels ( ) . Upstream : if ! ok { log . Info ( ctx , " " , log . Data { " " : workerID } ) return } messageCtx := context . Background ( ) processMessage ( messageCtx , message , handler ) message . Release ( ) case <- messageConsumer . Channels ( ) . Closer : log . Info ( ctx , " " , log . Data { " " : workerID } ) return } } } for w := 1 ; w <= cfg . KafkaConfig . NumWorkers ; w ++ { go consume ( w ) } }
Consume converts messages to event instances, and pass the event to the provided handler.
Consume converts messages to event instances and pass the event to the provided handler .
// consume loop, to be executed by each worker // workers to consume messages in parallel