fix skip() race condition and request overflow

[davidmoten]

OperatorSkip suffered from this race condition:

Suppose we wanted to skip 5 elements, then

• two concurrent requests of say 1 between onStart request of 0 and first emission would request 1+ (5-0) and 1 + (5-0) elements = 12 elements. To deliver the two requested we only need 7 from upstream so we have overrequested.

It also lacked protection from request overflow.

Two unit tests have been added that failed on previous code and now pass.

[akarnokd]

A shorter solution would be:

@Override
public void setProducer(Producer producer) {
    super.setProducer(producer);
    producer.request(toSkip);
}

which, in theory, should retain the ability to run unbounded and trigger the fast-paths.

[davidmoten]

Beautiful solution! I've updated the commit.

[akarnokd]

Sorry, I wasn't completely right and one slight change is required: replace super with child to avoid the possibility of interference from the Subscriber (because unspecified initial request is turned into MAX_VALUE request in some cases). The reactive-streams API is so much better/clearer in this regard...

[davidmoten]

righto, have done.

[akarnokd]

Thanks!

[benjchristensen]

The reactive-streams API is so much better/clearer in this regard...

Yes it is.