feat: listener should inject the context into the context so that it …

queue/listener.go

@@ -115,11 +115,14 @@ func (l *listener) startWorker(ctx context.Context, messages <-chan *message.Mes
 }
 
 func (l *listener) handleMessage(ctx context.Context, msg *message.Message) {
-	l.logger.WithField("message_uuid", msg.UUID).Debugf("queue listener handling message")
 	ctx = otel.GetTextMapPropagator().Extract(ctx, propagation.MapCarrier(msg.Metadata))
+	ctx = logging.ContextWithLogger(ctx, l.logger)
+	logger := logging.FromContext(ctx)
+
+	logger.WithField("message_uuid", msg.UUID).Debugf("queue listener handling message")
 	err := l.callbackFn(ctx, msg.Metadata, msg.Payload)
 	if err != nil {
-		l.logger.WithField("message_uuid", msg.UUID).WithField("err", err.Error()).Errorf("queue listener failed to process message")
+		logger.WithField("message_uuid", msg.UUID).WithField("err", err.Error()).Errorf("queue listener failed to process message")
 		msg.Nack()
 		return
 	}

queue/listener_test.go

@@ -1,10 +1,13 @@
 package queue_test
 
 import (
+	"bytes"
 	"context" //nolint: gosec
 	"os"
 	"testing"
 
+	"github.com/ThreeDotsLabs/watermill/message"
+	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
@@ -27,3 +30,38 @@ func TestNewListenerInvalidCallback(t *testing.T) {
 	assert.ErrorContains(t, err, "callback")
 	assert.Nil(t, listener)
 }
+
+func TestHandleMessageInjectsLoggerInContext(t *testing.T) {
+	// Create a logger that writes to a buffer so we can verify it's the one in context
+	var buf bytes.Buffer
+	l := logrus.New()
+	l.SetOutput(&buf)
+	l.SetLevel(logrus.DebugLevel)
+	logger := logging.NewLogrus(l)
+
+	done := make(chan struct{})
+	callback := func(ctx context.Context, meta map[string]string, msg []byte) error {
+		// Log using the logger from context — if ContextWithLogger was called,
+		// this writes to our buffer; if not, it writes to stderr (default fallback)
+		logging.FromContext(ctx).Infof("hello from callback")
+		close(done)
+		return nil
+	}
+
+	listener, err := queue.NewListener(logger, callback, 1)
+	require.NoError(t, err)
+
+	ch := make(chan *message.Message, 1)
+	ch <- message.NewMessage("test-uuid", []byte("test-payload"))
+
+	ctx, cancel := context.WithCancel(context.Background())
+	listener.Listen(ctx, ch)
+
+	<-done
+	cancel()
+	<-listener.Done()
+
+	// The injected logger writes to buf; the default fallback writes to stderr.
+	// If our buffer contains the callback message, the logger was properly injected.
+	assert.Contains(t, buf.String(), "hello from callback")
+}