Enable Superstream for consumer

pkg/stream/consumer.go

@@ -65,6 +65,9 @@ func (c *Consumer) subscribeProperties() raw.SubscribeProperties {
 	if c.opts.SingleActiveConsumer {
 		s["single-active-consumer"] = "true"
 	}
+	if c.opts.SuperStream {
+		s["super-stream"] = c.Stream
+	}
 	return s
 }
 

pkg/stream/consumer_test.go

@@ -173,7 +173,34 @@ var _ = Describe("Smart Consumer", func() {
 	})
 
 	Describe("SuperStream Consumer", func() {
+		It("can enable the feature", func() {
+			subscribeProperties := raw.SubscribeProperties{"super-stream": "test-stream"}
+			chunkChan := make(chan *raw.Chunk)
+			gomock.InOrder(fakeRawClient.EXPECT().
+				NotifyChunk(gomock.AssignableToTypeOf(chunkChan)).Return(chunkChan),
+				fakeRawClient.EXPECT().
+					Subscribe(
+						gomock.AssignableToTypeOf(ctxType), //context
+						gomock.Eq("test-stream"),           // stream
+						gomock.Eq(uint16(1)),               // offsetType
+						gomock.Eq(uint8(1)),                // subscriptionId
+						gomock.Eq(uint16(2)),               // credit
+						gomock.Eq(subscribeProperties),     // properties
+						gomock.Eq(uint64(1)),               // offset
+					))
 
+			opts := &ConsumerOptions{
+				SuperStream:    true,
+				OffsetType:     uint16(1),
+				SubscriptionId: uint8(1),
+				Credit:         uint16(2),
+				Offset:         uint64(1),
+			}
+			handleMessages := func(consumerContext ConsumerContext, message *amqp.Message) {
+				fmt.Printf("messages: %s\n", message.Data)
+			}
+			consumer, _ := NewConsumer("test-stream", fakeRawClient, handleMessages, opts)
+			Expect(consumer.Subscribe(context.Background())).To(Succeed())
+		})
 	})
-
 })