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

[deer]

• Adds a onSubscriberError(BiConsumer<Subscriber<?>, Throwable>) method to SubscriberRegistry that lets callers register an observer called whenever a subscriber throws from onNext, before that subscriber is terminated.
• The observer defaults to a no-op, so this is a fully backwards-compatible addition.
• Adds a test (shouldNotifyObserverWhenSubscriberThrows) that verifies the observer is invoked with the correct throwable and that the faulting subscriber is subsequently terminated.