fix(base-flow): add subscriber error observer to SubscriberRegistry

base-flow/src/main/java/build/base/flow/SubscriberRegistry.java

@@ -27,6 +27,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.BiConsumer;
 
 /**
  * A {@link Publicist} for registered {@link Subscriber}s to which items may be published.
@@ -48,6 +49,12 @@ public class SubscriberRegistry<T>
      */
     private final CompletableFuture<?> onCompleted;
 
+    /**
+     * Called when a {@link Subscriber} throws from {@link Subscriber#onNext(Object)}, before the subscriber
+     * is terminated.  Defaults to a no-op.
+     */
+    private BiConsumer<Subscriber<?>, Throwable> subscriberErrorObserver = (_, _) -> {};
+
     /**
      * Constructs a {@link SubscriberRegistry}.
      */
@@ -56,6 +63,17 @@ public SubscriberRegistry() {
         this.onCompleted = new CompletableFuture<>();
     }
 
+    /**
+     * Sets the observer to be notified when a {@link Subscriber} throws from {@link Subscriber#onNext(Object)}.
+     *
+     * @param observer a {@link BiConsumer} receiving the failing subscriber and the throwable it raised
+     * @return this registry, for fluent configuration
+     */
+    public SubscriberRegistry<T> onSubscriberError(final BiConsumer<Subscriber<?>, Throwable> observer) {
+        this.subscriberErrorObserver = Objects.requireNonNull(observer, "The observer can't be null");
+        return this;
+    }
+
     @Override
     public void subscribe(final Subscriber<? super T> subscriber) {
         Objects.requireNonNull(subscriber, "The subscriber can't be null");
@@ -289,6 +307,7 @@ private void deliver() {
                                 }
                             }
                             catch (final Throwable throwable) {
+                                SubscriberRegistry.this.subscriberErrorObserver.accept(this.subscriber, throwable);
                                 error(throwable);
                             }
                         }

base-flow/src/test/java/build/base/flow/SubscriberRegistryTests.java

@@ -274,6 +274,31 @@ void shouldPreventSubscriptionWhenError() {
             .isEqualTo(throwable);
     }
 
+    /**
+     * Ensure the subscriber error observer is notified when a {@link Subscriber} throws from
+     * {@link Subscriber#onNext(Object)}, and that the subscriber is subsequently terminated.
+     */
+    @Test
+    void shouldNotifyObserverWhenSubscriberThrows() {
+        final SubscriberRegistry<String> registry = new SubscriberRegistry<>();
+
+        final List<Throwable> observed = new ArrayList<>();
+        registry.onSubscriberError((_, throwable) -> observed.add(throwable));
+
+        final RuntimeException thrown = new RuntimeException("disk full");
+        final TrackingSubscriber<String> subscriber = new TrackingSubscriber<>() {
+            @Override
+            public void onNext(final String item) {
+                throw thrown;
+            }
+        };
+        registry.subscribe(subscriber);
+        registry.publish("hello");
+
+        assertThat(observed).containsExactly(thrown);
+        assertThat(subscriber.throwable()).contains(thrown);
+    }
+
     /**
      * Ensure {@link SubscriberRegistry} will complete only once, and subsequent completion and error attempts will
      * have no effect (return false).