Expand TestSubscriber's API, fix a bug in MultiFirstProcessor (helidon-io#1463)

Author: akarnokd

common/reactive/src/main/java/io/helidon/common/reactive/MultiFirstProcessor.java

@@ -36,6 +36,7 @@ protected void submit(T item) {
 
     @Override
     protected void next(T item) {
+        cancel();
         super.next(item);
         super.complete();
     }

common/reactive/src/test/java/io/helidon/common/reactive/BaseProcessorTest.java

@@ -152,12 +152,7 @@ public void onSubscribe(Subscription subscription) {
     @Test
     public void testDeferredOnSubscribe() {
         TestProcessor<String> processor = new TestProcessor<>();
-        TestSubscriber<String> subscriber = new TestSubscriber<String>() {
-            @Override
-            public void onSubscribe(Subscription subscription) {
-                subscription.request(1);
-            }
-        };
+        TestSubscriber<String> subscriber = new TestSubscriber<String>(1L);
         processor.subscribe(subscriber);
         processor.onSubscribe(new Subscription() {
             @Override
@@ -171,9 +166,7 @@ public void cancel() {
             }
         });
 
-        assertThat(subscriber.isComplete(), is(equalTo(true)));
-        assertThat(subscriber.getLastError(), is(nullValue()));
-        assertThat(subscriber.getItems(), hasItems("foo"));
+        subscriber.assertResult("foo");
     }
 
     @Test

common/reactive/src/test/java/io/helidon/common/reactive/MultiTest.java

@@ -145,9 +145,7 @@ public void testFromMulti() {
     public void testFirst() {
         MultiTestSubscriber<String> subscriber = new MultiTestSubscriber<>();
         Multi.just("foo", "bar").first().subscribe(subscriber);
-        assertThat(subscriber.isComplete(), is(equalTo(true)));
-        assertThat(subscriber.getLastError(), is(nullValue()));
-        assertThat(subscriber.getItems().get(0), is(equalTo("foo")));
+        subscriber.assertResult("foo");
     }
 
     @Test

common/reactive/src/test/java/io/helidon/common/reactive/SingleTest.java

@@ -97,14 +97,13 @@ public void testJustDoubleSubscriptionRequest() {
         SingleTestSubscriber<String> subscriber = new SingleTestSubscriber<String>() {
             @Override
             public void onSubscribe(Subscription subscription) {
+                super.onSubscribe(subscription);
                 subscription.request(1);
                 subscription.request(1);
             }
         };
         Single.<String>just("foo").subscribe(subscriber);
-        assertThat(subscriber.isComplete(), is(equalTo(true)));
-        assertThat(subscriber.getLastError(), is(nullValue()));
-        assertThat(subscriber.getItems(), hasItems("foo"));
+        subscriber.assertResult("foo");
     }
 
     @Test
@@ -121,6 +120,7 @@ public void testEmptyCanceledSubscription() {
         SingleTestSubscriber<Object> subscriber = new SingleTestSubscriber<Object>() {
             @Override
             public void onSubscribe(Subscription subscription) {
+                super.onSubscribe(subscription);
                 subscription.cancel();
             }
         };