Introduce abstraction to get session

Always return the connection session for now.
Can be leveraged to create a session for each publisher and
consumer.

Author: acogoluegnes

src/main/java/com/rabbitmq/client/amqp/impl/AmqpConsumer.java

@@ -55,6 +55,7 @@ final class AmqpConsumer extends ResourceBase implements Consumer {
   private final AtomicBoolean paused = new AtomicBoolean(false);
   private final AtomicReference<CountDownLatch> echoedFlowAfterPauseLatch = new AtomicReference<>();
   private final MetricsCollector metricsCollector;
+  private final SessionHandler sessionHandler;
   // native receiver internal state, accessed only in the native executor/scheduler
   private ProtonReceiver protonReceiver;
   private Scheduler protonExecutor;
@@ -72,9 +73,10 @@ final class AmqpConsumer extends ResourceBase implements Consumer {
             .observationCollector()
             .subscribe(builder.queue(), builder.messageHandler());
     this.address = "/queue/" + builder.queue();
-    this.nativeReceiver = createNativeReceiver(builder.connection().nativeSession(), this.address);
-    this.initStateFromNativeReceiver(this.nativeReceiver);
     this.connection = builder.connection();
+    this.sessionHandler = new SessionHandler.ConnectionNativeSessionSessionHandler(this.connection);
+    this.nativeReceiver = this.createNativeReceiver(this.sessionHandler.session(), this.address);
+    this.initStateFromNativeReceiver(this.nativeReceiver);
     this.metricsCollector = this.connection.metricsCollector();
     this.startReceivingLoop();
     this.state(OPEN);
@@ -192,7 +194,7 @@ private void startReceivingLoop() {
   }
 
   void recoverAfterConnectionFailure() {
-    this.nativeReceiver = createNativeReceiver(this.connection.nativeSession(false), this.address);
+    this.nativeReceiver = createNativeReceiver(this.sessionHandler.sessionNoCheck(), this.address);
     this.initStateFromNativeReceiver(this.nativeReceiver);
     this.paused.set(false);
     startReceivingLoop();
@@ -211,6 +213,7 @@ private void close(Throwable cause) {
       }
       try {
         this.nativeReceiver.close();
+        this.sessionHandler.close();
       } catch (Exception e) {
         LOGGER.warn("Error while closing receiver", e);
       }

src/main/java/com/rabbitmq/client/amqp/impl/AmqpPublisher.java

@@ -51,6 +51,7 @@ final class AmqpPublisher extends ResourceBase implements Publisher {
   private final Function<Message, Tracker> publishCall;
   private final DefaultAddressBuilder.DestinationSpec destinationSpec;
   private final Duration publishTimeout;
+  private final SessionHandler sessionHandler;
   private volatile ObservationCollector.ConnectionInfo connectionInfo;
 
   AmqpPublisher(AmqpPublisherBuilder builder) {
@@ -61,8 +62,8 @@ final class AmqpPublisher extends ResourceBase implements Publisher {
     this.destinationSpec = builder.destination();
     this.connection = builder.connection();
     this.publishTimeout = builder.publishTimeout();
-    this.sender =
-        this.createSender(builder.connection().nativeSession(), this.address, this.publishTimeout);
+    this.sessionHandler = new SessionHandler.ConnectionNativeSessionSessionHandler(this.connection);
+    this.sender = this.createSender(sessionHandler.session(), this.address, this.publishTimeout);
     this.metricsCollector = this.connection.metricsCollector();
     this.observationCollector = this.connection.observationCollector();
     this.state(OPEN);
@@ -127,7 +128,7 @@ public void publish(Message message, Callback callback) {
   void recoverAfterConnectionFailure() {
     this.connectionInfo = new Utils.ObservationConnectionInfo(this.connection.connectionAddress());
     this.sender =
-        this.createSender(this.connection.nativeSession(false), this.address, this.publishTimeout);
+        this.createSender(this.sessionHandler.sessionNoCheck(), this.address, this.publishTimeout);
   }
 
   @Override
@@ -162,6 +163,7 @@ private void close(Throwable cause) {
       this.connection.removePublisher(this);
       try {
         this.sender.close();
+        this.sessionHandler.close();
       } catch (Exception e) {
         LOGGER.warn("Error while closing sender", e);
       }

src/main/java/com/rabbitmq/client/amqp/impl/SessionHandler.java

@@ -0,0 +1,51 @@
+// Copyright (c) 2024 Broadcom. All Rights Reserved.
+// The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+// If you have any questions regarding licensing, please contact us at
+// [email protected].
+package com.rabbitmq.client.amqp.impl;
+
+import org.apache.qpid.protonj2.client.Session;
+
+interface SessionHandler extends AutoCloseable {
+
+  Session session();
+
+  Session sessionNoCheck();
+
+  void close();
+
+  class ConnectionNativeSessionSessionHandler implements SessionHandler {
+
+    private final AmqpConnection connection;
+
+    ConnectionNativeSessionSessionHandler(AmqpConnection connection) {
+      this.connection = connection;
+    }
+
+    @Override
+    public Session session() {
+      return this.connection.nativeSession();
+    }
+
+    @Override
+    public Session sessionNoCheck() {
+      return this.connection.nativeSession(false);
+    }
+
+    @Override
+    public void close() {}
+  }
+}