Subscription to UnicastProcessor retains Subscriber's reference when disposed

[simonbasle]

Expected behavior

UnicastProcessor<String> processor = UnicastProcessor.create();
System.out.println(processor.downstreamCount());
Disposable subscription = processor.subscribe();
System.out.println(processor.downstreamCount());
subscription.dispose();
System.out.println(processor.downstreamCount());

to print 0 1 0

Actual behavior

Prints 0 1 1.

This is due to the processor's cancel() method not clearing the actual field, which in turn makes downstreamCount() return 1.

Reactor Core version

3.1.2