package rx;

import io.reactivex.rxjava3.annotations.NonNull;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.functions.Consumer;
import io.reactivex.rxjava3.functions.Function;
import io.reactivex.rxjava3.schedulers.Schedulers;

import java.util.concurrent.TimeUnit;

public class TimeoutExample {
 public static void main(String[] args) throws InterruptedException {
 Observable<String> observable =
 Observable.fromCallable(() -> readFile())
// Observable.just("Hello, world!")
 .delay(0, TimeUnit.SECONDS)
 .timeout(2, TimeUnit.SECONDS)
// .subscribeOn(Schedulers.newThread())
 .map(new Function<String, String>() {
 @Override
 public String apply(String s) throws Throwable {
 System.out.println("111 map:" + s + Thread.currentThread().getName());
 return s;
 }
 })
 .doOnNext(new Consumer<String>() {
 @Override
 public void accept(String s) throws Throwable {
 System.out.println("222 nextMap:" + s + Thread.currentThread().getName());
 }
 });
 observable.subscribe(
 new Consumer<String>() {
 @Override
 public void accept(String s) throws Throwable {
 System.err.println("333 subscribe:" + s + Thread.currentThread().getName());
 getHw();
 }
 },
 throwable -> System.err.println("Timeout occurred!")
 );
// Thread.sleep(3000);
 }

 private static <@NonNull T> String readFile() throws InterruptedException {
 Thread.sleep(1000);
 return "Hello world! ";
 }

 private static String getHw() throws InterruptedException {
 Thread.sleep(1000);
 System.out.println("hhhwww");
 return "Hello world! ";
 }


}