Add settings to the stream queues (#87)

Adds additional configuration knobs for RabbitMQ stream queues by extending `StreamQueueSpecification` and introducing a duration-to-`x-max-age` formatter, with accompanying unit tests.

**Changes:**
- Extend `StreamQueueSpecification` to support `x-stream-filter-size-bytes`, `x-max-age` and `x-stream-max-segment-size-bytes`.
- Add `durationToMaxAge(time.Duration)` helper to format RabbitMQ `x-max-age` values.
- Add Ginkgo/Gomega tests for the new arguments and duration conversion.

Closes #64
---------

Signed-off-by: Gabriele Santomaggio <G.santomaggio@gmail.com>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Author: Gsantomaggio

pkg/rabbitmqamqp/entities.go

@@ -1,5 +1,10 @@
 package rabbitmqamqp
 
+import (
+	"fmt"
+	"time"
+)
+
 type iEntityIdentifier interface {
 	Id() string
 }
@@ -400,6 +405,9 @@ type StreamQueueSpecification struct {
 	Name               string
 	MaxLengthBytes     int64
 	InitialClusterSize int
+	MaxAge             time.Duration
+	SegmentSize        int64
+	FilterSizeBytes    int64
 	Arguments          map[string]any
 }
 
@@ -433,11 +441,61 @@ func (s *StreamQueueSpecification) buildArguments() map[string]any {
 		result["x-stream-initial-cluster-size"] = s.InitialClusterSize
 	}
 
+	if s.MaxAge != 0 {
+		result["x-max-age"] = durationToMaxAge(s.MaxAge)
+	}
+
+	if s.SegmentSize != 0 {
+		result["x-stream-max-segment-size-bytes"] = s.SegmentSize
+	}
+
+	if s.FilterSizeBytes != 0 {
+		result["x-stream-filter-size-bytes"] = s.FilterSizeBytes
+	}
+
 	result["x-queue-type"] = string(s.queueType())
 
 	return result
 }
 
+// durationToMaxAge converts a time.Duration to the RabbitMQ stream x-max-age string format.
+// RabbitMQ accepts durations expressed as a number followed by a unit: Y, M, D, h, m, s.
+// The largest fitting unit is used (e.g. 48h → "2D", 3600s → "1h").
+func durationToMaxAge(d time.Duration) string {
+	if d == 0 {
+		return "0s"
+	}
+	type unit struct {
+		suffix string
+		size   time.Duration
+	}
+	units := []unit{
+		{"Y", 365 * 24 * time.Hour},
+		{"M", 30 * 24 * time.Hour},
+		{"D", 24 * time.Hour},
+		{"h", time.Hour},
+		{"m", time.Minute},
+		{"s", time.Second},
+	}
+	for _, u := range units {
+		if d%u.size == 0 {
+			return fmt.Sprintf("%d%s", d/u.size, u.suffix)
+		}
+	}
+	// Fallback: round up to whole seconds for positive durations, enforcing a minimum of 1s.
+	if d > 0 {
+		secs := int64(d / time.Second)
+		if d%time.Second != 0 {
+			secs++
+		}
+		if secs < 1 {
+			secs = 1
+		}
+		return fmt.Sprintf("%ds", secs)
+	}
+	return fmt.Sprintf("%ds", int64(d.Seconds()))
+}
+
 // / **** Exchange ****
 
 // TExchangeType represents the type of exchange

pkg/rabbitmqamqp/entities_test.go

@@ -1,6 +1,8 @@
 package rabbitmqamqp
 
 import (
+	"time"
+
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 )
@@ -19,4 +21,83 @@ var _ = Describe("Entities", func() {
 			Expect(generatedName).ToNot(Equal(anotherGeneratedName), "different instances should generate different names")
 		})
 	})
+
+	Describe("StreamQueueSpecification", func() {
+		It("should set x-stream-max-segment-size-bytes when SegmentSize is set", func() {
+			spec := &StreamQueueSpecification{
+				Name:        "my-stream",
+				SegmentSize: 500_000_000,
+			}
+			args := spec.buildArguments()
+			Expect(args["x-stream-max-segment-size-bytes"]).To(Equal(int64(500_000_000)))
+		})
+
+		It("should not set x-stream-max-segment-size-bytes when SegmentSize is zero", func() {
+			spec := &StreamQueueSpecification{Name: "my-stream"}
+			args := spec.buildArguments()
+			Expect(args).ToNot(HaveKey("x-stream-max-segment-size-bytes"))
+		})
+
+		It("should set x-max-age when MaxAge is set", func() {
+			spec := &StreamQueueSpecification{
+				Name:   "my-stream",
+				MaxAge: 7 * 24 * time.Hour,
+			}
+			args := spec.buildArguments()
+			Expect(args["x-max-age"]).To(Equal("7D"))
+		})
+
+		It("should not set x-max-age when MaxAge is zero", func() {
+			spec := &StreamQueueSpecification{Name: "my-stream"}
+			args := spec.buildArguments()
+			Expect(args).ToNot(HaveKey("x-max-age"))
+		})
+
+		It("should set both x-max-age and x-stream-max-segment-size-bytes together", func() {
+			spec := &StreamQueueSpecification{
+				Name:        "my-stream",
+				MaxAge:      30 * 24 * time.Hour,
+				SegmentSize: 100_000_000,
+			}
+			args := spec.buildArguments()
+			Expect(args["x-max-age"]).To(Equal("1M"))
+			Expect(args["x-stream-max-segment-size-bytes"]).To(Equal(int64(100_000_000)))
+		})
+
+		It("should set x-stream-filter-size-bytes when FilterSizeBytes is set", func() {
+			spec := &StreamQueueSpecification{
+				Name:            "my-stream",
+				FilterSizeBytes: 16,
+			}
+			args := spec.buildArguments()
+			Expect(args["x-stream-filter-size-bytes"]).To(Equal(int64(16)))
+		})
+
+		It("should not set x-stream-filter-size-bytes when FilterSizeBytes is zero", func() {
+			spec := &StreamQueueSpecification{Name: "my-stream"}
+			args := spec.buildArguments()
+			Expect(args).ToNot(HaveKey("x-stream-filter-size-bytes"))
+		})
+
+		It("should always set x-queue-type to stream", func() {
+			spec := &StreamQueueSpecification{Name: "my-stream"}
+			args := spec.buildArguments()
+			Expect(args["x-queue-type"]).To(Equal("stream"))
+		})
+	})
+
+	Describe("durationToMaxAge", func() {
+		DescribeTable("converts duration to RabbitMQ max-age string",
+			func(d time.Duration, expected string) {
+				Expect(durationToMaxAge(d)).To(Equal(expected))
+			},
+			Entry("seconds", 30*time.Second, "30s"),
+			Entry("minutes", 5*time.Minute, "5m"),
+			Entry("hours", 2*time.Hour, "2h"),
+			Entry("days", 3*24*time.Hour, "3D"),
+			Entry("months (30 days)", 30*24*time.Hour, "1M"),
+			Entry("years (365 days)", 365*24*time.Hour, "1Y"),
+			Entry("zero", time.Duration(0), "0s"),
+		)
+	})
 })