import io.reactivex.rxjava3.core.Flowable;

import io.reactivex.rxjava3.schedulers.Schedulers;

import java.util.concurrent.ThreadLocalRandom;

public class Ch8_05 {

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(s -> System.out.println("Subscriber received " + s),

Throwable::printStackTrace,

() -> 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();

}

}

}