Do not defer error emission during result consumption.

Consuming results through Result.getRowsUpdated() or Result.map() now emits error signals directly without accumulating these.

This behavior prevents consuming results if an error token is present and arrangements that adopt one of the resulting publishers into a Mono (via Mono.from(…)) no longer causes a false impression of a successful call when consuming the first signal only without awaiting a completion signal.

[resolves #180]

Signed-off-by: Mark Paluch <mpaluch@vmware.com>

Author: mp911de

src/main/java/io/r2dbc/mssql/DefaultMssqlResult.java

@@ -26,7 +26,6 @@
 import io.r2dbc.mssql.message.token.ReturnValue;
 import io.r2dbc.mssql.message.token.RowToken;
 import io.r2dbc.mssql.util.Assert;
-import io.r2dbc.spi.R2dbcException;
 import io.r2dbc.spi.Readable;
 import io.r2dbc.spi.Result;
 import io.r2dbc.spi.Row;
@@ -66,8 +65,6 @@ final class DefaultMssqlResult implements MssqlResult {
 
     private volatile MssqlRowMetadata rowMetadata;
 
-    private volatile RuntimeException throwable;
-
     private DefaultMssqlResult(String sql, ConnectionContext context, Codecs codecs, Flux<io.r2dbc.mssql.message.Message> messages, boolean expectReturnValues) {
 
         this.sql = sql;
@@ -124,24 +121,11 @@ public Mono<Integer> getRowsUpdated() {
 
                 if (message instanceof ErrorToken) {
 
-                    R2dbcException mssqlException = ExceptionFactory.createException((ErrorToken) message, this.sql);
-
-                    Throwable exception = this.throwable;
-                    if (exception != null) {
-                        exception.addSuppressed(mssqlException);
-                    } else {
-                        this.throwable = mssqlException;
-                    }
-
+                    sink.error(ExceptionFactory.createException((ErrorToken) message, this.sql));
                     return;
                 }
 
                 ReferenceCountUtil.release(message);
-            }).doOnComplete(() -> {
-                RuntimeException exception = this.throwable;
-                if (exception != null) {
-                    throw exception;
-                }
             }).reduce(Long::sum).map(Long::intValue);
     }
 
@@ -242,16 +226,7 @@ private <T> Flux<T> doMap(boolean rows, boolean outparameters, Function<? super
                 }
 
                 if (message instanceof ErrorToken) {
-
-                    R2dbcException mssqlException = ExceptionFactory.createException((ErrorToken) message, this.sql);
-
-                    Throwable exception = this.throwable;
-                    if (exception != null) {
-                        exception.addSuppressed(mssqlException);
-                    } else {
-                        this.throwable = mssqlException;
-                    }
-
+                    sink.error(ExceptionFactory.createException((ErrorToken) message, this.sql));
                     return;
                 }
 
@@ -266,12 +241,7 @@ private <T> Flux<T> doMap(boolean rows, boolean outparameters, Function<? super
             mapped = mapped.concatWith(mappedReturnValues);
         }
 
-        return mapped.doOnComplete(() -> {
-            RuntimeException exception = this.throwable;
-            if (exception != null) {
-                throw exception;
-            }
-        });
+        return mapped;
     }
 
     @Override

src/main/java/io/r2dbc/mssql/MssqlSegmentResult.java

@@ -68,8 +68,6 @@ final class MssqlSegmentResult implements MssqlResult {
 
     private final Flux<Segment> segments;
 
-    private volatile RuntimeException throwable;
-
     private MssqlSegmentResult(String sql, ConnectionContext context, Codecs codecs, Flux<Segment> segments) {
 
         this.sql = sql;
@@ -200,16 +198,11 @@ public Mono<Integer> getRowsUpdated() {
                 }
 
                 if (isError(segment)) {
-                    handleError((Message) segment);
+                    sink.error(((Message) segment).exception());
                     return;
                 }
 
                 ReferenceCountUtil.release(segment);
-            }).doOnComplete(() -> {
-                RuntimeException exception = this.throwable;
-                if (exception != null) {
-                    throw exception;
-                }
             }).reduce(Long::sum).map(Long::intValue);
     }
 
@@ -235,7 +228,7 @@ public <T> Flux<T> map(Function<? super Readable, ? extends T> mappingFunction)
     private <T> Flux<T> doMap(boolean rows, boolean outparameters, Function<? super Readable, ? extends T> mappingFunction) {
 
         return this.segments
-            .<T>handle((segment, sink) -> {
+            .handle((segment, sink) -> {
 
                 if (rows && segment instanceof RowSegment) {
 
@@ -264,16 +257,11 @@ private <T> Flux<T> doMap(boolean rows, boolean outparameters, Function<? super
                 }
 
                 if (isError(segment)) {
-                    handleError((Message) segment);
+                    sink.error(((Message) segment).exception());
                     return;
                 }
 
                 ReferenceCountUtil.release(segment);
-            }).doOnComplete(() -> {
-                RuntimeException exception = this.throwable;
-                if (exception != null) {
-                    throw exception;
-                }
             });
     }
 
@@ -319,17 +307,6 @@ public <T> Flux<T> flatMap(Function<Segment, ? extends Publisher<? extends T>> m
             });
     }
 
-    private void handleError(Message segment) {
-        R2dbcException mssqlException = segment.exception();
-
-        Throwable exception = this.throwable;
-        if (exception != null) {
-            exception.addSuppressed(mssqlException);
-        } else {
-            this.throwable = mssqlException;
-        }
-    }
-
     private boolean isError(Segment segment) {
         return segment instanceof MssqlMessage && ((MssqlMessage) segment).isError();
     }

src/test/java/io/r2dbc/mssql/MssqlResultUnitTests.java

@@ -21,14 +21,13 @@
 import io.r2dbc.mssql.message.Message;
 import io.r2dbc.mssql.message.token.DoneToken;
 import io.r2dbc.mssql.message.token.ErrorToken;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
 import reactor.core.publisher.Flux;
 import reactor.test.StepVerifier;
 
-import java.util.Iterator;
-import java.util.stream.Stream;
-
-import static org.assertj.core.api.Assertions.assertThat;
+import java.util.Arrays;
+import java.util.List;
 
 /**
  * Unit tests for {@link DefaultMssqlResult}.
@@ -37,20 +36,59 @@
  */
 class MssqlResultUnitTests {
 
-    @Test
-    void shouldDeferErrorSignal() {
+    @ParameterizedTest
+    @MethodSource("factories")
+    void shouldEmitErrorSignalInOrder(ResultFactory factory) {
 
         ErrorToken error = new ErrorToken(0, 0, Byte.MIN_VALUE, Byte.MIN_VALUE, "foo", "", "", 0);
         DoneToken done = DoneToken.create(0);
-        Iterator<Message> iterator = Stream.of(error, done).map(Message.class::cast).iterator();
 
-        MssqlResult result = MssqlSegmentResult.toResult("", new ConnectionContext(), new DefaultCodecs(), Flux.fromIterable(() -> iterator), false);
+        MssqlResult countThenError = factory.create(Flux.just(done, error));
 
-        result.getRowsUpdated()
+        countThenError.getRowsUpdated()
             .as(StepVerifier::create)
             .expectError()
             .verify();
 
-        assertThat(iterator.hasNext()).isFalse();
+        MssqlResult errorThenCount = factory.create(Flux.just(error, done));
+
+        errorThenCount.getRowsUpdated()
+            .as(StepVerifier::create)
+            .expectError()
+            .verify();
     }
+
+    static List<ResultFactory> factories() {
+
+        return Arrays.asList(new ResultFactory() {
+
+            @Override
+            MssqlResult create(Flux<Message> messages) {
+                return DefaultMssqlResult.toResult("", new ConnectionContext(), new DefaultCodecs(), messages, false);
+            }
+
+            @Override
+            public String toString() {
+                return "DefaultMssqlResult";
+            }
+        }, new ResultFactory() {
+
+            @Override
+            MssqlResult create(Flux<Message> messages) {
+                return MssqlSegmentResult.toResult("", new ConnectionContext(), new DefaultCodecs(), messages, false);
+            }
+
+            @Override
+            public String toString() {
+                return "MssqlSegmentResult";
+            }
+        });
+    }
+
+    static abstract class ResultFactory {
+
+        abstract MssqlResult create(Flux<Message> messages);
+
+    }
+
 }