Armeria: Add a StreamMessage subscription option that notifies cancellation to onError()

[trustin]

When a subscriber of a StreamMessage wants to get notified when a subscription has been cancelled, it has to add a callback to StreamMessage.completionFuture(). It is error-prone and cumbersome:

StreamMessage m = ...;
MySubscriber s = ...;
m.subscribe(s);
m.completionFuture().handle(s);

class MySubscriber implements Subscriber<...>, BiFunction<...> {
    ...
    public void onComplete() { /* no-op */ }
    public void onError(Throwable cause) { /* no-op */ }
    public Void apply(result, cause) {
        if (cause == null) { /* handle success */ }
        else { /* handle failure or cancellation */ }
    }
}

Instead, we could provide an option that makes StreamMessage to notify a cancellation event to onError() so that a user doesn't have to add additional callback:

StreamMessage m = ...;
MySubscriber s = ...;
m.subscribe(s, /* some option */);

class MySubscriber implements Subscriber<...>{
    ...
    public void onComplete() { /* handle success */ }
    public void onError(Throwable cause) { /* handle failure or cancellation */ }
}

Currently, subscribe() method is already crowded with overloading, so we may want to do something like this:

m.subscribe(s, SubscriptionOption.NOTIFY_CANCELLATION,
            SubscriptionOption.WITH_POOLED_OBJECTS);

// or

m.subscribe(s, executor, SubscriptionOption.NOTIFY_CANCELLATION,
            SubscriptionOption.WITH_POOLED_OBJECTS);

Alternatively, we could even make Executor an option:

m.subscribe(s, ImmutableMap.of(
    SubscriptionOption.NOTIFY_CANCELLATION, true,
    SubscriptionOption.WITH_POOLED_OBJECTS, true,
    SubscriptionOption.EXECUTOR, executor));

I think I prefer the one that does not use a Map, although the latter is more flexible. Thoughts? /cc @anuraaga

[anuraaga]

I agree the map version seems clunky and harder to use. I think many nio APIs have the same pattern, with a parameter or two and a vararg of options

[minwoox]

So the reason that we introduced the BiFunction is that we do not call onError(cause) when the stream is aborted or canceled.
But in this change, if the caller calls subscribe() with the option, we call onError(cause) on CancelledSubscriptionException right?

[trustin]

Correct.

[minwoox]

I found out that we introduced BiConsumer(currently, BiFunction) to HttpMessageAggregator in #571 because the aggregator wasn't notified when the StreamMessage was aborted. At that time, if we cancel or abort a StreamMessage, we just called Subscription.cancel() so the subscriber couldn't do anything, but just waiting for forever. But after #667, we introduced AbortedStreamException and did not call Subscription.cancel(), but just call Subscription.abort() so the subscriber gets the AbortedStreamException by onError().
If calling onError() with the subscription exception is violating the reactive streams spec, could we need to provide it even though it's an option?