Remove reactive streams ClientSession#getWrapped method

* Also remove the com.mongodb.internal.async.client.AsyncClientSession
  internal type

JAVA-4790

Author: jyemin

driver-core/src/main/com/mongodb/internal/async/client/AsyncClientSession.java

@@ -18,7 +18,6 @@
 package com.mongodb.reactivestreams.client;
 
 import com.mongodb.TransactionOptions;
-import com.mongodb.internal.async.client.AsyncClientSession;
 import org.reactivestreams.Publisher;
 
 /**
@@ -62,13 +61,6 @@ public interface ClientSession extends com.mongodb.session.ClientSession {
      */
     TransactionOptions getTransactionOptions();
 
-    /**
-     * For internal use only.
-     *
-     * @return the wrapped session
-     */
-    AsyncClientSession getWrapped();
-
     /**
      * Start a transaction in the context of this session with default transaction options. A transaction can not be started if there is
      * already an active transaction on this session.

driver-core/src/test/functional/com/mongodb/internal/async/client/TestHelper.groovy

@@ -23,7 +23,6 @@
 import com.mongodb.connection.ClusterType;
 import com.mongodb.connection.ServerDescription;
 import com.mongodb.internal.async.SingleResultCallback;
-import com.mongodb.internal.async.client.AsyncClientSession;
 import com.mongodb.internal.binding.AbstractReferenceCounted;
 import com.mongodb.internal.binding.AsyncClusterAwareReadWriteBinding;
 import com.mongodb.internal.binding.AsyncConnectionSource;
@@ -33,19 +32,19 @@
 import com.mongodb.internal.session.ClientSessionContext;
 import com.mongodb.internal.session.SessionContext;
 import com.mongodb.lang.Nullable;
+import com.mongodb.reactivestreams.client.ClientSession;
 import org.bson.BsonTimestamp;
 
 import static com.mongodb.assertions.Assertions.notNull;
 import static com.mongodb.connection.ClusterType.LOAD_BALANCED;
 
 public class ClientSessionBinding extends AbstractReferenceCounted implements AsyncReadWriteBinding {
     private final AsyncClusterAwareReadWriteBinding wrapped;
-    private final AsyncClientSession session;
+    private final ClientSession session;
     private final boolean ownsSession;
     private final ClientSessionContext sessionContext;
 
-    public  ClientSessionBinding(final AsyncClientSession session, final boolean ownsSession,
-                                 final AsyncClusterAwareReadWriteBinding wrapped) {
+    public ClientSessionBinding(final ClientSession session, final boolean ownsSession, final AsyncClusterAwareReadWriteBinding wrapped) {
         this.wrapped = notNull("wrapped", wrapped).retain();
         this.ownsSession = ownsSession;
         this.session = notNull("session", session);
@@ -239,9 +238,9 @@ public int release() {
 
     private final class AsyncClientSessionContext extends ClientSessionContext implements SessionContext {
 
-        private final AsyncClientSession clientSession;
+        private final ClientSession clientSession;
 
-        AsyncClientSessionContext(final AsyncClientSession clientSession) {
+        AsyncClientSessionContext(final ClientSession clientSession) {
             super(clientSession);
             this.clientSession = clientSession;
         }

driver-reactive-streams/src/main/com/mongodb/reactivestreams/client/ClientSession.java

@@ -23,7 +23,6 @@
 import com.mongodb.ReadConcern;
 import com.mongodb.TransactionOptions;
 import com.mongodb.WriteConcern;
-import com.mongodb.internal.async.client.AsyncClientSession;
 import com.mongodb.internal.operation.AbortTransactionOperation;
 import com.mongodb.internal.operation.AsyncReadOperation;
 import com.mongodb.internal.operation.AsyncWriteOperation;
@@ -43,7 +42,7 @@
 import static com.mongodb.assertions.Assertions.notNull;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 
-final class ClientSessionPublisherImpl extends BaseClientSessionImpl implements ClientSession, AsyncClientSession {
+final class ClientSessionPublisherImpl extends BaseClientSessionImpl implements ClientSession {
 
     private final OperationExecutor executor;
     private TransactionState transactionState = TransactionState.NONE;
@@ -124,11 +123,6 @@ public void startTransaction(final TransactionOptions transactionOptions) {
         clearTransactionContext();
     }
 
-    @Override
-    public AsyncClientSession getWrapped() {
-        return this;
-    }
-
     @Override
     public Publisher<Void> commitTransaction() {
         if (transactionState == TransactionState.ABORTED) {

driver-reactive-streams/src/main/com/mongodb/reactivestreams/client/internal/ClientSessionBinding.java

@@ -167,7 +167,7 @@ private AsyncReadWriteBinding getReadWriteBinding(final RequestContext requestCo
 
         final AsyncClusterAwareReadWriteBinding asyncReadWriteBinding = readWriteBinding;
         if (session != null) {
-            return new ClientSessionBinding(session.getWrapped(), ownsSession, asyncReadWriteBinding);
+            return new ClientSessionBinding(session, ownsSession, asyncReadWriteBinding);
         } else {
             return asyncReadWriteBinding;
         }

driver-reactive-streams/src/main/com/mongodb/reactivestreams/client/internal/ClientSessionPublisherImpl.java

@@ -27,20 +27,20 @@ import com.mongodb.connection.ServerConnectionState
 import com.mongodb.connection.ServerDescription
 import com.mongodb.connection.ServerType
 import com.mongodb.internal.IgnorableRequestContext
-import com.mongodb.internal.async.client.AsyncClientSession
 import com.mongodb.internal.binding.AsyncClusterAwareReadWriteBinding
 import com.mongodb.internal.binding.AsyncClusterBinding
 import com.mongodb.internal.binding.AsyncConnectionSource
 import com.mongodb.internal.connection.Cluster
 import com.mongodb.internal.connection.Server
 import com.mongodb.internal.connection.ServerTuple
 import com.mongodb.internal.session.ClientSessionContext
+import com.mongodb.reactivestreams.client.ClientSession
 import spock.lang.Specification
 
 class ClientSessionBindingSpecification extends Specification {
     def 'should return the session context from the binding'() {
         given:
-        def session = Stub(AsyncClientSession)
+        def session = Stub(ClientSession)
         def wrappedBinding = Stub(AsyncClusterAwareReadWriteBinding)
         def binding = new ClientSessionBinding(session, false, wrappedBinding)
 
@@ -53,7 +53,7 @@ class ClientSessionBindingSpecification extends Specification {
 
     def 'should return the session context from the connection source'() {
         given:
-        def session = Stub(AsyncClientSession)
+        def session = Stub(ClientSession)
         def wrappedBinding = Mock(AsyncClusterAwareReadWriteBinding) {
             getCluster() >> {
                 Mock(Cluster) {
@@ -99,7 +99,7 @@ class ClientSessionBindingSpecification extends Specification {
 
     def 'should close client session when binding reference count drops to zero if it is owned by the binding'() {
         given:
-        def session = Mock(AsyncClientSession)
+        def session = Mock(ClientSession)
         def wrappedBinding = createStubBinding()
         def binding = new ClientSessionBinding(session, true, wrappedBinding)
         binding.retain()
@@ -119,7 +119,7 @@ class ClientSessionBindingSpecification extends Specification {
 
     def 'should close client session when binding reference count drops to zero due to connection source if it is owned by the binding'() {
         given:
-        def session = Mock(AsyncClientSession)
+        def session = Mock(ClientSession)
         def wrappedBinding = createStubBinding()
         def binding = new ClientSessionBinding(session, true, wrappedBinding)
         def futureResultCallback = new FutureResultCallback<AsyncConnectionSource>()
@@ -150,7 +150,7 @@ class ClientSessionBindingSpecification extends Specification {
 
     def 'should not close client session when binding reference count drops to zero if it is not owned by the binding'() {
         given:
-        def session = Mock(AsyncClientSession)
+        def session = Mock(ClientSession)
         def wrappedBinding = createStubBinding()
         def binding = new ClientSessionBinding(session, false, wrappedBinding)
         binding.retain()
@@ -170,7 +170,7 @@ class ClientSessionBindingSpecification extends Specification {
 
     def 'owned session is implicit'() {
         given:
-        def session = Mock(AsyncClientSession)
+        def session = Mock(ClientSession)
         def wrappedBinding = createStubBinding()
 
         when: