Update to latest go-amqp

Includes update due to API breaking changes.

Author: jhendrixMSFT

go.mod

@@ -11,3 +11,5 @@ require (
 	github.com/stretchr/testify v1.6.1
 	golang.org/x/crypto v0.0.0-20201002170205-7f63de1d35b0
 )
+
+replace github.com/Azure/go-amqp => github.com/jhendrixMSFT/go-amqp v0.12.7-0.20210920180154-4b0772ea90e6

go.sum

@@ -1,7 +1,3 @@
-github.com/Azure/azure-sdk-for-go v51.1.0+incompatible h1:7uk6GWtUqKg6weLv2dbKnzwb0ml1Qn70AdtRccZ543w=
-github.com/Azure/azure-sdk-for-go v51.1.0+incompatible/go.mod h1:9XXNKU+eRnpl9moKnB4QOLf1HestfXbmab5FXxiDBjc=
-github.com/Azure/go-amqp v0.13.13 h1:OBPwCO50EzniOyZR0M4VbGJYDxceIy3SFOnKVMJktdY=
-github.com/Azure/go-amqp v0.13.13/go.mod h1:D5ZrjQqB1dyp1A+G73xeL/kNn7D5qHJIIsNNps7YNmk=
 github.com/Azure/go-autorest v14.2.0+incompatible h1:V5VMDjClD3GiElqLWO7mz2MxNAK/vTfRHdAubSIPRgs=
 github.com/Azure/go-autorest v14.2.0+incompatible/go.mod h1:r+4oMnoxhatjLLJ6zxSWATqVooLgysK6ZNox3g/xq24=
 github.com/Azure/go-autorest/autorest v0.11.18 h1:90Y4srNYrwOtAgVo3ndrQkTYn6kf1Eg/AjTFJ8Is2aM=
@@ -12,10 +8,6 @@ github.com/Azure/go-autorest/autorest/date v0.3.0 h1:7gUk1U5M/CQbp9WoqinNzJar+8K
 github.com/Azure/go-autorest/autorest/date v0.3.0/go.mod h1:BI0uouVdmngYNUzGWeSYnokU+TrmwEsOqdt8Y6sso74=
 github.com/Azure/go-autorest/autorest/mocks v0.4.1 h1:K0laFcLE6VLTOwNgSxaGbUcLPuGXlNkbVvq4cW4nIHk=
 github.com/Azure/go-autorest/autorest/mocks v0.4.1/go.mod h1:LTp+uSrOhSkaKrUy935gNZuuIPPVsHlr9DSOxSayd+k=
-github.com/Azure/go-autorest/autorest/to v0.4.0 h1:oXVqrxakqqV1UZdSazDOPOLvOIz+XA683u8EctwboHk=
-github.com/Azure/go-autorest/autorest/to v0.4.0/go.mod h1:fE8iZBn7LQR7zH/9XU2NcPR4o9jEImooCeWJcYV/zLE=
-github.com/Azure/go-autorest/autorest/validation v0.3.1 h1:AgyqjAd94fwNAoTjl/WQXg4VvFeRFpO+UhNyRXqF1ac=
-github.com/Azure/go-autorest/autorest/validation v0.3.1/go.mod h1:yhLgjC0Wda5DYXl6JAsWyUe4KVNffhoDhG0zVzUMo3E=
 github.com/Azure/go-autorest/logger v0.2.1 h1:IG7i4p/mDa2Ce4TRyAO8IHnVhAVF3RFU+ZtXWSmf4Tg=
 github.com/Azure/go-autorest/logger v0.2.1/go.mod h1:T9E3cAhj2VqvPOtCYAvby9aBXkZmbF5NWuPV8+WeEW8=
 github.com/Azure/go-autorest/tracing v0.6.0 h1:TYi4+3m5t6K48TGI9AUdb+IzbnSxvnvUMfuitfgcfuo=
@@ -31,8 +23,8 @@ github.com/fortytw2/leaktest v1.3.0 h1:u8491cBMTQ8ft8aeV+adlcytMZylmA5nnwwkRZjI8
 github.com/fortytw2/leaktest v1.3.0/go.mod h1:jDsjWgpAGjm2CA7WthBh/CdZYEPF31XHquHwclZch5g=
 github.com/google/go-cmp v0.5.1 h1:JFrFEBb2xKufg6XkJsJr+WbKb4FQlURi5RUcBveYu9k=
 github.com/google/go-cmp v0.5.1/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
-github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
-github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/jhendrixMSFT/go-amqp v0.12.7-0.20210920180154-4b0772ea90e6 h1:33ICQrTJW7tk1gs8J/HnSiTBRPgV7/TqxRNCKDYAfEU=
+github.com/jhendrixMSFT/go-amqp v0.12.7-0.20210920180154-4b0772ea90e6/go.mod h1:9YJ3RhxRT1gquYnzpZO1vcYMMpAdJT+QEg6fwmw9Zlg=
 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/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=

rpc/rpc.go

@@ -65,7 +65,7 @@ type (
 
 		// for unit tests
 		uuidNewV4     func() (uuid.UUID, error)
-		messageAccept func(message *amqp.Message, ctx context.Context) error
+		messageAccept func(ctx context.Context, message *amqp.Message) error
 	}
 
 	// Response is the simplified response structure from an RPC like call
@@ -128,7 +128,6 @@ func NewLinkWithSession(session *amqp.Session, address string, opts ...LinkOptio
 		id:            id,
 
 		uuidNewV4:               uuid.NewV4,
-		messageAccept:           (*amqp.Message).Accept,
 		responseMap:             map[string]chan rpcResponse{},
 		startResponseRouterOnce: &sync.Once{},
 	}
@@ -160,7 +159,6 @@ func NewLinkWithSession(session *amqp.Session, address string, opts ...LinkOptio
 		} else {
 			receiverOpts = append(receiverOpts, amqp.LinkSourceFilter(name, code, link.sessionID))
 		}
-		receiverOpts = append(receiverOpts)
 	}
 
 	receiver, err := session.NewReceiver(receiverOpts...)
@@ -175,6 +173,7 @@ func NewLinkWithSession(session *amqp.Session, address string, opts ...LinkOptio
 
 	link.sender = sender
 	link.receiver = receiver
+	link.messageAccept = receiver.AcceptMessage
 
 	return link, nil
 }
@@ -356,7 +355,7 @@ func (l *Link) RPC(ctx context.Context, msg *amqp.Message) (*Response, error) {
 		Message:     res,
 	}
 
-	if err := l.messageAccept(res, ctx); err != nil {
+	if err := l.messageAccept(ctx, res); err != nil {
 		tab.For(ctx).Error(err)
 		return response, err
 	}

rpc/rpc_test.go

@@ -180,7 +180,7 @@ func TestRPCBasic(t *testing.T) {
 		uuidNewV4: func() (uuid.UUID, error) {
 			return fakeUUID, nil
 		},
-		messageAccept: func(message *amqp.Message, ctx context.Context) error {
+		messageAccept: func(ctx context.Context, message *amqp.Message) error {
 			return nil
 		},
 	}
@@ -225,7 +225,7 @@ func TestRPCFailedSend(t *testing.T) {
 		uuidNewV4: func() (uuid.UUID, error) {
 			return fakeUUID, nil
 		},
-		messageAccept: func(message *amqp.Message, ctx context.Context) error {
+		messageAccept: func(ctx context.Context, message *amqp.Message) error {
 			panic("Should not be called")
 		},
 	}