Fix the issue that Sample doesn't call 'unsubscribe'

[zsxwing]

Fix #1958

[akarnokd]

Unfortunately, this makes the following program hang:

Observable.range(0, 100).subscribeOn(Schedulers.io())
        .concatMap(e -> Observable.timer(100, TimeUnit.MILLISECONDS).map(f -> e))
        .sample(1000, TimeUnit.MILLISECONDS)
        .delay(1000, TimeUnit.MILLISECONDS)
        .toBlocking().forEach(System.out::println);
        System.out.println("Done");

Having:

        SamplerSubscriber<T> sampler = new SamplerSubscriber<T>(s);
        child.add(sampler); // unsubscribe goes only upstream this way.
        worker.schedulePeriodically(sampler, time, time, unit);

Makes the example work.

[zsxwing]

I see, it doesn't work with delay. So if an operator calls unsubscribe explicitly, it should not use new Subscriber(Subscriber). Right?

[benjchristensen]

Correct, it should not chain them together as it will unsubscribe up and down the stream. If it wants to unsubscribe up then it must decouple the chain.

[benjchristensen]

I have tested this code locally and it is working. The code was force-pushed to match the feedback from @akarnokd.

Thanks @zsxwing