Expose AssertSubscriber or similar

[rstoyanchev]

Now that the Processor API is deprecated, the need for this is more evident. While replacing usages in RSocket https://github.com/rsocket/rsocket-java/pull/1003 I ran into cases where MonoProcessor is used to store signals so they can be asserted after. This is something that Reactor has for internal use only, the AssertSubscriber, but it would be useful to make it available for application use. RSocket has a copy of that class and I think there may be more copies out there.

Compared to StepVerifier, sometimes we cannot declare expectations up front because there may be more than one possible outcome, in which case we need to verify conditionally depending on the specific outcome.

[simonbasle]

I've been thinking about making AssertSubscriber public vs creating a brand new TestSubscriber in reactor-test. The later seem like a preferable solution, as the design of AssertSubscriber is probably overcomplicated for a user-facing API, and a bit outdated.

Here are the key design points:

• favor exposing some state (list of onNext, terminal signal, list of "protocol errors"...) over assertions. AssertJ will probably be more powerful 99% of the time
• the only exception is the activation and verification of fusion: requested and expected modes would be provided at construction and will be verified at runtime (ie. fail fast if onSubscribe(s) doesn't receive a QueueSubscription or requestFusion doesn't return the expected result)
• expose a cancel() and request(long) method, but don't implement Subscription
• offer a builder/spec API
• offer the possibility to create a ConditionalSubscriber in addition to a plain Subscriber in the builder
• since this is a Subscriber, we don't need to bother about the Fuseable interface
• the resulting subscriber will be a CoreSubscriber and Scannable

Regarding cancellation, other that the fusion failing I don't think the TestSubscriber should internally cancel the subscription.

That said, it could be important in some tests to ensure cancellation, so that created resources are disposed properly. A cancel() in a finally {} block is always possible, but that means we have to ensure the Subscription is still accessible late in tests (so no nulling out the subscription).