/*
 * Copyright 2019 Google LLC
 *
 * 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
 *
 *     https://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.
 */

package com.google.cloud.spanner.r2dbc.client;

import com.google.auth.oauth2.GoogleCredentials;
import com.google.spanner.v1.ExecuteSqlRequest;
import com.google.spanner.v1.PartialResultSet;
import com.google.spanner.v1.SpannerGrpc;
import com.google.spanner.v1.SpannerGrpc.SpannerStub;
import io.grpc.CallCredentials;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.auth.MoreCallCredentials;
import io.grpc.stub.ClientCallStreamObserver;
import io.grpc.stub.ClientResponseObserver;
import java.io.IOException;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscription;
import reactor.core.publisher.Mono;

/**
 * gRPC-based {@link Client} implementation.
 */
public class GrpcClient implements Client {

  public static final String HOST = "spanner.googleapis.com";
  public static final int PORT = 443;

  private final SpannerStub spanner;

  /**
   * Initializes the Cloud Spanner gRPC async stub.
   */
  public GrpcClient() throws IOException {
    // Create a channel
    ManagedChannel channel = ManagedChannelBuilder
        .forAddress(HOST, PORT)
        .build();

    // Create blocking and async stubs using the channel
    CallCredentials callCredentials = MoreCallCredentials
        .from(GoogleCredentials.getApplicationDefault());

    // Create the asynchronous stub for Cloud Spanner
    this.spanner = SpannerGrpc.newStub(channel)
        .withCallCredentials(callCredentials);
  }

  @Override
  public Mono<Void> close() {
    return null;
  }

  @Override
  public Publisher<PartialResultSet> executeStreamingSql(ExecuteSqlRequest request) {
    return subscriber -> spanner.executeStreamingSql(request,
        new ClientResponseObserver<ExecuteSqlRequest, PartialResultSet>() {
          @Override
          public void beforeStart(
              ClientCallStreamObserver<ExecuteSqlRequest> clientCallStreamObserver) {

            clientCallStreamObserver.disableAutoInboundFlowControl();

            subscriber.onSubscribe(new Subscription() {
              @Override
              public void request(long l) {
                clientCallStreamObserver.request((int) l);
              }

              @Override
              public void cancel() {
                clientCallStreamObserver.cancel(null, null);
              }
            });
          }

          @Override
          public void onNext(PartialResultSet partialResultSet) {
            subscriber.onNext(partialResultSet);
          }

          @Override
          public void onError(Throwable throwable) {
            subscriber.onError(throwable);
          }

          @Override
          public void onCompleted() {
            subscriber.onComplete();
          }
        });

  }
}

@@ -1186,40 +1186,35 @@
 ort 
-org.
 react
-ivestreams.Subscription
+or.core.publisher.Flux
 ;%0Aim
@@ -2223,59 +2223,123 @@
 urn 
-subscriber -%3E spanner.executeStreamingSql(request,%0A
+Flux.create(sink -%3E %7B%0A      ClientResponseObserver%3CExecuteSqlRequest, PartialResultSet%3E clientResponseObserver =%0A  
     
@@ -2412,32 +2412,34 @@
 %3E() %7B%0A          
+  
 @Override%0A      
@@ -2424,32 +2424,34 @@
       @Override%0A
+  
           public
@@ -2460,108 +2460,43 @@
 oid 
-beforeStart(%0A              ClientCallStreamObserver%3CExecuteSqlRequest%3E clientCallStreamObserver
+onNext(PartialResultSet value
 ) %7B%0A
-%0A
+  
     
@@ -2507,67 +2507,26 @@
     
-clientCallStreamObserver.disableAutoInboundFlowControl(
+sink.next(value
 );%0A
-%0A
     
@@ -2537,54 +2537,11 @@
     
-subscriber.onSubscribe(new Subscription() %7B%0A  
+%7D%0A%0A
     
@@ -2562,34 +2562,32 @@
 ide%0A            
-  
 public void requ
@@ -2586,82 +2586,57 @@
 oid 
-request(long l) %7B%0A                clientCallStreamObserver.request((int) l
+onError(Throwable t) %7B%0A              sink.error(t
 );%0A 
@@ -2638,39 +2638,35 @@
 t);%0A            
-  
 %7D%0A%0A
-  
             @Ove
@@ -2675,34 +2675,32 @@
 ide%0A            
-  
 public void canc
@@ -2695,22 +2695,27 @@
 ic void 
-cancel
+onCompleted
 () %7B%0A   
@@ -2729,100 +2729,42 @@
     
-  clientCallStreamObserver.cancel(null, null);%0A              %7D%0A            %7D);%0A          %7D%0A%0A
+sink.complete();%0A            %7D%0A%0A  
     
@@ -2771,32 +2771,34 @@
       @Override%0A
+  
           public
@@ -2807,98 +2807,139 @@
 oid 
-onNext(PartialResultSet partialResultSet) %7B%0A            subscriber.onNext(partialResultSet
+beforeStart(ClientCallStreamObserver%3CExecuteSqlRequest%3E requestStream) %7B%0A              requestStream.disableAutoInboundFlowControl(
 );%0A 
@@ -2951,127 +2951,146 @@
     
-%7D%0A%0A
     
-      @Override%0A          public void onError(Throwable throwable) %7B%0A            subscriber.onError(throwable
+sink.onRequest(demand -%3E requestStream.request((int) demand));%0A              sink.onCancel(() -%3E requestStream.cancel(null, null)
 );%0A
+  
     
@@ -3089,33 +3089,32 @@
 ;%0A            %7D%0A
-%0A
           @Overr
@@ -3111,117 +3111,92 @@
     
-@Override%0A          public void onCompleted() %7B%0A            subscriber.onComplete();%0A          %7D%0A    
+%7D;%0A      this.spanner.executeStreamingSql(request, clientResponseObserver);%0A
     %7D);%0A
 %0A  %7D
@@ -3191,15 +3191,14 @@
     %7D);%0A
-%0A
   %7D%0A%7D%0A

Change ExecuteStreamingSql style to use Flux.create (#25)