consumer offsets a bit unclear

Is your feature request related to a problem? Please describe.

Hello!

In my project I've to manually track offsets. It's done in a different (main) goroutine, while the consumer callback is called in it's own goroutine, so due to async nature i can't rely on consumer.Offset tracking outside of consumer routine, so I have to do something like that:

package main

import (
    "context"
    "errors"
    "fmt"
    "os"
    "os/signal"
    "strconv"

    "github.com/rabbitmq/rabbitmq-stream-go-client/pkg/amqp"
    "github.com/rabbitmq/rabbitmq-stream-go-client/pkg/stream"
)

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

type offsetMsg struct {
    offt int
    msg  *amqp.Message
}

func newHandler(acceptChan chan offsetMsg) func(stream.ConsumerContext, *amqp.Message) {
    return func(consumerContext stream.ConsumerContext, message *amqp.Message) {
        // fmt.Println(string(message.GetData()), message.DeliveryTag, consumerContext.Consumer.GetOffset())
        acceptChan <- offsetMsg{offt: int(consumerContext.Consumer.GetOffset()), msg: message}
    }
}

func storeOffset(c *stream.Consumer, offt int64) error {
    err := c.StoreCustomOffset(offt)
    if err != nil {
        fmt.Println(err)
    } else {
        fmt.Println("offset stored", offt)
    }
    return err
}

func main() {
    ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, os.Kill)
    defer stop()
    env, err := stream.NewEnvironment(
        stream.NewEnvironmentOptions().
            SetHost("localhost").
            SetPort(5552).
            SetUser("guest").
            SetPassword("guest"))
    CheckErr(err)

    streamName := "stream"
    err = env.DeclareStream(streamName,
        &stream.StreamOptions{
            MaxLengthBytes: stream.ByteCapacity{}.MB(500),
        },
    )
    CheckErr(err)

    consName := "consumer_"
    producer, err := env.NewProducer(streamName, nil)
    CheckErr(err)

    go func() {
        for i := 0; i < 5; i++ {
            err := producer.Send(amqp.NewMessage([]byte("hello_world_" + strconv.Itoa(i))))
            CheckErr(err)
        }
    }()

    offt, err := env.QueryOffset(consName, streamName)
    if err != nil {
        if !errors.Is(err, stream.OffsetNotFoundError) {
            panic(err)
        }
    } else {
        offt++
    }
    fmt.Println("queried offset")

    acceptCh := make(chan offsetMsg) //*amqp.Message)
    handler := newHandler(acceptCh)

    cons, err := env.NewConsumer(streamName, handler, stream.
        NewConsumerOptions().
        SetManualCommit().
        SetOffset(stream.OffsetSpecification{}.Offset(offt)).SetConsumerName(consName),
    )

    go func() {
        counter := 0
        for {
            select {
            case <-ctx.Done():
                return
            case data := <-acceptCh:
                counter++
                fmt.Printf("data: %s, offset: %d, deliveryTag: %v\n", string(data.msg.GetData()), data.offt, data.msg.DeliveryTag)
                if counter%5 == 0 {
                    storeOffset(cons, int64(data.offt))
                }
            }
        }
    }()

    CheckErr(err)
    defer cons.Close()

    <-ctx.Done()
}

The output is next:

queried offset
data: hello_world_0, offset: 0, deliveryTag: []
data: hello_world_1, offset: 1, deliveryTag: []
data: hello_world_2, offset: 2, deliveryTag: []
data: hello_world_3, offset: 3, deliveryTag: []
data: hello_world_4, offset: 4, deliveryTag: []
offset stored 4

Also, as I saw in library - the consumer routine works with array of offsetMessages, where offsetMessage - pointer to *amqp.Message and offset. As a result there are two unnecessary wraps. Maybe it's worth to send offsetMessage to the callback, or extend the amqp.Message with the offset field (like it's done inside amqp lib DeliveryTag field).

Here is an amqp-code:

package main

import (
    "context"
    "fmt"
    "os"
    "os/signal"
    "syscall"

    amqp "github.com/rabbitmq/amqp091-go"
)

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

func main() {
    ctx, done := signal.NotifyContext(context.Background(), os.Interrupt, os.Kill)
    defer done()
    produce("amqp://localhost:5672", "exch", "fanout", "key")
    c := NewConsumer("amqp://localhost:5672", "exch", "fanout", "q", "key", "tag")
    SetupCloseHandler(c)
    <-ctx.Done()
}

func produce(uri, exchange, exchangeType, routingKey string) {
    connection, err := amqp.Dial(uri)
    CheckErr(err)
    defer connection.Close()

    channel, err := connection.Channel()
    CheckErr(err)
    err = channel.ExchangeDeclare(exchange, exchangeType, true, false, false, false, nil)
    CheckErr(err)

    for i := 0; i < 5; i++ {
        body := []byte(fmt.Sprintf("body_%d", i))
        err := channel.Publish(
            exchange,   // publish to an exchange
            routingKey, // routing to 0 or more queues
            false,      // mandatory
            false,      // immediate
            amqp.Publishing{
                Headers:         amqp.Table{},
                ContentType:     "text/plain",
                ContentEncoding: "",
                Body:            []byte(body),
                DeliveryMode:    amqp.Transient, // 1=non-persistent, 2=persistent
                Priority:        0,              // 0-9
                // a bunch of application/implementation-specific fields
            },
        )
        CheckErr(err)
    }
}

type Consumer struct {
    conn    *amqp.Connection
    channel *amqp.Channel
    tag     string
    done    chan error
}

func SetupCloseHandler(consumer *Consumer) {
    c := make(chan os.Signal, 2)
    signal.Notify(c, os.Interrupt, syscall.SIGTERM)
    go func() {
        <-c
        fmt.Printf("Ctrl+C pressed in Terminal")
        consumer.channel.Close()
        consumer.conn.Close()
        fmt.Println(<-consumer.done)
        os.Exit(0)
    }()
}

func NewConsumer(amqpURI, exchange, exchangeType, queueName, key, ctag string) *Consumer {
    c := &Consumer{
        conn: nil, channel: nil, tag: ctag, done: make(chan error),
    }

    var err error
    config := amqp.Config{Properties: amqp.NewConnectionProperties()}
    config.Properties.SetClientConnectionName("sample-consumer")

    c.conn, err = amqp.DialConfig(amqpURI, config)
    CheckErr(err)

    go func() {
        fmt.Printf("closing: %s", <-c.conn.NotifyClose(make(chan *amqp.Error)))
    }()

    c.channel, err = c.conn.Channel()
    CheckErr(err)

    err = c.channel.ExchangeDeclare(exchange, exchangeType, true, false, false, false, nil)
    CheckErr(err)

    queue, err := c.channel.QueueDeclare(queueName, true, false, false, false, nil)
    CheckErr(err)

    err = c.channel.QueueBind(queue.Name, key, exchange, false, nil)
    CheckErr(err)

    deliveries, err := c.channel.Consume(queue.Name, c.tag, false, false, false, false, nil)
    CheckErr(err)

    go handle(deliveries, c.done)

    return c
}

func handle(deliveries <-chan amqp.Delivery, done chan error) {
    cleanup := func() {
        fmt.Printf("handle: deliveries channel closed")
        done <- nil
    }

    defer cleanup()

    for d := range deliveries {
        fmt.Printf("data: %s, deliveryTag: %d\n", string(d.Body), d.DeliveryTag)
        CheckErr(d.Ack(false))
    }
}

It's output is next:

data: body_0, deliveryTag: 1
data: body_1, deliveryTag: 2
data: body_2, deliveryTag: 3
data: body_3, deliveryTag: 4
data: body_4, deliveryTag: 5

Describe the solution you'd like

Maybe it's worth to make api of amqp Messages similar to amqp library or extend current amqp message with new Offset field. And the second proposal: maybe it's worth to add the ability for user to interact with channel based approach, like it's done inside amqp lib (for ex create new API method that will return consumer channel to user application)

Describe alternatives you've considered

No response

Additional context

No response

rabbitmq / rabbitmq-stream-go-client