RPC high level constructs (#59)

* Set Go toolchain

* Initial version of RPC server

The happy path works. The server can connect to a queue and responds
messages if the request's replyTo property is set.

* Bump Ginkgo Gomega and use floating 4.1 image in test

* Add VS Code gitignores

* Add correlationID extractor and post-processor

RPC Tests were refactored to extract common setup code into a before
each node.

* RPC server test for custom functionality

A test that verifies that custom correction ID and custom post processor
are called by the server.

* Add code docs for RPC server functions

The test for RpcServer.Close() was returning too soon. The simplest
solution is an ugly sleep of 1 second.

* Add functions to handle consumer credits manually

RPC server construct provides an utility to pause the message
processing. This relies on the ability to "drain" credits, and issue new
credits at a later point to unpause. This commit exposes that
functionality internally.

* Implement pause/unpause in RPC server

The RPC server now can be constructred from the AMQP connection. Tests
were refactored to use this contructor, instead of manually setting up
publishers and consumers.

Because setting the reply-to address in the post-processor forces the
users of custom post-processors to remember/add a line to set the
reply-to address always. By setting the reply-to prior to the
post-processor, the user has the option to modify the reply-to before
sending, if desired. By setting the reply-to prior to post-processor, we
free up some cognitive load on the users who wish to use a custom
post-processor.

* Add E2E test for RPC constructs

* Refactor pointer-to functions: one generic function to rule them all
* The automatic reconnection  was racy. It could try to open a connection
at the same time that `Close()` was called on `AmqpConnection`. This
required introducing an internal field to determine when the connection
is "closed", and a specific error to signal in the reconnection that the
connection is "closed" after calling `Close()`.
* Add examples and fix typo
Signed-off-by: Gabriele Santomaggio <[email protected]>

---------

Signed-off-by: Gabriele Santomaggio <[email protected]>
Co-authored-by: Gabriele Santomaggio <[email protected]>

Author: Zerpet

.ci/ubuntu/gha-setup.sh

@@ -7,7 +7,7 @@ set -o xtrace
 script_dir="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"
 readonly script_dir
 echo "[INFO] script_dir: '$script_dir'"
-readonly rabbitmq_image=rabbitmq:4.1.0-beta.4-management-alpine
+readonly rabbitmq_image=rabbitmq:4.1-management-alpine
 
 
 readonly docker_name_prefix='rabbitmq-amqp-go-client'

.github/workflows/build-test.yaml

@@ -9,7 +9,9 @@ jobs:
     strategy:
       fail-fast: true
       matrix:
-        go: [ '1.22']
+        go:
+        - stable
+        - oldstable
     steps:
       - uses: actions/checkout@v4
       - uses: actions/setup-go@v5

.gitignore

@@ -27,3 +27,15 @@ go.work.sum
 coverage.txt
 .DS_Store
 .ci/ubuntu/log/rabbitmq.log
+
+# Visual Studio Code
+.vscode/*
+!.vscode/settings.json
+!.vscode/tasks.json
+!.vscode/launch.json
+!.vscode/extensions.json
+!.vscode/*.code-snippets
+!*.code-workspace
+
+# Built Visual Studio Code Extensions
+*.vsix

docs/examples/README.md

@@ -9,4 +9,5 @@
 - [Video](video) - From the YouTube tutorial [AMQP 1.0 with Golang](https://youtu.be/iR1JUFh3udI)
 - [TLS](tls) - An example of how to use TLS with the AMQP 1.0 client.
 - [Advanced Settings](advanced_settings) - An example of how to use the advanced connection settings of the AMQP 1.0 client.
-- [Broadcast](broadcast) - An example of how to use fanout to broadcast messages to multiple auto-deleted queues.
+- [Broadcast](broadcast) - An example of how to use fanout to broadcast messages to multiple auto-deleted queues.
+- [RPC Echo](rpc_echo_server) - An example of how to implement RPC with the AMQP 1.0 client.

docs/examples/rpc_echo_server/main.go

@@ -0,0 +1,109 @@
+package main
+
+import (
+	"bufio"
+	"context"
+	"fmt"
+	"os"
+	"os/signal"
+
+	"github.com/Azure/go-amqp"
+	"github.com/rabbitmq/rabbitmq-amqp-go-client/pkg/rabbitmqamqp"
+)
+
+type echoRpcServer struct {
+	conn   *rabbitmqamqp.AmqpConnection
+	server rabbitmqamqp.RpcServer
+}
+
+func (s *echoRpcServer) stop(ctx context.Context) {
+	s.server.Close(ctx)
+	s.conn.Close(ctx)
+}
+
+func newEchoRpcServer(conn *rabbitmqamqp.AmqpConnection) *echoRpcServer {
+	_, err := conn.Management().DeclareQueue(context.TODO(), &rabbitmqamqp.QuorumQueueSpecification{
+		Name: rpcServerQueueName,
+	})
+	if err != nil {
+		panic(err)
+	}
+	srv, err := conn.NewRpcServer(context.TODO(), rabbitmqamqp.RpcServerOptions{
+		RequestQueue: rpcServerQueueName,
+		Handler: func(ctx context.Context, request *amqp.Message) (*amqp.Message, error) {
+			return request, nil
+		},
+	})
+	if err != nil {
+		panic(err)
+	}
+	return &echoRpcServer{
+		conn:   conn,
+		server: srv,
+	}
+}
+
+const rpcServerQueueName = "rpc-queue"
+
+func main() {
+	// Dial rabbit for RPC server connection
+	srvConn, err := rabbitmqamqp.Dial(context.TODO(), "amqp://localhost:5672", nil)
+	if err != nil {
+		panic(err)
+	}
+
+	srv := newEchoRpcServer(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.NewRpcClient(context.TODO(), &rabbitmqamqp.RpcClientOptions{
+		RequestQueueName: rpcServerQueueName,
+	})
+	if err != nil {
+		panic(err)
+	}
+
+	// Set up a channel to listen for OS signals
+	sigs := make(chan os.Signal, 1)
+	signal.Notify(sigs, os.Interrupt) // Listen for Ctrl+C
+
+	// Goroutine to handle graceful shutdown
+	go func() {
+		<-sigs // Wait for Ctrl+C
+		fmt.Println("\nReceived Ctrl+C, gracefully shutting down...")
+		srv.stop(context.TODO())
+		_ = clientConn.Close(context.TODO())
+		_ = srvConn.Close(context.TODO())
+		os.Exit(0)
+	}()
+
+	reader := bufio.NewReader(os.Stdin)
+	fmt.Println("Type a message and press Enter to send (Ctrl+C to quit):")
+
+	for {
+		fmt.Print("Enter message: ")
+		input, _ := reader.ReadString('\n')
+		// Remove newline character from input
+		message := input[:len(input)-1]
+
+		if message == "" {
+			continue
+		}
+
+		resp, err := rpcClient.Publish(context.TODO(), amqp.NewMessage([]byte(message)))
+		if err != nil {
+			fmt.Printf("Error calling RPC: %v\n", err)
+			continue
+		}
+		m, ok := <-resp
+		if !ok {
+			fmt.Println("timed out waiting for response")
+			continue
+		}
+		fmt.Printf("response: %s\n", m.GetData())
+	}
+}

go.mod

@@ -2,22 +2,25 @@ module github.com/rabbitmq/rabbitmq-amqp-go-client
 
 go 1.23.0
 
+toolchain go1.24.5
+
 require (
 	github.com/Azure/go-amqp v1.4.0
 	github.com/golang-jwt/jwt/v5 v5.2.2
 	github.com/google/uuid v1.6.0
-	github.com/onsi/ginkgo/v2 v2.22.1
-	github.com/onsi/gomega v1.36.2
+	github.com/onsi/ginkgo/v2 v2.23.4
+	github.com/onsi/gomega v1.38.0
 )
 
 require (
 	github.com/go-logr/logr v1.4.2 // indirect
 	github.com/go-task/slim-sprig/v3 v3.0.0 // indirect
-	github.com/google/go-cmp v0.6.0 // indirect
-	github.com/google/pprof v0.0.0-20241210010833-40e02aabc2ad // indirect
-	golang.org/x/net v0.38.0 // indirect
-	golang.org/x/sys v0.31.0 // indirect
-	golang.org/x/text v0.23.0 // indirect
-	golang.org/x/tools v0.28.0 // indirect
+	github.com/google/go-cmp v0.7.0 // indirect
+	github.com/google/pprof v0.0.0-20250403155104-27863c87afa6 // indirect
+	go.uber.org/automaxprocs v1.6.0 // indirect
+	golang.org/x/net v0.41.0 // indirect
+	golang.org/x/sys v0.33.0 // indirect
+	golang.org/x/text v0.26.0 // indirect
+	golang.org/x/tools v0.33.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

go.sum

@@ -10,31 +10,40 @@ github.com/go-task/slim-sprig/v3 v3.0.0 h1:sUs3vkvUymDpBKi3qH1YSqBQk9+9D/8M2mN1v
 github.com/go-task/slim-sprig/v3 v3.0.0/go.mod h1:W848ghGpv3Qj3dhTPRyJypKRiqCdHZiAzKg9hl15HA8=
 github.com/golang-jwt/jwt/v5 v5.2.2 h1:Rl4B7itRWVtYIHFrSNd7vhTiz9UpLdi6gZhZ3wEeDy8=
 github.com/golang-jwt/jwt/v5 v5.2.2/go.mod h1:pqrtFR0X4osieyHYxtmOUWsAWrfe1Q5UVIyoH402zdk=
-github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
-github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
-github.com/google/pprof v0.0.0-20241210010833-40e02aabc2ad h1:a6HEuzUHeKH6hwfN/ZoQgRgVIWFJljSWa/zetS2WTvg=
-github.com/google/pprof v0.0.0-20241210010833-40e02aabc2ad/go.mod h1:vavhavw2zAxS5dIdcRluK6cSGGPlZynqzFM8NdvU144=
+github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
+github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
+github.com/google/pprof v0.0.0-20250403155104-27863c87afa6 h1:BHT72Gu3keYf3ZEu2J0b1vyeLSOYI8bm5wbJM/8yDe8=
+github.com/google/pprof v0.0.0-20250403155104-27863c87afa6/go.mod h1:boTsfXsheKC2y+lKOCMpSfarhxDeIzfZG1jqGcPl3cA=
 github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
 github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
-github.com/onsi/ginkgo/v2 v2.22.1 h1:QW7tbJAUDyVDVOM5dFa7qaybo+CRfR7bemlQUN6Z8aM=
-github.com/onsi/ginkgo/v2 v2.22.1/go.mod h1:S6aTpoRsSq2cZOd+pssHAlKW/Q/jZt6cPrPlnj4a1xM=
-github.com/onsi/gomega v1.36.2 h1:koNYke6TVk6ZmnyHrCXba/T/MoLBXFjeC1PtvYgw0A8=
-github.com/onsi/gomega v1.36.2/go.mod h1:DdwyADRjrc825LhMEkD76cHR5+pUnjhUN8GlHlRPHzY=
+github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
+github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
+github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
+github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
+github.com/onsi/ginkgo/v2 v2.23.4 h1:ktYTpKJAVZnDT4VjxSbiBenUjmlL/5QkBEocaWXiQus=
+github.com/onsi/ginkgo/v2 v2.23.4/go.mod h1:Bt66ApGPBFzHyR+JO10Zbt0Gsp4uWxu5mIOTusL46e8=
+github.com/onsi/gomega v1.38.0 h1:c/WX+w8SLAinvuKKQFh77WEucCnPk4j2OTUr7lt7BeY=
+github.com/onsi/gomega v1.38.0/go.mod h1:OcXcwId0b9QsE7Y49u+BTrL4IdKOBOKnD6VQNTJEB6o=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/prashantv/gostub v1.1.0 h1:BTyx3RfQjRHnUWaGF9oQos79AlQ5k8WNktv7VGvVH4g=
+github.com/prashantv/gostub v1.1.0/go.mod h1:A5zLQHz7ieHGG7is6LLXLz7I8+3LZzsrV0P1IAHhP5U=
 github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk=
 github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
-golang.org/x/net v0.38.0 h1:vRMAPTMaeGqVhG5QyLJHqNDwecKTomGeqbnfZyKlBI8=
-golang.org/x/net v0.38.0/go.mod h1:ivrbrMbzFq5J41QOQh0siUuly180yBYtLp+CKbEaFx8=
-golang.org/x/sys v0.31.0 h1:ioabZlmFYtWhL+TRYpcnNlLwhyxaM9kWTDEmfnprqik=
-golang.org/x/sys v0.31.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
-golang.org/x/text v0.23.0 h1:D71I7dUrlY+VX0gQShAThNGHFxZ13dGLBHQLVl1mJlY=
-golang.org/x/text v0.23.0/go.mod h1:/BLNzu4aZCJ1+kcD0DNRotWKage4q2rGVAg4o22unh4=
-golang.org/x/tools v0.28.0 h1:WuB6qZ4RPCQo5aP3WdKZS7i595EdWqWR8vqJTlwTVK8=
-golang.org/x/tools v0.28.0/go.mod h1:dcIOrVd3mfQKTgrDVQHqCPMWy6lnhfhtX3hLXYVLfRw=
-google.golang.org/protobuf v1.36.1 h1:yBPeRvTftaleIgM3PZ/WBIZ7XM/eEYAaEyCwvyjq/gk=
-google.golang.org/protobuf v1.36.1/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE=
-gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
+go.uber.org/automaxprocs v1.6.0 h1:O3y2/QNTOdbF+e/dpXNNW7Rx2hZ4sTIPyybbxyNqTUs=
+go.uber.org/automaxprocs v1.6.0/go.mod h1:ifeIMSnPZuznNm6jmdzmU3/bfk01Fe2fotchwEFJ8r8=
+golang.org/x/net v0.41.0 h1:vBTly1HeNPEn3wtREYfy4GZ/NECgw2Cnl+nK6Nz3uvw=
+golang.org/x/net v0.41.0/go.mod h1:B/K4NNqkfmg07DQYrbwvSluqCJOOXwUjeb/5lOisjbA=
+golang.org/x/sys v0.33.0 h1:q3i8TbbEz+JRD9ywIRlyRAQbM0qF7hu24q3teo2hbuw=
+golang.org/x/sys v0.33.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
+golang.org/x/text v0.26.0 h1:P42AVeLghgTYr4+xUnTRKDMqpar+PtX7KWuNQL21L8M=
+golang.org/x/text v0.26.0/go.mod h1:QK15LZJUUQVJxhz7wXgxSy/CJaTFjd0G+YLonydOVQA=
+golang.org/x/tools v0.33.0 h1:4qz2S3zmRxbGIhDIAgjxvFutSvH5EfnsYrRBj0UI0bc=
+golang.org/x/tools v0.33.0/go.mod h1:CIJMaWEY88juyUfo7UbgPqbC8rU2OqfAV1h2Qp0oMYI=
+google.golang.org/protobuf v1.36.6 h1:z1NpPI8ku2WgiWnf+t9wTPsn6eP1L7ksHUlkfLvd9xY=
+google.golang.org/protobuf v1.36.6/go.mod h1:jduwjTPXsFjZGTmRluh+L6NjiWu7pchiJ2/5YcXBHnY=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY=
+gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

pkg/rabbitmqamqp/address_test.go

@@ -15,7 +15,7 @@ var _ = Describe("address builder test ", func() {
 				Expect(err.Error()).To(Equal(expectedErr))
 			},
 			Entry("when both exchange and queue are set",
-				stringPtr("my_exchange"), nil, stringPtr("my_queue"),
+				ptr("my_exchange"), nil, ptr("my_queue"),
 				"exchange and queue cannot be set together"),
 			Entry("when neither exchange nor queue is set",
 				nil, nil, nil,
@@ -32,13 +32,13 @@ var _ = Describe("address builder test ", func() {
 				Expect(address).To(Equal(expected))
 			},
 			Entry("with exchange and key",
-				stringPtr("my_exchange"), stringPtr("my_key"),
+				ptr("my_exchange"), ptr("my_key"),
 				"/exchanges/my_exchange/my_key"),
 			Entry("with exchange only",
-				stringPtr("my_exchange"), nil,
+				ptr("my_exchange"), nil,
 				"/exchanges/my_exchange"),
 			Entry("with special characters",
-				stringPtr("my_ exchange/()"), stringPtr("my_key "),
+				ptr("my_ exchange/()"), ptr("my_key "),
 				"/exchanges/my_%20exchange%2F%28%29/my_key%20"),
 		)
 	})