Added request-reply support for commands

components/cqrs/command_config.go

@@ -18,6 +18,12 @@ type CommandConfig struct {
 	OnSend   OnCommandSendFn
 	OnHandle OnCommandHandleFn
 
+	// RequestReplyEnabled enables request-reply pattern for commands.
+	// Reply is sent **just** from the CommandBus.SendAndWait method.
+	// This configuration doesn't affect CommandBus.Send method.
+	RequestReplyEnabled bool
+	RequestReplyBackend RequestReplyBackend
+
 	Marshaler CommandEventMarshaler
 	Logger    watermill.LoggerAdapter
 
@@ -45,6 +51,10 @@ func (c CommandConfig) Validate() error {
 		err = stdErrors.Join(err, errors.New("missing Marshaler"))
 	}
 
+	if c.RequestReplyEnabled && c.RequestReplyBackend == nil {
+		err = stdErrors.Join(err, errors.New("missing RequestReply.Backend"))
+	}
+
 	return err
 }
 

components/cqrs/command_processor.go

@@ -1,6 +1,7 @@
 package cqrs
 
 import (
+	stdErrors "errors"
 	"fmt"
 
 	"github.com/pkg/errors"
@@ -188,10 +189,28 @@ func (p CommandProcessor) routerHandlerFunc(handler CommandHandler, logger water
 			Message: msg,
 		})
 
+		var replyErr error
+		// todo: test
+		if p.config.RequestReplyEnabled {
+			replyErr = p.config.RequestReplyBackend.OnCommandProcessed(msg, cmd, err)
+		}
+
+
 		if p.config.AckCommandHandlingErrors && err != nil {
+			// we want to nack if we are using request-reply,
+			// and we failed to send information about failure
+			// todo: test
+			if replyErr != nil {
+				return replyErr
+			}
+
 			logger.Error("Error when handling command", err, nil)
 			return nil
+		} else if replyErr != nil {
+			// todo: test
+			err = stdErrors.Join(err, replyErr)
 		}
+
 		if err != nil {
 			logger.Debug("Error when handling command", watermill.LogFields{"err": err})
 			return err

components/cqrs/command_request_reply.go

@@ -0,0 +1,40 @@
+package cqrs
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/ThreeDotsLabs/watermill/message"
+)
+
+type RequestReplyBackend interface {
+	ModifyCommandMessageBeforePublish(cmdMsg *message.Message, command any) error
+
+	ListenForReply(ctx context.Context, cmdMsg *message.Message, cmd any) (<-chan CommandReply, error)
+
+	OnCommandProcessed(cmdMsg *message.Message, cmd any, handleErr error) error
+}
+
+// ReplyTimeoutError is returned when the reply timeout is exceeded.
+type ReplyTimeoutError struct {
+	Duration time.Duration
+	Err      error
+}
+
+func (e ReplyTimeoutError) Error() string {
+	return fmt.Sprintf("reply timeout after %s: %s", e.Duration, e.Err)
+}
+
+// CommandHandlerError is returned when the command handler returns an error.
+type CommandHandlerError struct {
+	Err error
+}
+
+func (e CommandHandlerError) Error() string {
+	return e.Err.Error()
+}
+
+func (e CommandHandlerError) Unwrap() error {
+	return e.Err
+}

components/cqrs/command_request_reply_bus.go

@@ -0,0 +1,51 @@
+package cqrs
+
+import (
+	"context"
+
+	"github.com/ThreeDotsLabs/watermill/message"
+	"github.com/pkg/errors"
+)
+
+type CommandReply struct {
+	// HandlerErr contains the error returned by the command handler or by RequestReplyBackend if sending reply failed.
+	//
+	// If error from handler is returned, CommandHandlerError is returned.
+	// If listening for reply timed out, HandlerErr is ReplyTimeoutError.
+	// If processing was successful, HandlerErr is nil.
+	Err error
+
+	// ReplyMsg contains the reply message from the command handler.
+	// Warning: ReplyMsg is nil if timeout occurred.
+	ReplyMsg *message.Message
+}
+
+// todo: test
+// todo: add cancel func?
+// SendAndWait sends command to the command bus and waits for the command execution.
+func (c CommandBus) SendAndWait(ctx context.Context, cmd interface{}) (<-chan CommandReply, error) {
+	if !c.config.RequestReplyEnabled {
+		return nil, errors.New("RequestReply is not enabled in config")
+	}
+
+	msg, topicName, err := c.newMessage(ctx, cmd)
+	if err != nil {
+		return nil, err
+	}
+
+	if err := c.config.RequestReplyBackend.ModifyCommandMessageBeforePublish(msg, cmd); err != nil {
+		return nil, errors.Wrap(err, "cannot modify command message before publish")
+	}
+
+	// todo: wait for 1 reply by default?
+	replyChan, err := c.config.RequestReplyBackend.ListenForReply(ctx, msg, cmd)
+	if err != nil {
+		return nil, errors.Wrap(err, "cannot listen for reply")
+	}
+
+	if err := c.publisher.Publish(topicName, msg); err != nil {
+		return nil, err
+	}
+
+	return replyChan, nil
+}