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.

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

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.

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

Thanks @zsxwing