implement direct reply to feature

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

Author: Gsantomaggio

docs/examples/rpc_echo_server/main.go

@@ -52,9 +52,12 @@ func main() {
 	if err != nil {
 		panic(err)
 	}
+	_ = srvConn.Management().DeleteQueue(context.TODO(), requestQueue)
 
 	srv := newEchoResponder(srvConn)
+	reply, _ := srv.server.GetRequestQueue()
 
+	fmt.Printf("request queue %s \n", reply)
 	// Dial rabbit for RPC client connection
 	clientConn, err := rabbitmqamqp.Dial(context.TODO(), "amqp://localhost:5672", nil)
 	if err != nil {
@@ -63,11 +66,15 @@ func main() {
 
 	requester, err := clientConn.NewRequester(context.TODO(), &rabbitmqamqp.RequesterOptions{
 		RequestQueueName: requestQueue,
-		DirectReplyTo:    true,
+		// Enable Direct Reply To feature
+		// see: https://www.rabbitmq.com/direct-reply-to.html
+		DirectReplyTo: true,
 	})
 	if err != nil {
 		panic(err)
 	}
+	reply, _ = requester.GetReplyQueue()
+	fmt.Printf("replyTo to %s \n", reply)
 
 	// Set up a channel to listen for OS signals
 	sigs := make(chan os.Signal, 1)
@@ -95,7 +102,6 @@ func main() {
 		if message == "" {
 			continue
 		}
-
 		resp, err := requester.Publish(context.TODO(), amqp.NewMessage([]byte(message)))
 		if err != nil {
 			fmt.Printf("Error calling RPC: %v\n", err)

pkg/rabbitmqamqp/requester.go

@@ -37,6 +37,7 @@ type Requester interface {
 	Close(context.Context) error
 	Message(body []byte) *amqp.Message
 	Publish(context.Context, *amqp.Message) (<-chan *amqp.Message, error)
+	GetReplyQueue() (string, error)
 }
 
 // CorrelationIdSupplier is an interface for providing correlation IDs for RPC requests.
@@ -224,6 +225,10 @@ func (a *amqpRequester) Publish(ctx context.Context, message *amqp.Message) (<-c
 	return ch, nil
 }
 
+func (a *amqpRequester) GetReplyQueue() (string, error) {
+	return a.consumer.GetQueue()
+}
+
 func (a *amqpRequester) isClosed() bool {
 	a.mu.Lock()
 	defer a.mu.Unlock()

pkg/rabbitmqamqp/responder.go

@@ -63,6 +63,8 @@ type Responder interface {
 	Pause()
 	// Unpause requests to receive messages again.
 	Unpause() error
+
+	GetRequestQueue() (string, error)
 }
 
 type ResponderOptions struct {
@@ -190,6 +192,10 @@ func (a *amqpResponder) Unpause() error {
 	return nil
 }
 
+func (a *amqpResponder) GetRequestQueue() (string, error) {
+	return a.consumer.GetQueue()
+}
+
 func (a *amqpResponder) handle() {
 	/*
 		The RPC server has the following behavior: