implement direct reply to feature

Signed-off-by: Gabriele Santomaggio <G.santomaggio@gmail.com>

Author: Gsantomaggio

pkg/rabbitmqamqp/amqp_connection.go

@@ -184,18 +184,16 @@ func (a *AmqpConnection) NewConsumer(ctx context.Context, queueName string, opti
 
 	if options != nil && options.isDirectReplyToEnable() {
 		return newConsumer(ctx, a, "", options)
-	} else {
-		destination := &QueueAddress{
-			Queue: queueName,
-		}
-		destinationAdd, err := destination.toAddress()
-		if err != nil {
-			return nil, err
-		}
-		return newConsumer(ctx, a, destinationAdd, options)
-
 	}
 
+	destination := &QueueAddress{
+		Queue: queueName,
+	}
+	destinationAdd, err := destination.toAddress()
+	if err != nil {
+		return nil, err
+	}
+	return newConsumer(ctx, a, destinationAdd, options)
 }
 
 // NewResponder creates a new RPC server that processes requests from the

pkg/rabbitmqamqp/amqp_consumer.go

@@ -101,7 +101,7 @@ func (c *Consumer) Id() string {
 
 func newConsumer(ctx context.Context, connection *AmqpConnection, destinationAdd string, options IConsumerOptions) (*Consumer, error) {
 	id := fmt.Sprintf("consumer-%s", uuid.New().String())
-	if options != nil && options.id() != "" {
+	if options != nil && len(options.id()) > 0 {
 		id = options.id()
 	}
 

pkg/rabbitmqamqp/requester.go

@@ -33,6 +33,7 @@ import (
 //   - `Message` provides a basic AMQP message structure for RPC requests.
 //   - `Publish` sends the request message and returns a channel that will receive
 //     the reply message, or be closed if a timeout occurs or the client is closed.
+//   - `GetReplyQueue` returns the address of the reply queue used by the requester.
 type Requester interface {
 	Close(context.Context) error
 	Message(body []byte) *amqp.Message