// Path: cloud-spanner-r2dbc/src/main/java/com/google/cloud/spanner/r2dbc/util/ObservableReactiveUtil.java
// public class ObservableReactiveUtil {
// 
//   /**
//    * Invokes a lambda that in turn issues a remote call, directing the response to a {@link Mono}
//    * stream.
//    *
//    * @param remoteCall lambda capable of invoking the correct remote call, making use of the
//    *     {@link Mono}-converting {@link StreamObserver} implementation.
//    * @param <ResponseT> type of remote call response
//    *
//    * @return {@link Mono} containing the response of the unary call.
//    */
//   public static <ResponseT> Mono<ResponseT> unaryCall(
//       Consumer<StreamObserver<ResponseT>> remoteCall) {
//     return Mono.create(sink -> remoteCall.accept(new UnaryStreamObserver<>(sink)));
//   }
// 
//   /**
//    * Invokes a lambda that issues a streaming call and directs the response to a {@link Flux}
//    * stream.
//    *
//    * @param remoteCall call to make
//    * @param <RequestT> request type
//    * @param <ResponseT> response type
//    *
//    * @return {@link Flux} of response objects resulting from the streaming call.
//    */
//   public static <RequestT, ResponseT> Flux<ResponseT> streamingCall(
//       Consumer<StreamObserver<ResponseT>> remoteCall) {
// 
//     return Flux.create(sink -> {
//       StreamingObserver<RequestT, ResponseT> observer = new StreamingObserver<>(sink);
//       remoteCall.accept(observer);
//       sink.onRequest(demand -> observer.request(demand));
//     });
//   }
// 
//   static class StreamingObserver<RequestT, ResponseT>
//       implements ClientResponseObserver<RequestT, ResponseT>  {
//     ClientCallStreamObserver<RequestT> rsObserver;
//     FluxSink<ResponseT> sink;
// 
//     public StreamingObserver(FluxSink<ResponseT> sink) {
//       this.sink = sink;
//     }
// 
//     @Override
//     public void onNext(ResponseT value) {
//       this.sink.next(value);
//     }
// 
//     @Override
//     public void onError(Throwable throwable) {
//       this.sink.error(SpannerExceptionUtil.createR2dbcException(throwable));
//     }
// 
//     @Override
//     public void onCompleted() {
//       this.sink.complete();
//     }
// 
//     @Override
//     public void beforeStart(ClientCallStreamObserver<RequestT> requestStream) {
//       this.rsObserver = requestStream;
//       requestStream.disableAutoInboundFlowControl();
//       this.sink.onCancel(() -> requestStream.cancel("Flux requested cancel.", null));
//     }
// 
//     public void request(long n) {
//       this.rsObserver.request(n > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) n);
//     }
//   }
// 
//   /**
//    * Forwards the result of a unary gRPC call to a {@link MonoSink}.
//    *
//    * <p>Unary gRPC calls expect a single response or an error, so completion of the call without an
//    * emitted value is an error condition.
//    *
//    * @param <ResponseT> type of expected gRPC call response value.
//    */
//   private static class UnaryStreamObserver<ResponseT> implements StreamObserver<ResponseT> {
// 
//     private boolean terminalEventReceived;
// 
//     private final MonoSink<ResponseT> sink;
// 
//     public UnaryStreamObserver(MonoSink<ResponseT> sink) {
//       this.sink = sink;
//     }
// 
//     @Override
//     public void onNext(ResponseT response) {
//       this.terminalEventReceived = true;
//       this.sink.success(response);
//     }
// 
//     @Override
//     public void onError(Throwable throwable) {
//       this.terminalEventReceived = true;
//       this.sink.error(SpannerExceptionUtil.createR2dbcException(throwable));
//     }
// 
//     @Override
//     public void onCompleted() {
//       if (!this.terminalEventReceived) {
//         this.sink.error(
//             new RuntimeException("Unary gRPC call completed without yielding a value or an error"));
//       }
//     }
//   }
// }


import io.grpc.stub.StreamObserver;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;
import com.google.auth.oauth2.GoogleCredentials;
import com.google.cloud.ServiceOptions;
import com.google.cloud.spanner.r2dbc.util.ObservableReactiveUtil;
import com.google.protobuf.Empty;
import com.google.spanner.v1.DatabaseName;
import com.google.spanner.v1.DeleteSessionRequest;
import com.google.spanner.v1.ListSessionsRequest;
import com.google.spanner.v1.ListSessionsResponse;
import com.google.spanner.v1.Session;
import com.google.spanner.v1.SpannerGrpc;
import com.google.spanner.v1.SpannerGrpc.SpannerStub;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.auth.MoreCallCredentials;
   * Exceptions:
   * - you are running in a CI environment and there are other tests running.
   * - you want to verify session state after a test. In this case, make another method
   *   verifying the correct number of sessions (exactly 1, greater than 10 etc.)
   */
  static void verifyNoLeftoverSessions() throws Exception {
    List<String> activeSessions = getSessionNames();
    if (activeSessions.isEmpty()) {
      System.out.println("No leftover sessions, yay!");
    } else {
      throw new IllegalStateException("Expected no sessions, but found " + activeSessions.size());
    }
  }

  private static List<String> getSessionNames() throws Exception {
    String databaseName = DatabaseName.format(ServiceOptions.getDefaultProjectId(),
        DatabaseProperties.INSTANCE, DatabaseProperties.DATABASE);

    String nextPageToken = null;
    List<String> sessionNames = new ArrayList<>();

    do {
      ListSessionsRequest.Builder requestBuilder =
          ListSessionsRequest.newBuilder().setDatabase(databaseName);

      if (nextPageToken != null) {
        requestBuilder.setPageToken(nextPageToken);
      }

      ListSessionsResponse listSessionsResponse =
// Path: cloud-spanner-r2dbc/src/main/java/com/google/cloud/spanner/r2dbc/util/ObservableReactiveUtil.java
// public class ObservableReactiveUtil {
// 
//   /**
//    * Invokes a lambda that in turn issues a remote call, directing the response to a {@link Mono}
//    * stream.
//    *
//    * @param remoteCall lambda capable of invoking the correct remote call, making use of the
//    *     {@link Mono}-converting {@link StreamObserver} implementation.
//    * @param <ResponseT> type of remote call response
//    *
//    * @return {@link Mono} containing the response of the unary call.
//    */
//   public static <ResponseT> Mono<ResponseT> unaryCall(
//       Consumer<StreamObserver<ResponseT>> remoteCall) {
//     return Mono.create(sink -> remoteCall.accept(new UnaryStreamObserver<>(sink)));
//   }
// 
//   /**
//    * Invokes a lambda that issues a streaming call and directs the response to a {@link Flux}
//    * stream.
//    *
//    * @param remoteCall call to make
//    * @param <RequestT> request type
//    * @param <ResponseT> response type
//    *
//    * @return {@link Flux} of response objects resulting from the streaming call.
//    */
//   public static <RequestT, ResponseT> Flux<ResponseT> streamingCall(
//       Consumer<StreamObserver<ResponseT>> remoteCall) {
// 
//     return Flux.create(sink -> {
//       StreamingObserver<RequestT, ResponseT> observer = new StreamingObserver<>(sink);
//       remoteCall.accept(observer);
//       sink.onRequest(demand -> observer.request(demand));
//     });
//   }
// 
//   static class StreamingObserver<RequestT, ResponseT>
//       implements ClientResponseObserver<RequestT, ResponseT>  {
//     ClientCallStreamObserver<RequestT> rsObserver;
//     FluxSink<ResponseT> sink;
// 
//     public StreamingObserver(FluxSink<ResponseT> sink) {
//       this.sink = sink;
//     }
// 
//     @Override
//     public void onNext(ResponseT value) {
//       this.sink.next(value);
//     }
// 
//     @Override
//     public void onError(Throwable throwable) {
//       this.sink.error(SpannerExceptionUtil.createR2dbcException(throwable));
//     }
// 
//     @Override
//     public void onCompleted() {
//       this.sink.complete();
//     }
// 
//     @Override
//     public void beforeStart(ClientCallStreamObserver<RequestT> requestStream) {
//       this.rsObserver = requestStream;
//       requestStream.disableAutoInboundFlowControl();
//       this.sink.onCancel(() -> requestStream.cancel("Flux requested cancel.", null));
//     }
// 
//     public void request(long n) {
//       this.rsObserver.request(n > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) n);
//     }
//   }
// 
//   /**
//    * Forwards the result of a unary gRPC call to a {@link MonoSink}.
//    *
//    * <p>Unary gRPC calls expect a single response or an error, so completion of the call without an
//    * emitted value is an error condition.
//    *
//    * @param <ResponseT> type of expected gRPC call response value.
//    */
//   private static class UnaryStreamObserver<ResponseT> implements StreamObserver<ResponseT> {
// 
//     private boolean terminalEventReceived;
// 
//     private final MonoSink<ResponseT> sink;
// 
//     public UnaryStreamObserver(MonoSink<ResponseT> sink) {
//       this.sink = sink;
//     }
// 
//     @Override
//     public void onNext(ResponseT response) {
//       this.terminalEventReceived = true;
//       this.sink.success(response);
//     }
// 
//     @Override
//     public void onError(Throwable throwable) {
//       this.terminalEventReceived = true;
//       this.sink.error(SpannerExceptionUtil.createR2dbcException(throwable));
//     }
// 
//     @Override
//     public void onCompleted() {
//       if (!this.terminalEventReceived) {
//         this.sink.error(
//             new RuntimeException("Unary gRPC call completed without yielding a value or an error"));
//       }
//     }
//   }
// }


// Path: cloud-spanner-r2dbc/src/test/java/com/google/cloud/spanner/r2dbc/it/SessionCleanupUtils.java
import io.grpc.stub.StreamObserver;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;
import com.google.auth.oauth2.GoogleCredentials;
import com.google.cloud.ServiceOptions;
import com.google.cloud.spanner.r2dbc.util.ObservableReactiveUtil;
import com.google.protobuf.Empty;
import com.google.spanner.v1.DatabaseName;
import com.google.spanner.v1.DeleteSessionRequest;
import com.google.spanner.v1.ListSessionsRequest;
import com.google.spanner.v1.ListSessionsResponse;
import com.google.spanner.v1.Session;
import com.google.spanner.v1.SpannerGrpc;
import com.google.spanner.v1.SpannerGrpc.SpannerStub;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.auth.MoreCallCredentials;

   * Exceptions:
   * - you are running in a CI environment and there are other tests running.
   * - you want to verify session state after a test. In this case, make another method
   *   verifying the correct number of sessions (exactly 1, greater than 10 etc.)
   */
  static void verifyNoLeftoverSessions() throws Exception {
    List<String> activeSessions = getSessionNames();
    if (activeSessions.isEmpty()) {
      System.out.println("No leftover sessions, yay!");
    } else {
      throw new IllegalStateException("Expected no sessions, but found " + activeSessions.size());
    }
  }

  private static List<String> getSessionNames() throws Exception {
    String databaseName = DatabaseName.format(ServiceOptions.getDefaultProjectId(),
        DatabaseProperties.INSTANCE, DatabaseProperties.DATABASE);

    String nextPageToken = null;
    List<String> sessionNames = new ArrayList<>();

    do {
      ListSessionsRequest.Builder requestBuilder =
          ListSessionsRequest.newBuilder().setDatabase(databaseName);

      if (nextPageToken != null) {
        requestBuilder.setPageToken(nextPageToken);
      }

      ListSessionsResponse listSessionsResponse =
          ObservableReactiveUtil.<ListSessionsResponse>unaryCall(