Allow an AMQP Session to be reused when creating an RPC link.

rpc/rpc.go

@@ -70,35 +70,40 @@ func NewLink(conn *amqp.Client, address string) (*Link, error) {
 		return nil, err
 	}
 
-	authSender, err := authSession.NewSender(
-		amqp.LinkTargetAddress(address),
-	)
-	if err != nil {
-		return nil, err
-	}
+	return NewLinkWithSession(conn, authSession, address)
+}
 
-	linkID, err := uuid.NewV4()
-	if err != nil {
-		return nil, err
-	}
+func NewLinkWithSession(conn *amqp.Client, session *amqp.Session, address string) (*Link, error) {
 
-	id := linkID.String()
-	clientAddress := strings.Replace("$", "", address, -1) + replyPostfix + id
-	authReceiver, err := authSession.NewReceiver(
-		amqp.LinkSourceAddress(address),
-		amqp.LinkTargetAddress(clientAddress),
-	)
-	if err != nil {
-		return nil, err
-	}
+		authSender, err := session.NewSender(
+			amqp.LinkTargetAddress(address),
+		)
+		if err != nil {
+			return nil, err
+		}
 
-	return &Link{
-		sender:        authSender,
-		receiver:      authReceiver,
-		session:       authSession,
-		clientAddress: clientAddress,
-		id:            id,
-	}, nil
+		linkID, err := uuid.NewV4()
+		if err != nil {
+			return nil, err
+		}
+
+		id := linkID.String()
+		clientAddress := strings.Replace("$", "", address, -1) + replyPostfix + id
+		authReceiver, err := session.NewReceiver(
+			amqp.LinkSourceAddress(address),
+			amqp.LinkTargetAddress(clientAddress),
+		)
+		if err != nil {
+			return nil, err
+		}
+
+		return &Link{
+			sender:        authSender,
+			receiver:      authReceiver,
+			session:       session,
+			clientAddress: clientAddress,
+			id:            id,
+		}, nil
 }
 
 // RetryableRPC attempts to retry a request a number of times with delay