Ensure atomic cleanup in Flux/Mono.usingWhen(…)

[mp911de]

Expected behavior

Cleanup functions should be called only once.

Actual behavior

Cleanup close can be called multiple times if a subscription terminates and then gets cancelled.

Steps to reproduce

Publisher<Void> cleanup = …,
Flux flux = Flux.usingWhen(…, cleanup, cleanup, cleanup);

flux.subscribe(new CoreSubscriber<Object>() {
	Subscription subscription;

	@Override
	public void onSubscribe(Subscription s) {
		s.request(1);
		subscription = s;
	}

	@Override
	public void onNext(Object o) {}

	@Override
	public void onError(Throwable t) {}

	@Override
	public void onComplete() {
		subscription.cancel();
	}
});

Reactor Core version

Reactor 3.2.5