Merge pull request #74 from Gsantomaggio/e2e-behaviour

E2E test for consuming 100 messages.

Author: ablease

internal/command_types.go

@@ -118,6 +118,7 @@ const (
 	CommandSubscribeResponse               uint16 = 0x8007
 	CommandCreditResponse                  uint16 = 0x8009
 	CommandQueryOffsetResponse             uint16 = 0x800b
+	CommandUnsubscribeResponse             uint16 = 0x800c
 	CommandCreateResponse                  uint16 = 0x800d
 	CommandDeleteResponse                  uint16 = 0x800e
 	CommandMetadataResponse                uint16 = 0x800f

pkg/e2e/end_to_end_test.go

@@ -75,7 +75,7 @@ var _ = Describe("E2E", Serial, Label("e2e"), func() {
 			for {
 				select {
 				case <-ctx.Done():
-					Fail(fmt.Sprintf("context timed out: expected to receive 1_000_000 confirmations: received %d", len(confirmedIds)))
+					Fail(fmt.Sprintf("context timed out: expected to receive 100_000 confirmations: received %d", len(confirmedIds)))
 				case confirm, ok := <-c:
 					if !ok {
 						return
@@ -113,7 +113,98 @@ var _ = Describe("E2E", Serial, Label("e2e"), func() {
 
 		By("closing the connection")
 		Expect(streamClient.Close(itCtx)).To(Succeed())
+
 	}, SpecTimeout(120*time.Second))
+
+	// Send and Recveive Messages, assert messages received are valid.
+	It("sends, and receives messages", Label("behaviour"), func(ctx SpecContext) {
+		h := slog.HandlerOptions{Level: slog.LevelDebug}.NewTextHandler(GinkgoWriter)
+		debugLogger := slog.New(h)
+		itCtx := raw.NewContextWithLogger(ctx, *debugLogger)
+		streamClientConfiguration, err := raw.NewClientConfiguration("rabbitmq-stream://guest:guest@localhost/%2F")
+		Expect(err).ToNot(HaveOccurred())
+
+		By("preparing the environment")
+		streamClient, err := raw.DialConfig(itCtx, streamClientConfiguration)
+		Expect(err).ToNot(HaveOccurred())
+
+		const stream = "e2e-consume-test"
+		// Ensure we don't leak open connection on test failures
+		DeferCleanup(func(ctx SpecContext) error {
+			if streamClient.IsOpen() {
+				_ = streamClient.DeleteStream(ctx, stream)
+				return streamClient.Close(ctx)
+			}
+			return nil
+		})
+
+		Expect(streamClient.IsOpen()).To(BeTrue(), "expected stream client to be open")
+		Expect(streamClient.ExchangeCommandVersions(ctx)).To(Succeed())
+
+		Expect(streamClient.DeclareStream(itCtx, stream, constants.StreamConfiguration{})).To(Succeed())
+
+		const publisherId = 2
+		Expect(
+			streamClient.DeclarePublisher(itCtx, publisherId, "e2e-send-and-receive", stream),
+		).To(Succeed())
+
+		c := streamClient.NotifyPublish(make(chan *raw.PublishConfirm, 100))
+
+		const numMessages = 100
+		for i := 0; i < numMessages; i++ {
+			Expect(
+				streamClient.Send(itCtx, publisherId, wrap[common.PublishingMessager](raw.NewPublishingMessage(uint64(i), &plainTextMessage{messageBody}))),
+			).To(Succeed())
+		}
+
+		var countOfPublishingIds int
+		for confirm := range c {
+			Expect(confirm.PublisherID()).To(BeNumerically("==", publisherId))
+			countOfPublishingIds += len(confirm.PublishingIds())
+			if countOfPublishingIds >= numMessages {
+				break
+			}
+		}
+
+		// Assert number of PublishConfirms matches the number of messages sent
+		Expect(countOfPublishingIds).To(Equal(numMessages))
+
+		By("receiving the expected number of messages")
+		chunks := streamClient.NotifyChunk(make(chan *raw.Chunk, 10))
+		var subscriptionId uint8 = 2
+		Expect(
+			streamClient.Subscribe(itCtx, stream, constants.OffsetTypeFirst, subscriptionId, 10, nil, 0),
+		).To(Succeed())
+
+		var numOfEntries uint16 = 0
+		for chunk := range chunks {
+			numOfEntries += chunk.NumEntries
+			debugLogger.Info("chunk received", "chunk", chunk)
+
+			m := &plainTextMessage{}
+			for i := uint16(0); i < chunk.NumEntries; i++ {
+				x, z := i*104, (i+1)*104
+				Expect(m.UnmarshalBinary(chunk.Messages[x:z])).To(Succeed())
+				Expect(m.body).To(Equal(messageBody))
+			}
+
+			if numOfEntries >= numMessages {
+				break
+			}
+
+			Expect(streamClient.Credit(ctx, subscriptionId, 10)).To(Succeed())
+		}
+
+		By("unsubscribing")
+		Expect(streamClient.Unsubscribe(ctx, subscriptionId)).To(Succeed())
+		//Expect(chunks).To(BeClosed())
+
+		By("cleaning up")
+		Expect(streamClient.DeletePublisher(ctx, publisherId)).To(Succeed())
+		Expect(streamClient.DeleteStream(itCtx, stream)).To(Succeed())
+		Expect(streamClient.Close(itCtx)).To(Succeed())
+	}, SpecTimeout(15*time.Second))
+
 })
 
 func wrap[T any](v T) []T {

pkg/e2e/types_test.go

@@ -3,6 +3,7 @@
 package e2e_test
 
 import (
+	"bytes"
 	"encoding/binary"
 	"io"
 )
@@ -11,13 +12,25 @@ type plainTextMessage struct {
 	body string
 }
 
+func (p *plainTextMessage) UnmarshalBinary(data []byte) error {
+	var dataLen uint32
+	err := binary.Read(bytes.NewReader(data), binary.BigEndian, &dataLen)
+	if err != nil {
+		return err
+	}
+
+	p.body = string(data[4 : dataLen+4])
+
+	return nil
+}
+
 func (p *plainTextMessage) WriteTo(w io.Writer) (n int64, err error) {
 	n = 0
 	err = binary.Write(w, binary.BigEndian, uint32(len(p.body)))
 	if err != nil {
 		return
 	}
-	n += 2
+	n += 4
 
 	n32, err := w.Write([]byte(p.body))
 	n += int64(n32)

pkg/raw/client.go

@@ -286,7 +286,8 @@ func (tc *Client) handleIncoming(ctx context.Context) error {
 				internal.CommandDeclarePublisherResponse,
 				internal.CommandDeletePublisherResponse,
 				internal.CommandCloseResponse,
-				internal.CommandSubscribeResponse:
+				internal.CommandSubscribeResponse,
+				internal.CommandUnsubscribeResponse:
 				createResp := new(internal.SimpleResponse)
 				err = createResp.Read(buffer)
 				if err != nil {
@@ -366,7 +367,7 @@ func (tc *Client) handleIncoming(ctx context.Context) error {
 				case <-ctx.Done():
 					return ctx.Err()
 				case tc.chunkCh <- chunkResponse:
-					log.Debug("sent a subscription chunk", "subscriptionId", chunkResponse.SubscriptionId)
+					log.Debug("received a chunk", "subscriptionId", chunkResponse.SubscriptionId)
 				}
 			case internal.CommandExchangeCommandVersionsResponse:
 				exchangeResponse := new(internal.ExchangeCommandVersionsResponse)
@@ -1071,6 +1072,7 @@ func (tc *Client) Unsubscribe(ctx context.Context, subscriptionId uint8) error {
 	if err != nil {
 		return err
 	}
+	// FIXME: close subscription channel
 	return streamErrorOrNil(unSubscribeResponse.ResponseCode())
 }
 

pkg/raw/stream_suite_test.go

@@ -589,7 +589,7 @@ func (rmq *fakeRabbitMQServer) fakeRabbitMQUnsubscribe(ctx context.Context, subs
 
 	/// there server says ok! :)
 	/// writing the response to the client
-	writeResponse(ctx, rmq, bufio.NewWriter(rmq.connection), internal.CommandCreate)
+	writeResponse(ctx, rmq, bufio.NewWriter(rmq.connection), internal.CommandUnsubscribeResponse)
 }
 
 func (rmq *fakeRabbitMQServer) fakeRabbitMQServerClosesConnection() {