Fix incorrect request-n management for Publisher.multicast()

[NiteshKant]

Motivation

Publisher<Integer> multi = Publisher.range(1, 10)
                .beforeRequest(n -> System.out.println("Requested from main source: " + n))
                .multicastToExactly(2, 1);
        multi.forEach(t -> System.out.println("From first => " + t));
        multi.forEach(t -> System.out.println("From second => " + t));

Produces the output:

Requested from main source: 1
From first => 1
From second => 1

This shows that multicastToExactly() is not requesting more than queueSize() items unless the subscriber requests more.

Modification

• Remove the conditional request-n call in MulticastUtils.updateRequestN() based on drainCount. Since drained item count is already included in emitted, this additional constraint is not required.
• SubscriberUtils.calculateSourceRequested() did not handle the case when emitted > sourceRequested resulting in higher request-n numbers. For cases like groupBy/multicast when items maybe buffered before they are requested by the subscriber, we need to handle emitted > sourceRequested.

Result

multicastToExactly() correctly requests all items from the original source.

[Scottmitch]

good catch!