Merge branch 'master' of github.com:GoogleCloudPlatform/cloud-spanner…

…-r2dbc into codecs-refactoring

# Conflicts:
#	src/test/java/com/google/cloud/spanner/r2dbc/SpannerStatementTest.java

.travis.yml

@@ -15,11 +15,11 @@ before_install:
     fi;
 
 install:
-  - ./mvnw clean install -DskipTests=true -Dmaven.javadoc.skip=true -B -V -q
+  - ./mvnw clean install -DskipTests=true -Dmaven.javadoc.skip=true -B -V
 
 script:
   - if [ "$TRAVIS_SECURE_ENV_VARS" == "true" ]; then
-      ./mvnw verify -B -V -q;
+      ./mvnw verify -B -V;
     else
-      ./mvnw verify -B -V -q -DskipITs;
+      ./mvnw verify -B -V -DskipITs;
     fi;

src/test/java/com/google/cloud/spanner/r2dbc/SpannerConnectionFactoryTest.java

@@ -26,6 +26,7 @@
 import org.junit.Test;
 import org.mockito.Mockito;
 import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
 
 /**
  * Test for {@link SpannerConnectionFactory}.
@@ -63,9 +64,10 @@ public void createReturnsNewSpannerConnection() {
         .thenReturn(Mono.just(session));
 
     SpannerConnectionFactory factory = new SpannerConnectionFactory(mockClient, this.config);
-    SpannerConnection connection = Mono.from(factory.create()).block();
-
-    assertThat(connection.getSession().getName()).isEqualTo("jam session");
+    Mono<SpannerConnection> connection = Mono.from(factory.create());
 
+    StepVerifier.create(connection.map(con -> con.getSession().getName()))
+            .expectNext("jam session")
+            .verifyComplete();
   }
 }

src/test/java/com/google/cloud/spanner/r2dbc/SpannerConnectionTest.java

@@ -116,8 +116,9 @@ public void beginAndCommitTransactions() {
     when(this.mockClient.commitTransaction(TEST_SESSION,  Transaction.getDefaultInstance()))
         .thenReturn(commitTransactionProbe.mono());
 
-    Mono.from(connection.beginTransaction()).block();
-    Mono.from(connection.commitTransaction()).block();
+    Mono.from(connection.beginTransaction())
+            .then(Mono.from(connection.commitTransaction()))
+            .subscribe();
     verify(this.mockClient, times(1))
         .beginTransaction(TEST_SESSION);
     verify(this.mockClient, times(1))

src/test/java/com/google/cloud/spanner/r2dbc/SpannerResultTest.java

@@ -16,7 +16,6 @@
 
 package com.google.cloud.spanner.r2dbc;
 
-import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import com.google.protobuf.Value;
@@ -26,11 +25,11 @@
 import com.google.spanner.v1.Type;
 import com.google.spanner.v1.TypeCode;
 import java.util.Collections;
-import java.util.List;
 import org.junit.Before;
 import org.junit.Test;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
 
 /**
  * Test for {@link SpannerResult}.
@@ -68,9 +67,10 @@ public void setup() {
 
   @Test
   public void getRowsUpdatedTest() {
-    assertThat(
-        ((Mono) new SpannerResult(this.resultSet,Mono.just(2)).getRowsUpdated()).block())
-        .isEqualTo(2);
+    StepVerifier.create(
+        ((Mono) new SpannerResult(this.resultSet,Mono.just(2)).getRowsUpdated()))
+        .expectNext(2)
+        .verifyComplete();
   }
 
   @Test
@@ -93,17 +93,15 @@ public void mapTest() {
         .getFields(0)
         .getName();
 
-    List<String> result =
+    Flux<String> result =
         new SpannerResult(this.resultSet, Mono.just(0))
             .map((row, metadata) ->
                 row.get(0, String.class)
                     + "-"
-                    + metadata.getColumnMetadata(0).getName())
-            .collectList()
-            .block();
-
-    assertThat(result)
-        .containsExactly("key1-" + columnName, "key2-" + columnName);
+                    + metadata.getColumnMetadata(0).getName());
 
+    StepVerifier.create(result)
+        .expectNext("key1-" + columnName, "key2-" + columnName)
+        .verifyComplete();
   }
 }

src/test/java/com/google/cloud/spanner/r2dbc/SpannerStatementTest.java

@@ -39,7 +39,6 @@
 import io.r2dbc.spi.Result;
 import java.util.Collections;
 import java.util.HashMap;
-import java.util.List;
 import java.util.Map;
 import org.junit.Test;
 import reactor.core.publisher.Flux;
@@ -86,12 +85,14 @@ public void executeDummyImplementation() {
     SpannerStatement statement
         = new SpannerStatement(mockClient, TEST_SESSION, null, sql);
 
-    Flux<SpannerResult> result = (Flux<SpannerResult>)statement.execute();
+    Flux<SpannerResult> result = (Flux<SpannerResult>) statement.execute();
 
     assertThat(result).isNotNull();
 
-    assertThat(result.next().block().map((r, m) -> (String)r.get(0)).blockFirst())
-        .isEqualTo("Odyssey");
+    StepVerifier.create(result.flatMap(
+        spannerResult -> spannerResult.map((row, rowMetadata) -> (String) row.get(0))))
+            .expectNext("Odyssey")
+            .verifyComplete();
 
     verify(mockClient).executeStreamingSql(eq(TEST_SESSION), isNull(), eq(sql), any(), any());
   }
@@ -143,9 +144,10 @@ public void executeDummyImplementationBind() {
         .bind("id", "b2")
         .execute();
 
-    StepVerifier.create(result)
-        .expectNextMatches(spannerResult -> firstColumIsEqualTo(spannerResult, "Odyssey"))
-        .expectNextMatches(spannerResult -> firstColumIsEqualTo(spannerResult, "Fables"))
+    StepVerifier.create(result.flatMap(
+        spannerResult -> spannerResult.map((row, rowMetadata) -> (String) row.get(0))))
+        .expectNext("Odyssey")
+        .expectNext("Fables")
         .verifyComplete();
 
     verify(mockClient, times(1)).executeStreamingSql(TEST_SESSION, null, sql,
@@ -154,10 +156,6 @@ public void executeDummyImplementationBind() {
         idBinding2, types);
   }
 
-  private boolean firstColumIsEqualTo(SpannerResult spannerResult, String pattern) {
-    return spannerResult.map((row, rowMetadata) -> (String)row.get(0)).blockFirst().equals(pattern);
-  }
-
   @Test
   public void readOneResultSetQueryTest() {
     PartialResultSet p1 = PartialResultSet.newBuilder().setMetadata(
@@ -173,10 +171,13 @@ public void readOneResultSetQueryTest() {
     Mono<Result> resultMono = Mono
         .from(new SpannerStatement(this.mockClient, null, null, "").execute());
 
-    assertThat(resultMono.flatMap(r -> Mono.from(r.getRowsUpdated())).block()).isZero();
-    assertThat(resultMono.flatMapMany(r -> r
-        .map((row, meta) -> row.get(0, Boolean.class).toString() + "-" + row.get(1, String.class)))
-        .collectList().block()).containsExactly("false-abc");
+    StepVerifier.create(resultMono.flatMap(r -> Mono.from(r.getRowsUpdated())))
+        .expectNext(0)
+        .verifyComplete();
+    StepVerifier.create(resultMono.flatMapMany(r -> r
+        .map((row, meta) -> row.get(0, Boolean.class).toString() + "-" + row.get(1, String.class))))
+        .expectNext("false-abc")
+        .verifyComplete();
   }
 
   @Test
@@ -193,8 +194,10 @@ public void readMultiResultSetQueryTest() {
 
     when(this.mockClient.executeStreamingSql(any(), any(), any(), any(), any())).thenReturn(inputs);
 
-    assertThat(Mono.from(new SpannerStatement(this.mockClient, null, null, "").execute())
-        .flatMap(r -> Mono.from(r.getRowsUpdated())).block()).isZero();
+    StepVerifier.create(Flux.from(new SpannerStatement(this.mockClient, null, null, "").execute())
+        .flatMap(r -> Mono.from(r.getRowsUpdated())))
+        .expectNext(0)
+        .verifyComplete();
   }
 
   @Test
@@ -207,8 +210,10 @@ public void readDmlQueryTest() {
 
     when(this.mockClient.executeStreamingSql(any(), any(), any(), any(), any())).thenReturn(inputs);
 
-    assertThat(Mono.from(new SpannerStatement(this.mockClient, null, null, "").execute())
-        .flatMap(r -> Mono.from(r.getRowsUpdated())).block()).isEqualTo(555);
+    StepVerifier.create(Flux.from(new SpannerStatement(this.mockClient, null, null, "").execute())
+        .flatMap(r -> Mono.from(r.getRowsUpdated())))
+        .expectNext(555)
+        .verifyComplete();
   }
 
   @Test
@@ -226,15 +231,19 @@ public void noopMapOnUpdateQueriesWhenNoRowsAffected() {
     SpannerStatement statement
         = new SpannerStatement(mockClient, TEST_SESSION, null, sql);
 
-    SpannerResult result = ((Flux<SpannerResult>) statement.execute()).next().block();
+    Flux<SpannerResult> result = (Flux<SpannerResult>) statement.execute();
 
-    List<String> rowStrings = result.map((r, m) -> (String)r.get(0)).collectList().block();
-    assertThat(rowStrings).isEmpty();
+    StepVerifier.create(result.flatMap(
+        spannerResult -> spannerResult.map((row, rowMetadata) -> (String) row.get(0))
+            .collectList()))
+        .expectNext(Collections.emptyList())
+        .verifyComplete();
 
-    int rowsUpdated = Mono.from(result.getRowsUpdated()).block();
-    assertThat(rowsUpdated).isEqualTo(0);
+    StepVerifier.create(result.flatMap(results -> Mono.from(results.getRowsUpdated())))
+        .expectNext(0)
+        .verifyComplete();
 
-    verify(mockClient, times(1)).executeStreamingSql(TEST_SESSION, null, sql,
+    verify(mockClient, times(2)).executeStreamingSql(TEST_SESSION, null, sql,
         Struct.newBuilder().build(), Collections.EMPTY_MAP);
   }
 }

src/test/java/com/google/cloud/spanner/r2dbc/result/PartialResultRowExtractorTest.java

@@ -16,8 +16,6 @@
 
 package com.google.cloud.spanner.r2dbc.result;
 
-import static org.assertj.core.api.Assertions.assertThat;
-
 import com.google.cloud.spanner.r2dbc.SpannerColumnMetadata;
 import com.google.cloud.spanner.r2dbc.SpannerRow;
 import com.google.protobuf.ListValue;
@@ -31,9 +29,9 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.stream.Collectors;
-import java.util.stream.StreamSupport;
 import org.junit.Test;
 import reactor.core.publisher.Flux;
+import reactor.test.StepVerifier;
 
 /**
  * Tests the partial result flux converter.
@@ -190,34 +188,36 @@ public void handleEmptyPartialResultSet() {
 
     Flux<PartialResultSet> inputs = Flux.just(emptyResultSet);
 
-    List<SpannerRow> results =
-        inputs.flatMapIterable(new PartialResultRowExtractor())
-            .collectList()
-            .block();
-    assertThat(results).isEmpty();
+    Flux<SpannerRow> results =
+        inputs.flatMapIterable(new PartialResultRowExtractor());
+
+    StepVerifier.create(results)
+        .verifyComplete();
   }
 
   private void verifyRows(Flux<PartialResultSet> inputs) {
-    List<SpannerRow> results =
-        inputs.flatMapIterable(new PartialResultRowExtractor())
-            .collectList()
-            .block();
+    Flux<SpannerRow> results =
+        inputs.flatMapIterable(new PartialResultRowExtractor());
 
-    List<ColumnMetadata> columnMetadata = this.resultSetMetadata.getRowType().getFieldsList()
+    List<String> expectedColNames = this.resultSetMetadata.getRowType().getFieldsList()
         .stream()
         .map(SpannerColumnMetadata::new)
+        .map(ColumnMetadata::getName)
         .collect(Collectors.toList());
 
-    List<String> expectedColNames = columnMetadata.stream().map(ColumnMetadata::getName)
-        .collect(Collectors.toList());
+    String[] expectedColumnNames = new String[expectedColNames.size()];
+    expectedColNames.toArray(expectedColumnNames);
 
-    results.forEach(row -> assertThat(
-        StreamSupport.stream(row.getRowMetadata().getColumnMetadatas().spliterator(), false)
-            .map(
-                ColumnMetadata::getName).collect(Collectors.toList()))
-        .isEqualTo(expectedColNames));
+    StepVerifier.create(results
+        .flatMap(row -> Flux.fromIterable(row.getRowMetadata().getColumnMetadatas())
+            .map(ColumnMetadata::getName)))
+        .expectNext(expectedColumnNames)
+        .expectNext(expectedColumnNames)
+        .verifyComplete();
 
-    assertThat(results.get(0).getValues()).containsExactly(this.a1, this.a2, this.a3);
-    assertThat(results.get(1).getValues()).containsExactly(this.b1, this.b2, this.b3);
+    StepVerifier.create(results.map(SpannerRow::getValues))
+        .expectNext(Arrays.asList(this.a1, this.a2, this.a3))
+        .expectNext(Arrays.asList(this.b1, this.b2, this.b3))
+        .verifyComplete();
   }
 }