import io.reactivex.rxjava3.core.Flowable;

import io.reactivex.rxjava3.schedulers.Schedulers;

import org.reactivestreams.Subscriber;

import org.reactivestreams.Subscription;

import java.util.concurrent.ThreadLocalRandom;

public class Ch8_06 {

public static void main(String[] args) {

Flowable.range(1, 1000)

.doOnNext(s -> System.out.println("Source pushed " + s))

.observeOn(Schedulers.io())

.map(i -> intenseCalculation(i))

.subscribe(new Subscriber<Integer>() {

@Override

public void onSubscribe(Subscription subscription) {

subscription.request(Long.MAX_VALUE);

}

@Override

public void onNext(Integer s) {

sleep(50);

System.out.println("Subscriber received " + s);

}

@Override

public void onError(Throwable e) {

e.printStackTrace();

}

@Override

public void onComplete() {

System.out.println("Done!");

}

});

sleep(20000);

}

public static <T> T intenseCalculation(T value) {

//sleep up to 200 milliseconds

sleep(ThreadLocalRandom.current().nextInt(200));

return value;

}

private static void sleep(long millis) {

try {

Thread.sleep(millis);

} catch (InterruptedException e) {

e.printStackTrace();

}

}

}