rename rpcclient to Requester

and rpcServer to Responder

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

Author: Gsantomaggio

docs/examples/rpc_echo_server/main.go

@@ -11,39 +11,39 @@ import (
 	"github.com/rabbitmq/rabbitmq-amqp-go-client/pkg/rabbitmqamqp"
 )
 
-type echoRpcServer struct {
+type echoResponder struct {
 	conn   *rabbitmqamqp.AmqpConnection
-	server rabbitmqamqp.RpcServer
+	server rabbitmqamqp.Responder
 }
 
-func (s *echoRpcServer) stop(ctx context.Context) {
+func (s *echoResponder) stop(ctx context.Context) {
 	s.server.Close(ctx)
 	s.conn.Close(ctx)
 }
 
-func newEchoRpcServer(conn *rabbitmqamqp.AmqpConnection) *echoRpcServer {
+func newEchoResponder(conn *rabbitmqamqp.AmqpConnection) *echoResponder {
 	_, err := conn.Management().DeclareQueue(context.TODO(), &rabbitmqamqp.QuorumQueueSpecification{
-		Name: rpcServerQueueName,
+		Name: requestQueue,
 	})
 	if err != nil {
 		panic(err)
 	}
-	srv, err := conn.NewRpcServer(context.TODO(), rabbitmqamqp.RpcServerOptions{
-		RequestQueue: rpcServerQueueName,
+	srv, err := conn.NewResponder(context.TODO(), rabbitmqamqp.ResponderOptions{
+		RequestQueue: requestQueue,
 		Handler: func(ctx context.Context, request *amqp.Message) (*amqp.Message, error) {
 			return request, nil
 		},
 	})
 	if err != nil {
 		panic(err)
 	}
-	return &echoRpcServer{
+	return &echoResponder{
 		conn:   conn,
 		server: srv,
 	}
 }
 
-const rpcServerQueueName = "rpc-queue"
+const requestQueue = "go-amqp1.0-request-queue"
 
 func main() {
 	// Dial rabbit for RPC server connection
@@ -52,16 +52,16 @@ func main() {
 		panic(err)
 	}
 
-	srv := newEchoRpcServer(srvConn)
+	srv := newEchoResponder(srvConn)
 
 	// Dial rabbit for RPC client connection
 	clientConn, err := rabbitmqamqp.Dial(context.TODO(), "amqp://localhost:5672", nil)
 	if err != nil {
 		panic(err)
 	}
 
-	rpcClient, err := clientConn.NewRequester(context.TODO(), &rabbitmqamqp.RequesterOptions{
-		RequestQueueName: rpcServerQueueName,
+	requester, err := clientConn.NewRequester(context.TODO(), &rabbitmqamqp.RequesterOptions{
+		RequestQueueName: requestQueue,
 	})
 	if err != nil {
 		panic(err)
@@ -94,7 +94,7 @@ func main() {
 			continue
 		}
 
-		resp, err := rpcClient.Publish(context.TODO(), amqp.NewMessage([]byte(message)))
+		resp, err := requester.Publish(context.TODO(), amqp.NewMessage([]byte(message)))
 		if err != nil {
 			fmt.Printf("Error calling RPC: %v\n", err)
 			continue

pkg/rabbitmqamqp/amqp_connection.go

@@ -192,10 +192,10 @@ func (a *AmqpConnection) NewConsumer(ctx context.Context, queueName string, opti
 	return newConsumer(ctx, a, destinationAdd, options)
 }
 
-// NewRpcServer creates a new RPC server that processes requests from the
+// NewResponder creates a new RPC server that processes requests from the
 // specified queue. The requestQueue in options is mandatory, while other
 // fields are optional and will use defaults if not provided.
-func (a *AmqpConnection) NewRpcServer(ctx context.Context, options RpcServerOptions) (RpcServer, error) {
+func (a *AmqpConnection) NewResponder(ctx context.Context, options ResponderOptions) (Responder, error) {
 	if err := options.validate(); err != nil {
 		return nil, fmt.Errorf("rpc server options validation: %w", err)
 	}
@@ -231,7 +231,7 @@ func (a *AmqpConnection) NewRpcServer(ctx context.Context, options RpcServerOpti
 		replyPostProcessor = defaultReplyPostProcessor
 	}
 
-	server := &amqpRpcServer{
+	server := &amqpResponder{
 		requestHandler:         handler,
 		requestQueue:           options.RequestQueue,
 		publisher:              publisher,
@@ -246,12 +246,12 @@ func (a *AmqpConnection) NewRpcServer(ctx context.Context, options RpcServerOpti
 
 // NewRequester creates a new RPC client that sends requests to the specified queue
 // and receives replies on a dynamically created reply queue.
-func (a *AmqpConnection) NewRequester(ctx context.Context, options *RequesterOptions) (RpcClient, error) {
+func (a *AmqpConnection) NewRequester(ctx context.Context, options *RequesterOptions) (Requester, error) {
 	if options == nil {
 		return nil, fmt.Errorf("options cannot be nil")
 	}
 	if options.RequestQueueName == "" {
-		return nil, fmt.Errorf("requestQueueName is mandatory")
+		return nil, fmt.Errorf("request QueueName is mandatory")
 	}
 
 	// Create publisher for sending requests

pkg/rabbitmqamqp/example_rpc_custom_test.go

@@ -38,7 +38,7 @@ func Example_customCorrelationId() {
 		panic(err)
 	}
 
-	server, err := srvConn.NewRpcServer(context.TODO(), rabbitmqamqp.RpcServerOptions{
+	server, err := srvConn.NewResponder(context.TODO(), rabbitmqamqp.ResponderOptions{
 		RequestQueue: rpcServerQueueNameCustom,
 		Handler: func(ctx context.Context, request *amqp.Message) (*amqp.Message, error) {
 			fmt.Printf("Received: %s\n", request.GetData())

pkg/rabbitmqamqp/example_rpc_test.go

@@ -13,7 +13,7 @@ import (
 
 type echoRpcServer struct {
 	conn   *rabbitmqamqp.AmqpConnection
-	server rabbitmqamqp.RpcServer
+	server rabbitmqamqp.Responder
 }
 
 func (s *echoRpcServer) stop(ctx context.Context) {
@@ -25,7 +25,7 @@ func newEchoRpcServer(conn *rabbitmqamqp.AmqpConnection) *echoRpcServer {
 	conn.Management().DeclareQueue(context.TODO(), &rabbitmqamqp.QuorumQueueSpecification{
 		Name: rpcServerQueueName,
 	})
-	srv, err := conn.NewRpcServer(context.TODO(), rabbitmqamqp.RpcServerOptions{
+	srv, err := conn.NewResponder(context.TODO(), rabbitmqamqp.ResponderOptions{
 		RequestQueue: rpcServerQueueName,
 		Handler: func(ctx context.Context, request *amqp.Message) (*amqp.Message, error) {
 			fmt.Printf("echo: %s\n", request.GetData())

pkg/rabbitmqamqp/requester.go

@@ -12,7 +12,7 @@ import (
 	"github.com/google/uuid"
 )
 
-// RpcClient is an interface for making RPC (Remote Procedure Call) requests over AMQP.
+// Requester is an interface for making RPC (Remote Procedure Call) requests over AMQP.
 // Implementations of this interface should handle the sending of requests and
 // the receiving of corresponding replies, managing correlation IDs and timeouts.
 //
@@ -33,7 +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.
-type RpcClient interface {
+type Requester interface {
 	Close(context.Context) error
 	Message(body []byte) *amqp.Message
 	Publish(context.Context, *amqp.Message) (<-chan *amqp.Message, error)

pkg/rabbitmqamqp/requester_test.go

@@ -10,7 +10,7 @@ import (
 	. "github.com/onsi/gomega"
 )
 
-var _ = Describe("RpcClient", func() {
+var _ = Describe("Requester", func() {
 	var (
 		conn      *AmqpConnection
 		queueName string

pkg/rabbitmqamqp/rpc_server.go pkg/rabbitmqamqp/responder.gopkg/rabbitmqamqp/rpc_server.go renamed to pkg/rabbitmqamqp/responder.go

@@ -6,10 +6,10 @@ import (
 	"sync"
 	"time"
 
-	amqp "github.com/Azure/go-amqp"
+	"github.com/Azure/go-amqp"
 )
 
-// RpcServerHandler is a function that processes a request message and returns a response message.
+// ResponderHandler is a function that processes a request message and returns a response message.
 // If the server wants to send a response to the client, it must return a response message.
 // If the function returns nil, the server will not send a response.
 // If the server does not send a response message, this high level RPC server doesn't make much sense,
@@ -22,9 +22,9 @@ import (
 //	func(ctx context.Context, request *amqp.Message) (*amqp.Message, error) {
 //		return amqp.NewMessage([]byte(fmt.Sprintf("Pong: %s", request.GetData()))), nil
 //	}
-type RpcServerHandler func(ctx context.Context, request *amqp.Message) (*amqp.Message, error)
+type ResponderHandler func(ctx context.Context, request *amqp.Message) (*amqp.Message, error)
 
-var noOpHandler RpcServerHandler = func(_ context.Context, _ *amqp.Message) (*amqp.Message, error) {
+var noOpHandler ResponderHandler = func(_ context.Context, _ *amqp.Message) (*amqp.Message, error) {
 	return nil, nil
 }
 
@@ -54,9 +54,9 @@ var defaultReplyPostProcessor ReplyPostProcessor = func(reply *amqp.Message, cor
 	return reply
 }
 
-// RpcServer is Remote Procedure Call server that receives a message, process them,
+// Responder is Remote Procedure Call server that receives a message, process them,
 // and sends a response.
-type RpcServer interface {
+type Responder interface {
 	// Close the RPC server and its underlying resources.
 	Close(context.Context) error
 	// Pause the server to stop receiving messages.
@@ -65,7 +65,7 @@ type RpcServer interface {
 	Unpause() error
 }
 
-type RpcServerOptions struct {
+type ResponderOptions struct {
 	// RequestQueue is the name of the queue to subscribe to. This queue must be pre-declared.
 	// The RPC server does not declare the queue, it is the responsibility of the caller to declare the queue.
 	//
@@ -86,7 +86,7 @@ type RpcServerOptions struct {
 	// 		}
 	//
 	// Mandatory.
-	Handler RpcServerHandler
+	Handler ResponderHandler
 	// CorrectionIdExtractor is a function that extracts a correction ID from the request message.
 	// The returned value should be an AMQP type that can be binary encoded.
 	//
@@ -121,17 +121,17 @@ type RpcServerOptions struct {
 	ReplyPostProcessor ReplyPostProcessor
 }
 
-func (r *RpcServerOptions) validate() error {
+func (r *ResponderOptions) validate() error {
 	if r.RequestQueue == "" {
 		return fmt.Errorf("requestQueue is mandatory")
 	}
 	return nil
 }
 
-type amqpRpcServer struct {
+type amqpResponder struct {
 	// TODO: handle state changes for reconnections
 	mu                     sync.Mutex
-	requestHandler         RpcServerHandler
+	requestHandler         ResponderHandler
 	requestQueue           string
 	publisher              *Publisher
 	consumer               *Consumer
@@ -145,7 +145,7 @@ type amqpRpcServer struct {
 // are closed gracefully and only once, even if Close is called multiple times.
 // The provided context (ctx) controls the timeout for the close operation, ensuring the operation
 // does not exceed the context's deadline.
-func (a *amqpRpcServer) Close(ctx context.Context) error {
+func (a *amqpResponder) Close(ctx context.Context) error {
 	// TODO: wait for unsettled messages
 	a.closer.Do(func() {
 		a.mu.Lock()
@@ -168,14 +168,14 @@ func (a *amqpRpcServer) Close(ctx context.Context) error {
 	return nil
 }
 
-func (a *amqpRpcServer) Pause() {
+func (a *amqpResponder) Pause() {
 	err := a.consumer.pause(context.Background())
 	if err != nil {
 		Warn("Did not pause consumer", "error", err)
 	}
 }
 
-func (a *amqpRpcServer) Unpause() error {
+func (a *amqpResponder) Unpause() error {
 	a.mu.Lock()
 	if a.closed {
 		a.mu.Unlock()
@@ -190,7 +190,7 @@ func (a *amqpRpcServer) Unpause() error {
 	return nil
 }
 
-func (a *amqpRpcServer) handle() {
+func (a *amqpResponder) handle() {
 	/*
 		The RPC server has the following behavior:
 		- when receiving a message request:
@@ -259,12 +259,12 @@ func (a *amqpRpcServer) handle() {
 	}
 }
 
-func (a *amqpRpcServer) isClosed() bool {
+func (a *amqpResponder) isClosed() bool {
 	a.mu.Lock()
 	defer a.mu.Unlock()
 	return a.closed
 }
 
-func (a *amqpRpcServer) issueCredits(credits uint32) error {
+func (a *amqpResponder) issueCredits(credits uint32) error {
 	return a.consumer.issueCredits(credits)
 }

pkg/rabbitmqamqp/rpc_server_test.go pkg/rabbitmqamqp/responder_test.gopkg/rabbitmqamqp/rpc_server_test.go renamed to pkg/rabbitmqamqp/responder_test.go

@@ -12,7 +12,7 @@ import (
 	"github.com/onsi/gomega/gbytes"
 )
 
-var _ = Describe("RpcServer", func() {
+var _ = Describe("Responder", func() {
 	var (
 		conn         *AmqpConnection
 		requestQueue string
@@ -48,7 +48,7 @@ var _ = Describe("RpcServer", func() {
 			requestPublisher.Close(ctx)
 		})
 
-		server, err := conn.NewRpcServer(context.Background(), RpcServerOptions{
+		server, err := conn.NewResponder(context.Background(), ResponderOptions{
 			RequestQueue: requestQueue,
 			Handler: func(ctx context.Context, request *amqp.Message) (*amqp.Message, error) {
 				if request.Properties == nil {
@@ -92,7 +92,7 @@ var _ = Describe("RpcServer", func() {
 
 	It("stops the handler when the RPC server closes", func(ctx SpecContext) {
 		// setup
-		server, err := conn.NewRpcServer(context.Background(), RpcServerOptions{
+		server, err := conn.NewResponder(context.Background(), ResponderOptions{
 			RequestQueue: requestQueue,
 		})
 		Expect(err).ToNot(HaveOccurred())
@@ -137,7 +137,7 @@ var _ = Describe("RpcServer", func() {
 			reply.ApplicationProperties["test"] = "success"
 			return reply
 		}
-		server, err := conn.NewRpcServer(context.Background(), RpcServerOptions{
+		server, err := conn.NewResponder(context.Background(), ResponderOptions{
 			RequestQueue: requestQueue,
 			Handler: func(ctx context.Context, request *amqp.Message) (*amqp.Message, error) {
 				m := amqp.NewMessage(request.GetData())

pkg/rabbitmqamqp/rpc_e2e_test.go

@@ -30,8 +30,8 @@ var _ = Describe("RPC E2E", Label("e2e"), func() {
 	var (
 		clientConn   *rabbitmqamqp.AmqpConnection
 		serverConn   *rabbitmqamqp.AmqpConnection
-		rpcClient    rabbitmqamqp.RpcClient
-		rpcServer    rabbitmqamqp.RpcServer
+		rpcClient    rabbitmqamqp.Requester
+		rpcServer    rabbitmqamqp.Responder
 		rpcQueueName string
 	)
 
@@ -73,7 +73,7 @@ var _ = Describe("RPC E2E", Label("e2e"), func() {
 			RequestQueueName: rpcQueueName,
 		})
 		Ω(err).ShouldNot(HaveOccurred())
-		rpcServer, err = serverConn.NewRpcServer(ctx, rabbitmqamqp.RpcServerOptions{
+		rpcServer, err = serverConn.NewResponder(ctx, rabbitmqamqp.ResponderOptions{
 			RequestQueue: rpcQueueName,
 			Handler: func(ctx context.Context, request *amqp.Message) (*amqp.Message, error) {
 				m.Lock()