Add missed when* operators for async sources

[idelpivnitskiy]

All asynchronous sources (Completable, Single, Publisher) have before*, when*, and after* operators, for example:

• Completable.beforeOnError(...)
• Completable.whenOnError(...)
• Completable.afterOnError(...)

However, a few when* operators were missed:

• Completable.whenOnSubscribe(Consumer<Cancellable>)
• Completable.whenSubscriber(Supplier<? extends Subscriber>)
• Single.whenOnSubscribe(Consumer<Cancellable>)
• Single.whenSubscriber(Supplier<? extends Subscriber<? super T>>)
• Publisher.whenSubscription(Supplier<? extends Subscription>)
• Publisher.whenSubscriber(Supplier<? extends Subscriber<? super T>>)

We should add these operators. when* usually behaves like after*.

[shekhar-rajak]

Hi,

I want to work on this. Should we have whenOnSubscribe method like this Publisher. whenOnSubscribe for Completable.whenOnSubscribe , Single.whenOnSubscribe ?

[idelpivnitskiy]

Thanks for your interest @shekhar-rajak!
Right, these two and all other operators from the issue description are missed and should be added.

[shekhar-rajak]

Thanks, @idelpivnitskiy I will work on it. Please let me know how can I verify/test/debug the method newly created.

[idelpivnitskiy]

Since this work does not add new logic (it only makes API consistent) and all new methods will delegate to the already existing after* methods, I don't think we need special tests here. See, Completable.whenOnError(...) as an example:

servicetalk/servicetalk-concurrent-api/src/main/java/io/servicetalk/concurrent/api/Completable.java

Lines 175 to 177 in b4413a1

	public final Completable whenOnError(Consumer<Throwable> onError) {
	return afterOnError(onError);
	}

[shekhar-rajak]

Since this work does not add new logic (it only makes API consistent) and all new methods will delegate to the already existing after* methods, I don't think we need special tests here. See, Completable.whenOnError(...) as an example:

servicetalk/servicetalk-concurrent-api/src/main/java/io/servicetalk/concurrent/api/Completable.java

Lines 175 to 177 in b4413a1

	public final Completable whenOnError(Consumer<Throwable> onError) {
	return afterOnError(onError);
	}

Thanks, I understand it.