Merge branch 'main' into panic-chunk-dispatching

Author: Gsantomaggio

README.md

@@ -1,13 +1,14 @@
 <h1 align="center">RabbitMQ Stream GO Client</h1>
 
 ---
-<div align="center">
+[![Go Report Card][ReportCard-Image]][ReportCard-Url] [![Build Status][Build-Status-Image]][Build-Status-Url] 
 
-![Build](https://github.com/rabbitmq/rabbitmq-stream-go-client/actions/workflows/build_and_test.yml/badge.svg)
-[![codecov](https://codecov.io/gh/rabbitmq/rabbitmq-stream-go-client/branch/main/graph/badge.svg?token=HZD4S71QIM)](https://codecov.io/gh/rabbitmq/rabbitmq-stream-go-client)
+[ReportCard-Url]: https://goreportcard.com/report/github.com/rabbitmq/rabbitmq-stream-go-client
+[ReportCard-Image]: https://goreportcard.com/badge/github.com/rabbitmq/rabbitmq-stream-go-client
+[Build-Status-Url]: https://github.com/rabbitmq/rabbitmq-stream-go-client/actions
+[Build-Status-Image]: https://github.com/rabbitmq/rabbitmq-stream-go-client/actions/workflows/build_and_test.yml/badge.svg?branch=main
 
 Go client for [RabbitMQ Stream Queues](https://github.com/rabbitmq/rabbitmq-server/tree/master/deps/rabbitmq_stream)
-</div>
 
 # Table of Contents
 

VERSION

@@ -1 +1 @@
-1.5.3
+1.5.4

go.mod

@@ -12,7 +12,7 @@ require (
 	github.com/pkg/errors v0.9.1
 	github.com/spaolacci/murmur3 v1.1.0
 	github.com/spf13/cobra v1.9.1
-	golang.org/x/text v0.24.0
+	golang.org/x/text v0.25.0
 )
 
 require (

go.sum

@@ -55,8 +55,8 @@ golang.org/x/net v0.36.0 h1:vWF2fRbw4qslQsQzgFqZff+BItCvGFQqKzKIzx1rmoA=
 golang.org/x/net v0.36.0/go.mod h1:bFmbeoIPfrw4sMHNhb4J9f6+tPziuGjq7Jk/38fxi1I=
 golang.org/x/sys v0.30.0 h1:QjkSwP/36a20jFYWkSue1YwXzLmsV5Gfq7Eiy72C1uc=
 golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
-golang.org/x/text v0.24.0 h1:dd5Bzh4yt5KYA8f9CJHCP4FB4D51c2c6JvN37xJJkJ0=
-golang.org/x/text v0.24.0/go.mod h1:L8rBsPeo2pSS+xqN0d5u2ikmjtmoJbDBT1b7nHvFCdU=
+golang.org/x/text v0.25.0 h1:qVyWApTSYLk/drJRO5mDlNYskwQznZmkpV2c8q9zls4=
+golang.org/x/text v0.25.0/go.mod h1:WEdwpYrmk1qmdHvhkSTNPm3app7v4rsT8F2UD6+VHIA=
 golang.org/x/tools v0.28.0 h1:WuB6qZ4RPCQo5aP3WdKZS7i595EdWqWR8vqJTlwTVK8=
 golang.org/x/tools v0.28.0/go.mod h1:dcIOrVd3mfQKTgrDVQHqCPMWy6lnhfhtX3hLXYVLfRw=
 google.golang.org/protobuf v1.36.1 h1:yBPeRvTftaleIgM3PZ/WBIZ7XM/eEYAaEyCwvyjq/gk=

pkg/integration_test/stream_integration_test.go

@@ -1,15 +1,16 @@
 package integration_test
 
 import (
+	"errors"
 	"fmt"
+	"github.com/rabbitmq/rabbitmq-stream-go-client/pkg/message"
 	"sync"
 	"time"
 
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 
 	"github.com/rabbitmq/rabbitmq-stream-go-client/pkg/amqp"
-	"github.com/rabbitmq/rabbitmq-stream-go-client/pkg/message"
 	stream "github.com/rabbitmq/rabbitmq-stream-go-client/pkg/stream"
 )
 
@@ -146,4 +147,97 @@ var _ = Describe("StreamIntegration", func() {
 			Expect(consumer.GetOffset()).To(BeNumerically("==", expectedCurrentOffset))
 		})
 	})
+
+	Context("Initial timestamp offset when no messages exist", func() {
+		var (
+			addresses []string = []string{
+				"rabbitmq-stream://guest:guest@localhost:5552/"}
+			streamName string = "empty-test-stream"
+			streamEnv  *stream.Environment
+		)
+
+		// init empty stream
+		BeforeEach(func() {
+			var err error
+			streamEnv, err = stream.NewEnvironment(
+				stream.NewEnvironmentOptions().SetUris(addresses))
+			Expect(err).ToNot(HaveOccurred())
+
+			err = streamEnv.DeclareStream(streamName,
+				stream.NewStreamOptions().SetMaxLengthBytes(stream.ByteCapacity{}.GB(2)))
+			Expect(err).ToNot(HaveOccurred())
+		})
+
+		AfterEach(func() {
+			Expect(streamEnv.DeleteStream(streamName)).
+				To(SatisfyAny(
+					Succeed(),
+					MatchError(stream.StreamDoesNotExist),
+				))
+		})
+
+		It("correctly handles offsets using timestamps when no messages exist", func() {
+			var err error
+			const consumerName = "timestamp-offset-consumer"
+
+			lastMinute := time.Now().Add(-time.Minute).UnixMilli()
+
+			// Implement the UpdateConsumer function to return a timestamp-based offset if no offset exists
+			// For example, we add a new consumer to the incoming stream and don't want to reread it from the beginning.
+			updateConsumer := func(streamName string, isActive bool) stream.OffsetSpecification {
+				offset, err := streamEnv.QueryOffset(consumerName, streamName)
+				if errors.Is(err, stream.OffsetNotFoundError) {
+					return stream.OffsetSpecification{}.Timestamp(lastMinute)
+				}
+
+				Expect(err).ToNot(HaveOccurred())
+
+				return stream.OffsetSpecification{}.Offset(offset + 1)
+			}
+
+			options := stream.NewConsumerOptions().
+				SetConsumerName(consumerName).
+				SetAutoCommit(stream.NewAutoCommitStrategy().
+					SetFlushInterval(time.Second)).
+				SetSingleActiveConsumer(stream.NewSingleActiveConsumer(updateConsumer))
+
+			// Create the consumer
+			consumer, err := streamEnv.NewConsumer(
+				streamName,
+				func(ctx stream.ConsumerContext, msg *amqp.Message) {},
+				options,
+			)
+			Expect(err).NotTo(HaveOccurred())
+
+			// Wait for a flush without messages
+			// An incorrect offset is stored during this flush
+			time.Sleep(time.Millisecond * 1200)
+			Expect(consumer.Close()).ToNot(HaveOccurred())
+
+			// Re-create the consumer
+			consumeIsStarted := make(chan struct{})
+			handleMessages := func(ctx stream.ConsumerContext, msg *amqp.Message) {
+				close(consumeIsStarted)
+			}
+
+			consumer, err = streamEnv.NewConsumer(streamName, handleMessages, options)
+			Expect(err).NotTo(HaveOccurred())
+
+			producer, err := streamEnv.NewProducer(streamName, nil)
+			Expect(err).ToNot(HaveOccurred())
+			body := `{"name": "item-1}`
+			err = producer.Send(amqp.NewMessage([]byte(body)))
+			Expect(err).ToNot(HaveOccurred())
+
+			// check if messages are consumed
+			select {
+			case <-consumeIsStarted:
+			case <-time.After(time.Second * 1):
+				Fail("Timeout waiting for consumer to start")
+			}
+
+			Expect(consumer.GetOffset()).To(BeNumerically("<=", 0))
+		})
+
+	})
 })

pkg/stream/client.go

@@ -926,7 +926,7 @@ func (c *Client) declareSubscriber(streamName string,
 	// copy the option offset to the consumer offset
 	// the option.offset won't change ( in case we need to retrive the original configuration)
 	// consumer.current offset will be moved when reading
-	if !options.IsSingleActiveConsumerEnabled() {
+	if !options.IsSingleActiveConsumerEnabled() && options.Offset.isOffset() {
 		consumer.setCurrentOffset(options.Offset.offset)
 	}
 

pkg/stream/constants.go

@@ -19,7 +19,7 @@ const (
 const initBufferPublishSize = 2 + 2 + 1 + 4
 
 const (
-	ClientVersion = "1.5.3"
+	ClientVersion = "1.5.4"
 
 	commandDeclarePublisher       = 1
 	commandPublish                = 2

pkg/stream/server_frame.go

@@ -596,7 +596,7 @@ func (c *Client) handleConsumerUpdate(readProtocol *ReaderProtocol, r *bufio.Rea
 		isActive == 1)
 	consumer.options.SingleActiveConsumer.offsetSpecification = responseOff
 
-	if isActive == 1 {
+	if isActive == 1 && responseOff.isOffset() {
 		consumer.setCurrentOffset(responseOff.offset)
 	}