Fix FluxReplay cancel and subscribe race between two Subscribers (#3707)

In FluxReplay combined with refCount(), when one Subscriber leaves,
while concurrently another one subscribes, there can be a chance that
the new one creates a new upstream connection (Subscription to the
source which is shared between Subscribers). At the same time the
connection which is being closed can replay stale values onto the new
Subscriber. This commit ensures that the ordering is correct and the
correct connection is used in such a racy case.

---------

Signed-off-by: Oleh Dokuka <[email protected]>
Co-authored-by: Oleh Dokuka <[email protected]>
Co-authored-by: Dariusz Jędrzejczyk <[email protected]>

Author: 3 people