Ensure that AsyncReadWriteBinding is released prior to subscriber not…

…ification (#676)

Previously the driver assumed that Mono#doOnTerminate is executed
prior to the subscriber being notified of completion.  But it turns
out that behavior is not guaranteed in the Californium release of
Project Reactor (though it is in later releases).  So instead,
now the SingleResultCallback that is used to notify the Mongo of
completion is wrapped by one that first releases the binding, and
Mono#doOnTerminate is no longer used.

JAVA-4027

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

@@ -66,12 +66,16 @@ public <T> Mono<T> execute(final AsyncReadOperation<T> operation, final ReadPref
                         binding.release();
                         return Mono.error(new MongoClientException("Read preference in a transaction must be primary"));
                     } else {
-                        return Mono.<T>create(sink -> operation.executeAsync(binding, sinkToCallback(sink)))
-                                .doOnTerminate(binding::release)
-                                .doOnError((t) -> {
-                                    labelException(session, t);
-                                    unpinServerAddressOnTransientTransactionError(session, t);
-                                });
+                        return Mono.<T>create(sink -> operation.executeAsync(binding, (result, t) -> {
+                            try {
+                                binding.release();
+                            } finally {
+                                sinkToCallback(sink).onResult(result, t);
+                            }
+                        })).doOnError((t) -> {
+                            labelException(session, t);
+                            unpinServerAddressOnTransientTransactionError(session, t);
+                        });
                     }
                 });
     }
@@ -86,13 +90,16 @@ public <T> Mono<T> execute(final AsyncWriteOperation<T> operation, final ReadCon
                                                               session == null && clientSession != null))
                 .switchIfEmpty(Mono.fromCallable(() -> getReadWriteBinding(ReadPreference.primary(), readConcern, session, false)))
                 .flatMap(binding ->
-                    Mono.<T>create(sink -> operation.executeAsync(binding, sinkToCallback(sink)))
-                                .doOnTerminate(binding::release)
-                                .doOnError((t) -> {
-                                    labelException(session, t);
-                                    unpinServerAddressOnTransientTransactionError(session, t);
-                                })
-
+                        Mono.<T>create(sink -> operation.executeAsync(binding, (result, t) -> {
+                            try {
+                                binding.release();
+                            } finally {
+                                sinkToCallback(sink).onResult(result, t);
+                            }
+                        })).doOnError((t) -> {
+                            labelException(session, t);
+                            unpinServerAddressOnTransientTransactionError(session, t);
+                        })
                 );
     }