Return errors from queue message

Author: hendoxc

pkg/chipingress/batch/client.go

@@ -2,6 +2,7 @@ package batch
 
 import (
 	"context"
+	"errors"
 	"sync"
 	"time"
 
@@ -118,10 +119,11 @@ func (b *Client) Stop() {
 // The callback receives an error parameter (nil on success).
 // Callbacks are invoked from goroutines
 // Returns immediately with no blocking - drops message if channel is full.
-// Returns true if message was queued, false if it was dropped.
-func (b *Client) QueueMessage(event *chipingress.CloudEventPb, callback func(error)) bool {
+// Returns an error if the message was dropped.
+func (b *Client) QueueMessage(event *chipingress.CloudEventPb, callback func(error)) error {
+
 	if event == nil {
-		return false
+		return nil
 	}
 
 	msg := &messageWithCallback{
@@ -130,10 +132,12 @@ func (b *Client) QueueMessage(event *chipingress.CloudEventPb, callback func(err
 	}
 
 	select {
+	case <-b.shutdownChan:
+		return errors.New("client is shutdown")
 	case b.messageBuffer <- msg:
-		return true
+		return nil
 	default:
-		return false
+		return errors.New("message buffer is full")
 	}
 }
 

pkg/chipingress/batch/client_test.go

@@ -338,8 +338,8 @@ func TestStart(t *testing.T) {
 
 		queued1 := client.QueueMessage(&chipingress.CloudEventPb{Id: "test-id-1", Source: "test-source", Type: "test.event.type"}, nil)
 		queued2 := client.QueueMessage(&chipingress.CloudEventPb{Id: "test-id-2", Source: "test-source", Type: "test.event.type"}, nil)
-		require.True(t, queued1)
-		require.True(t, queued2)
+		require.NoError(t, queued1)
+		require.NoError(t, queued2)
 
 		select {
 		case <-done:
@@ -813,4 +813,35 @@ func TestStop(t *testing.T) {
 		client.Stop()
 		client.Stop()
 	})
+
+	t.Run("QueueMessage returns error after Stop", func(t *testing.T) {
+		mockClient := mocks.NewClient(t)
+		client, err := NewBatchClient(mockClient, WithBatchSize(10))
+		require.NoError(t, err)
+
+		ctx, cancel := context.WithCancel(t.Context())
+		defer cancel()
+
+		client.Start(ctx)
+
+		// Queue message before stop - should succeed
+		err = client.QueueMessage(&chipingress.CloudEventPb{
+			Id:     "test-id-1",
+			Source: "test-source",
+			Type:   "test.event.type",
+		}, nil)
+		require.NoError(t, err)
+
+		// Stop the client
+		client.Stop()
+
+		// Queue message after stop - should fail
+		err = client.QueueMessage(&chipingress.CloudEventPb{
+			Id:     "test-id-2",
+			Source: "test-source",
+			Type:   "test.event.type",
+		}, nil)
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "shutdown")
+	})
 }