Ticket System - Retrieve Only One Message and Close Consumer

[gabsferrara]

I need to create a ticket system, and I will use RabbitMQ Stream because I want to store the offset by user. Then, I will receive one trigger per user and consume the next message for that user.

I implemented it like this:
` ticket := make(chan string, 1)
semaphore := make(chan bool, 1)
var wgTicket sync.WaitGroup

handleMessages := func(consumerContext stream.ConsumerContext, message *amqp.Message) {
	select {
	case _, ok := <-semaphore:
		if !ok {
			return
		} else if len(ticket) == 0 {
			err := consumerContext.Consumer.StoreOffset()
			ticket <- string(message.Data[0])
			semaphore <- true
			close(semaphore)
			wgTicket.Done()
			CheckErr(err)
		}
	}
}

consumerUpdate := func(streamName string, isActive bool) stream.OffsetSpecification {
	offset, err := env.QueryOffset(user, streamName)
	if err != nil {
		return stream.OffsetSpecification{}.First()
	}
	return stream.OffsetSpecification{}.Offset(offset + 1)
}
semaphore <- true
wgTicket.Add(1)
consumer, _ := env.NewConsumer(
	viper.GetString("TICKET_QUEUE"),
	handleMessages,
	stream.NewConsumerOptions().
		SetConsumerName(user).
		SetSingleActiveConsumer(
			stream.NewSingleActiveConsumer(consumerUpdate)))
wgTicket.Wait()
consumerClose := consumer.NotifyClose()
consumer.GetCloseHandler()
consumer.Close()
<-consumerClose
msg := <-ticket
close(ticket)

`

However, when I run it in my terminal, I receive the message, but I also get the following warning:
2024/06/19 08:39:42 [warn] - Consumer Ticket is closed
2024/06/19 08:39:42 [warn] - Consumer Ticket is closed
2024/06/19 08:39:42 [warn] - Consumer Ticket is closed
2024/06/19 08:39:42 [warn] - Consumer Ticket is closed

I'm not sure why I'm getting this warning. How can I avoid logging it or prevent it from appearing?

[Gsantomaggio]

Can you please create a full example so I can run it without writing the needed code ?
Thanks

[gabsferrara]

Sure:
Code to have a initial trigger:

package producer_trigger

import (
	"errors"
	"fmt"
	"github.com/rabbitmq/rabbitmq-stream-go-client/pkg/amqp"
	"github.com/rabbitmq/rabbitmq-stream-go-client/pkg/message"
	"github.com/rabbitmq/rabbitmq-stream-go-client/pkg/stream"
	"golang.org/x/exp/rand"
	"os"
	rabbit_stream "rabbit-stream"
	"sync"
	"time"
)

func CheckErr(err error) {
	if err != nil {
		fmt.Printf("%s ", err)
		os.Exit(1)
	}
}

func handlePublishConfirm(confirms stream.ChannelPublishConfirm) {
	go func() {
		for confirmed := range confirms {
			for _, msg := range confirmed {
				if msg.IsConfirmed() {
					fmt.Printf("message %s stored \n  ", msg.GetMessage().GetData())
				} else {
					fmt.Printf("message %s failed \n  ", msg.GetMessage().GetData())
				}

			}
		}
	}()
}

func Run() {
	fmt.Println("publish message with filter user.")
	fmt.Println("Connecting to RabbitMQ streaming ...")

	// Connect to the broker ( or brokers )
	rabbit_stream.LoadEnvVariables()
	address := fmt.Sprintf("rabbitmq-stream://guest:guest@localhost:5552")

	env, err := stream.NewEnvironment(
		stream.NewEnvironmentOptions().SetUri(address))
	CheckErr(err)

	err = env.DeleteStream("TRIGGER")
	if err != nil && errors.Is(err, stream.StreamDoesNotExist) {
		fmt.Println("Stream does not exist. ")
	} else {
		CheckErr(err)
	}

	err = env.DeclareStream("TRIGGER",
		&stream.StreamOptions{
			MaxLengthBytes: stream.ByteCapacity{}.KB(100),
			MaxAge:         time.Hour,
		},
	)

	CheckErr(err)

	producer, err := env.NewProducer("TRIGGER",
		stream.NewProducerOptions().SetFilter(
			// Here we enable the filter
			// for each message we set the filter key.
			// the filter result is a string
			stream.NewProducerFilter(func(message message.StreamMessage) string {
				return fmt.Sprintf("%s", message.GetApplicationProperties()["user"])
			})))
	CheckErr(err)
	chPublishConfirm := producer.NotifyPublishConfirmation()
	handlePublishConfirm(chPublishConfirm)

	var wg sync.WaitGroup
	wg.Add(4)

	go send(producer, "John", &wg)
	go send(producer, "Mary", &wg)
	go send(producer, "Newbie", &wg)
	go send(producer, "Gabriel", &wg)

	wg.Wait()

	fmt.Println("Finish")
	producer.Close()
	env.Close()
}

func send(producer *stream.Producer, user string, wg *sync.WaitGroup) {
	rand.Seed(uint64(time.Now().UnixNano()))
	qtd := rand.Intn(50)
	fmt.Printf("user: %s, qnt: %d \n", user, qtd)
	for i := 0; i < qtd; i++ {
		msg := amqp.NewMessage([]byte(fmt.Sprintf("message number %d from user %s", i, user)))
		msg.ApplicationProperties = map[string]interface{}{"user": user}
		err := producer.Send(msg)
		time.Sleep(time.Millisecond * time.Duration(rand.Intn(75)))
		CheckErr(err)
	}
	fmt.Printf("DONE ------- Trigger Create: %s \n", user)
	wg.Done()
}

Populate Ticket Stream Queue:

package producer_ticket

import (
	"errors"
	"fmt"
	"github.com/rabbitmq/rabbitmq-stream-go-client/pkg/amqp"
	"github.com/rabbitmq/rabbitmq-stream-go-client/pkg/stream"
	"os"
	rabbit_stream "rabbit-stream"
	"strconv"
	"sync"
)

func CheckErr(err error) {
	if err != nil {
		fmt.Printf("%s ", err)
		os.Exit(1)
	}
}

func handlePublishConfirm(confirms stream.ChannelPublishConfirm) {
	go func() {
		for confirmed := range confirms {
			for _, msg := range confirmed {
				if msg.IsConfirmed() {
					fmt.Printf("message %s stored \n  ", msg.GetMessage().GetData())
				} else {
					fmt.Printf("message %s failed \n  ", msg.GetMessage().GetData())
				}

			}
		}
	}()
}

func Run() {
	fmt.Println("Ticket system ...")
	fmt.Println("Connecting to RabbitMQ streaming ...")

	// Connect to the broker ( or brokers )
	rabbit_stream.LoadEnvVariables()
	address := fmt.Sprintf("rabbitmq-stream://guest:guest@localhost:5552")

	env, err := stream.NewEnvironment(
		stream.NewEnvironmentOptions().SetUri(address))
	CheckErr(err)

	err = env.DeleteStream("TICKET")
	if err != nil && errors.Is(err, stream.StreamDoesNotExist) {
		fmt.Println("Stream does not exist. ")
	} else {
		CheckErr(err)
	}

	err = env.DeclareStream("TICKET",
		&stream.StreamOptions{},
	)

	CheckErr(err)

	producer, err := env.NewProducer("TICKET",
		stream.NewProducerOptions())
	CheckErr(err)
	chPublishConfirm := producer.NotifyPublishConfirmation()
	handlePublishConfirm(chPublishConfirm)

	var wg sync.WaitGroup
	wg.Add(1)

	go send(producer, 0, &wg)
	wg.Wait()

	fmt.Println("Ticket ok")
	producer.Close()
	env.Close()
}

func send(producer *stream.Producer, serie int, wg *sync.WaitGroup) {

	for i := serie; i < serie+1000; i++ {
		msg := amqp.NewMessage([]byte(strconv.Itoa(i)))
		err := producer.Send(msg)
		CheckErr(err)
	}
	wg.Done()
}

Ticket Consumer:

package ticket

import (
	"fmt"
	"github.com/rabbitmq/rabbitmq-stream-go-client/pkg/amqp"
	"github.com/rabbitmq/rabbitmq-stream-go-client/pkg/stream"
	"os"
	rabbit_stream "rabbit-stream"
	"sync"
	"time"
)

func CheckErr(err error) {
	if err != nil {
		fmt.Printf("%s ", err)
		os.Exit(1)
	}
}

func Run() {
	forever := make(chan bool)
	fmt.Println("POC Ticket")
	fmt.Println("Connecting to RabbitMQ streaming ...")
	rabbit_stream.LoadEnvVariables()
	address := fmt.Sprintf("rabbitmq-stream://guest:guest@localhost:5552")

	var err error

	env, err := stream.NewEnvironment(
		stream.NewEnvironmentOptions().SetUri(address))
	CheckErr(err)

	err = env.DeclareStream("TRIGGER",
		&stream.StreamOptions{
			MaxLengthBytes: stream.ByteCapacity{}.KB(100),
			MaxAge:         time.Hour,
		},
	)
	CheckErr(err)

	consumerName := "TriggerConsumer"
	consumerUpdate := func(streamName string, isActive bool) stream.OffsetSpecification {
		offset, err := env.QueryOffset(consumerName, streamName)
		if err != nil {
			return stream.OffsetSpecification{}.First()
		}
		return stream.OffsetSpecification{}.Offset(offset + 1)
	}

	handleMessages := func(consumerContext stream.ConsumerContext, message *amqp.Message) {
		err := consumerContext.Consumer.StoreOffset()
		var wg sync.WaitGroup
		wg.Add(1)
		findTicket(env, message.ApplicationProperties["user"].(string), &wg)
		wg.Wait()
		CheckErr(err)
	}

	_, err = env.NewConsumer(
		"TRIGGER",
		handleMessages,
		stream.NewConsumerOptions().
			SetConsumerName(consumerName).
			SetSingleActiveConsumer(
				stream.NewSingleActiveConsumer(consumerUpdate)))

	CheckErr(err)

	<-forever
}

func findTicket(env *stream.Environment, user string, wg *sync.WaitGroup) {

	ticket := make(chan string, 1)
	semaphore := make(chan bool, 1)
	var wgTicket sync.WaitGroup

	handleMessages := func(consumerContext stream.ConsumerContext, message *amqp.Message) {
		select {
		case _, ok := <-semaphore:
			if !ok {
				return
			} else if len(ticket) == 0 {
				err := consumerContext.Consumer.StoreOffset()
				ticket <- string(message.Data[0])
				semaphore <- true
				close(semaphore)
				wgTicket.Done()
				CheckErr(err)
			}
		}
	}

	consumerUpdate := func(streamName string, isActive bool) stream.OffsetSpecification {
		offset, err := env.QueryOffset(user, streamName)
		if err != nil {
			return stream.OffsetSpecification{}.First()
		}
		return stream.OffsetSpecification{}.Offset(offset + 1)
	}
	semaphore <- true
	wgTicket.Add(1)
	consumer, _ := env.NewConsumer(
		"TICKET",
		handleMessages,
		stream.NewConsumerOptions().
			SetConsumerName(user).
			SetSingleActiveConsumer(
				stream.NewSingleActiveConsumer(consumerUpdate)))
	wgTicket.Wait()
	consumerClose := consumer.NotifyClose()
	consumer.GetCloseHandler()
	consumer.Close()
	<-consumerClose
	msg := <-ticket
	close(ticket)

	fmt.Printf("get a ticket %s, from user %s \n", msg, user)
	wg.Done()
}

[Gsantomaggio]

The log you see is this one.
It means you closed the consumer, but it still contains messages in the chunk.
It is a warning message, but I think it can be changed to debug.

It is not a problem, actually.

[gabsferrara]

So there is no way for me to fetch a single message and close it without generating these warning logs, only by downloading the library and customizing it locally, right?

[Gsantomaggio]

The client downloads a chunk.
The chunk contains more messages ( see https://www.rabbitmq.com/docs/streams#client-operations )
We can change the log level to warning to debug. So, the log 2024/06/19 08:39:42 [warn] - Consumer Ticket is closed will be enabled only in debug mode

[gabsferrara]

I had understood that the "chunk" only happened when I used the "last", and since the filter uses the Offset, it would only fetch a single message. Thank you for the explanation!

This is my first time engaging in discussions here on GitHub, and with my limited professional experience (two years), I'm not sure if I understood correctly. When you say "We can change the log level to warning to debug," are you planning to change the "Consumer xxx is closed" warning in the future?

I also wanted to ask (if it's not a problem or if you think it's better to open another discussion), when I use a consumer with a filter, is there a way for me to know if there was enough time for the query to occur and if there are indeed no messages? Assuming that if there are messages according to that filter, I will perform action X, and if there are no messages, I will perform action Y.

[gabsferrara]

Can I open an PR myself to change the log?

Edit:
I open ->[PR](i open #320)

[Gsantomaggio]

Thank you @gabsferrara but please check: #319

Note: you could still have:

2024/06/19 17:15:28 [warn] - received a credit for an unknown subscriptionId: 0

in your case, I suggest using the following:

	stream.NewConsumerOptions().
			SetConsumerName(consumerName).
			SetInitialCredits(1).
	

to reduce the InitialCredits.

[gabsferrara]

You're welcome.

[Gsantomaggio]

when I use a consumer with a filter, is there a way for me to know if there was enough time for the query to occur and if there are indeed no messages?

No, there is no way.

The filter uses bloom filter and is applied to the chunk level.

More info here: https://github.com/rabbitmq/rabbitmq-stream-go-client?tab=readme-ov-file#consume-filtering