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

@@ -60,7 +60,7 @@ func main() {
 		panic(err)
 	}
 
-	rpcClient, err := clientConn.NewRpcClient(context.TODO(), &rabbitmqamqp.RpcClientOptions{
+	rpcClient, err := clientConn.NewRequester(context.TODO(), &rabbitmqamqp.RequesterOptions{
 		RequestQueueName: rpcServerQueueName,
 	})
 	if err != nil {

pkg/rabbitmqamqp/amqp_connection.go

@@ -244,9 +244,9 @@ func (a *AmqpConnection) NewRpcServer(ctx context.Context, options RpcServerOpti
 	return server, nil
 }
 
-// NewRpcClient creates a new RPC client that sends requests to the specified queue
+// 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) NewRpcClient(ctx context.Context, options *RpcClientOptions) (RpcClient, error) {
+func (a *AmqpConnection) NewRequester(ctx context.Context, options *RequesterOptions) (RpcClient, error) {
 	if options == nil {
 		return nil, fmt.Errorf("options cannot be nil")
 	}
@@ -305,7 +305,7 @@ func (a *AmqpConnection) NewRpcClient(ctx context.Context, options *RpcClientOpt
 		correlationIdExtractor = defaultReplyCorrelationIdExtractor
 	}
 
-	client := &amqpRpcClient{
+	client := &amqpRequester{
 		requestQueue:           requestQueue,
 		replyToQueue:           &QueueAddress{Queue: replyQueueName},
 		publisher:              publisher,

pkg/rabbitmqamqp/example_rpc_custom_test.go

@@ -70,7 +70,7 @@ func Example_customCorrelationId() {
 	}
 	defer clientConn.Close(context.Background())
 
-	rpcClient, err := clientConn.NewRpcClient(context.TODO(), &rabbitmqamqp.RpcClientOptions{
+	rpcClient, err := clientConn.NewRequester(context.TODO(), &rabbitmqamqp.RequesterOptions{
 		RequestQueueName:      rpcServerQueueNameCustom,
 		CorrelationIdSupplier: &customCorrelationIDSupplier{},
 		CorrelationIdExtractor: func(message *amqp.Message) any {

pkg/rabbitmqamqp/example_rpc_test.go

@@ -58,7 +58,7 @@ func Example() {
 		panic(err)
 	}
 
-	rpcClient, err := clientConn.NewRpcClient(context.TODO(), &rabbitmqamqp.RpcClientOptions{
+	rpcClient, err := clientConn.NewRequester(context.TODO(), &rabbitmqamqp.RequesterOptions{
 		RequestQueueName: rpcServerQueueName,
 	})
 	if err != nil {

pkg/rabbitmqamqp/rpc_client.go pkg/rabbitmqamqp/requester.gopkg/rabbitmqamqp/rpc_client.go renamed to pkg/rabbitmqamqp/requester.go

@@ -86,16 +86,16 @@ var defaultReplyCorrelationIdExtractor CorrelationIdExtractor = func(message *am
 // address for the reply queue. The function must return the modified message.
 //
 // The default `RequestPostProcessor` implementation (used when `RequestPostProcessor`
-// is not explicitly set in `RpcClientOptions`) performs the following:
+// is not explicitly set in `RequesterOptions`) performs the following:
 //   - Assigns the `correlationID` to the `MessageID` property of the `amqp.Message`.
 //   - Sets the `ReplyTo` message property to a client-generated exclusive auto-delete queue.
 type RequestPostProcessor func(request *amqp.Message, correlationID any) *amqp.Message
 
 var DefaultRpcRequestTimeout = 30 * time.Second
 
-// RpcClientOptions is a struct that contains the options for the RPC client.
+// RequesterOptions is a struct that contains the options for the RPC client.
 // It is used to configure the RPC client.
-type RpcClientOptions struct {
+type RequesterOptions struct {
 	// The name of the queue to send requests to. This queue must exist.
 	//
 	// Mandatory.
@@ -130,7 +130,7 @@ type outstandingRequest struct {
 	// err    error
 }
 
-type amqpRpcClient struct {
+type amqpRequester struct {
 	requestQueue           ITargetAddress
 	replyToQueue           ITargetAddress
 	publisher              *Publisher
@@ -149,7 +149,7 @@ type amqpRpcClient struct {
 // Close shuts down the RPC client, closing its underlying publisher and consumer.
 // It ensures that all pending requests are cleaned up by closing their respective
 // channels. This method is safe to call multiple times.
-func (a *amqpRpcClient) Close(ctx context.Context) error {
+func (a *amqpRequester) Close(ctx context.Context) error {
 	var err error
 	a.closer.Do(func() {
 		a.mu.Lock()
@@ -173,7 +173,7 @@ func (a *amqpRpcClient) Close(ctx context.Context) error {
 	return err
 }
 
-func (a *amqpRpcClient) Message(body []byte) *amqp.Message {
+func (a *amqpRequester) Message(body []byte) *amqp.Message {
 	return amqp.NewMessage(body)
 }
 
@@ -184,9 +184,9 @@ func (a *amqpRpcClient) Message(body []byte) *amqp.Message {
 // an `outstandingRequest` is created and stored, and a channel is returned
 // for the reply. The channel will be closed if the request times out or the
 // client is closed before a reply is received.
-func (a *amqpRpcClient) Publish(ctx context.Context, message *amqp.Message) (<-chan *amqp.Message, error) {
+func (a *amqpRequester) Publish(ctx context.Context, message *amqp.Message) (<-chan *amqp.Message, error) {
 	if a.isClosed() {
-		return nil, fmt.Errorf("rpc client is closed")
+		return nil, fmt.Errorf("requester is closed")
 	}
 	replyTo, err := a.replyToQueue.toAddress()
 	if err != nil {
@@ -220,7 +220,7 @@ func (a *amqpRpcClient) Publish(ctx context.Context, message *amqp.Message) (<-c
 	return ch, nil
 }
 
-func (a *amqpRpcClient) isClosed() bool {
+func (a *amqpRequester) isClosed() bool {
 	a.mu.Lock()
 	defer a.mu.Unlock()
 	return a.closed
@@ -231,7 +231,7 @@ func (a *amqpRpcClient) isClosed() bool {
 // If a request's `sentAt` timestamp is older than the `requestTimeout`,
 // its channel is closed, and the request is removed from `pendingRequests`.
 // The goroutine exits when the `done` channel is closed, typically when the client is closed.
-func (a *amqpRpcClient) requestTimeoutTask() {
+func (a *amqpRequester) requestTimeoutTask() {
 	t := time.NewTicker(a.requestTimeout)
 	defer t.Stop()
 	for {
@@ -260,7 +260,7 @@ func (a *amqpRpcClient) requestTimeoutTask() {
 // corresponding request's channel, and the request is removed from `pendingRequests`.
 // If no match is found, the message is requeued. The goroutine exits when the `done`
 // channel is closed, typically when the client is closed.
-func (a *amqpRpcClient) messageReceivedHandler() {
+func (a *amqpRequester) messageReceivedHandler() {
 	for {
 		select {
 		case <-a.done:

pkg/rabbitmqamqp/rpc_client_test.go pkg/rabbitmqamqp/requester_test.gopkg/rabbitmqamqp/rpc_client_test.go renamed to pkg/rabbitmqamqp/requester_test.go

@@ -85,7 +85,7 @@ var _ = Describe("RpcClient", func() {
 		// Server goroutine to handle incoming requests
 		go pongRpcServer(ctx, publisher, consumer)
 
-		client, err := conn.NewRpcClient(ctx, &RpcClientOptions{
+		client, err := conn.NewRequester(ctx, &RequesterOptions{
 			RequestQueueName: queueName,
 		})
 		Ω(err).ShouldNot(HaveOccurred())
@@ -110,7 +110,7 @@ var _ = Describe("RpcClient", func() {
 
 	It("uses a custom correlation id extractor and post processor", func(ctx SpecContext) {
 		go pongRpcServer(ctx, publisher, consumer)
-		client, err := conn.NewRpcClient(ctx, &RpcClientOptions{
+		client, err := conn.NewRequester(ctx, &RequesterOptions{
 			RequestQueueName: queueName,
 			CorrelationIdExtractor: func(message *amqp.Message) any {
 				return message.ApplicationProperties["correlationId"]

pkg/rabbitmqamqp/rpc_e2e_test.go

@@ -69,7 +69,7 @@ var _ = Describe("RPC E2E", Label("e2e"), func() {
 		m := sync.Mutex{}
 		messagesReceivedByServer := 0
 		var err error
-		rpcClient, err = clientConn.NewRpcClient(ctx, &rabbitmqamqp.RpcClientOptions{
+		rpcClient, err = clientConn.NewRequester(ctx, &rabbitmqamqp.RequesterOptions{
 			RequestQueueName: rpcQueueName,
 		})
 		Ω(err).ShouldNot(HaveOccurred())
@@ -147,11 +147,11 @@ func ExampleRpcClient() {
 
 	// Create RPC client options
 	// RequestQueueName is mandatory. The queue must exist.
-	options := rabbitmqamqp.RpcClientOptions{
+	options := rabbitmqamqp.RequesterOptions{
 		RequestQueueName: "rpc-queue",
 	}
 	// Create a new RPC client
-	rpcClient, err := conn.NewRpcClient(context.TODO(), &options)
+	rpcClient, err := conn.NewRequester(context.TODO(), &options)
 	if err != nil {
 		panic(err)
 	}
@@ -200,13 +200,13 @@ func ExampleRpcClient_customCorrelationID() {
 	defer conn.Close(context.TODO())
 
 	// Create RPC client options
-	options := rabbitmqamqp.RpcClientOptions{
+	options := rabbitmqamqp.RequesterOptions{
 		RequestQueueName:      "rpc-queue", // the queue must exist
 		CorrelationIdSupplier: &fooCorrelationIdSupplier{},
 	}
 
 	// Create a new RPC client
-	rpcClient, _ := conn.NewRpcClient(context.TODO(), &options)
+	rpcClient, _ := conn.NewRequester(context.TODO(), &options)
 	pendingRequestCh, _ := rpcClient.Publish(context.TODO(), rpcClient.Message([]byte("hello world")))
 	replyFromServer := <-pendingRequestCh
 	fmt.Printf("reply correlation ID: %s\n", replyFromServer.Properties.CorrelationID)