// Publisher: 이벤트 발행 수행
@FunctionalInterface
public interface Publisher<T> {
    void subscribe(Subscriber<? super T> s);
}

// Subscriber: 자신을 Publisher 에게 등록 요청해 이벤트 소비
// 호출 순서 : onSubscribe -> onNext (여러번 호출 가능) -> onError 또는 onComplete 
public interface Subscriber<T> {
    void onSubscribe(Subscription s);
    void onNext(T t);
    void onError(Throwable t);
    void onComplete();
}

// Subscription: Publisher와 Subscriber 사이의 제어 흐름, 역압력을 관리한다.
public interface Subscription {
    void request(long n);
    void cancel();
}

// Processor: 리액티브 스트림에서 처리하는 이벤트를 가공, 변환할 때 사용된다. 
// 들어오는 메시지를 변형시켜 다음 구독자 한테 넘기기 위해 사용. (중간자 역할이며, 구독자와 발행자 두 역할 수행)
public interface Processor<T, R> extends Subscriber<T>, Publisher<R> { }

import java.util.concurrent.Flow.*;
public class Main {
    public static void main( String[] args ) {
        // Subscriber를 인수로 받아 Subscriber의 onSubscribe 메서드를 호출
        getTemperatures( "New York" ).subscribe( new TempSubscriber() );
    }
    private static Publisher<TempInfo> getTemperatures( String town ) {
        // 아래 코드 익숙하지가 않네요...
        return subscriber -> subscriber.onSubscribe(
            new TempSubscription( subscriber, town ) );
    }
}

import java.util.concurrent.Flow.*;
public class TempSubscriber implements Subscriber<TempInfo> {
    private Subscription subscription;
    @Override
    public void onSubscribe( Subscription subscription ) {
        this.subscription = subscription; // 구독 저장
        subscription.request( 1 ); // 1번째 요청
    }
    @Override
    public void onNext( TempInfo tempInfo ) {
        // Subscription이 전달한 온도를 출력하고 새 레포트를 다시 요청
        System.out.println( tempInfo );
        subscription.request( 1 );
    }
    @Override
    public void onError( Throwable t ) {
        System.err.println(t.getMessage());
    }
    @Override
    public void onComplete() {
        System.out.println("Done!");
    }
}

import java.util.concurrent.Flow.*;
public class TempSubscription implements Subscription {
    private final Subscriber<? super TempInfo> subscriber;
    private final String town;
    public TempSubscription( Subscriber<? super TempInfo> subscriber,
                            String town ) {
        this.subscriber = subscriber;
        this.town = town;
    }
    @Override
    public void request( long n ) {
        // 요청한 이벤트 수 만큼 처리해준다 
        for (long i = 0L; i < n; i++) {
            try {
                subscriber.onNext( TempInfo.fetch( town ) );
            } catch (Exception e) {
                subscriber.onError( e );
                break;
            }
        }
    }
    @Override
    public void cancel() {
        subscriber.onComplete();
    }
}

import java.util.Random;
public class TempInfo {
    public static final Random random = new Random();
    private final String town;
    private final int temp;
    public TempInfo(String town, int temp) {
        this.town = town;
        this.temp = temp;
    }
    // 특정 타운의 온도를 리턴
    public static TempInfo fetch(String town) {
        if (random.nextInt(10) == 0)
            throw new RuntimeException("Error!");
        return new TempInfo(town, random.nextInt(100));
    }
    @Override
    public String toString() {
        return town + " : " + temp;
    }
    public int getTemp() {
        return temp;
    }
    public String getTown() {
        return town;
    }
}

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class TempSubscription implements Subscription {
    private static final ExecutorService executor =
        Executors.newSingleThreadExecutor();
    @Override
    public void request( long n ) {
        // 다른스레드에서 다음요소를 구독자에게 보낸다.
        executor.submit( () -> {
            for (long i = 0L; i < n; i++) {
                try {
                    subscriber.onNext( TempInfo.fetch( town ) );
                } catch (Exception e) {
                    subscriber.onError( e );
                    break;
                }
            }
        });
    }
}

import java.util.concurrent.Flow.*;
public class Main {
    public static void main( String[] args ) {
        getCelsiusTemperatures( "New York" )
            .subscribe( new TempSubscriber() );
    }
    public static Publisher<TempInfo> getCelsiusTemperatures(String town) {
        return subscriber -> {
            TempProcessor processor = new TempProcessor();
            processor.subscribe( subscriber );
            processor.onSubscribe( new TempSubscription(processor, town) );
        };
    }
}

import java.util.concurrent.Flow.*;
// 화씨를 섭씨로 변환
public class TempProcessor implements Processor<TempInfo, TempInfo> {
    private Subscriber<? super TempInfo> subscriber;
    @Override
    public void subscribe( Subscriber<? super TempInfo> subscriber ) {
        this.subscriber = subscriber;
    }
    
    // 로직을 포함하는 유일한 메서드
    @Override
    public void onNext( TempInfo temp ) {
        // 섭씨로 변환한 다음 TempInfo 를 다시 전송
        subscriber.onNext( new TempInfo( temp.getTown(),
                                        (temp.getTemp() - 32) * 5 / 9) ); 
    }
    
    @Override
    public void onSubscribe( Subscription subscription ) {
        // 업스트림 구독자에 전달 
        subscriber.onSubscribe( subscription );
    }
    @Override
    public void onError( Throwable throwable ) {
        // 업스트림 구독자에 전달 
        subscriber.onError( throwable );
    }
    @Override
    public void onComplete() {
        // 업스트림 구독자에 전달 
        subscriber.onComplete();
    }
}

// Observable의 구독자는 onNext("first"), onNext("second"), onComplete()의 순서로 메시지를 받는다.
Observable<String> strings = Observable.just( "first", "second" );
Disposable disposable = strings.subscribe(System.out::println); // 구독하기 
disposable.dispose(); // 구독 취소 
System.out.println(disposable.isDisposed());

// main 프로그램은 실행하자 마자 따로 실행할 코드가 없으므로 바로 종료되고
// 프로그램이 종료되었으므로 어떤 결과를 출력하기도 전에 데몬 스레드도 종료되면서
// 이런 현상이 일어난다ㄴ
Observable<Long> onePerSec = Observable.interval(1, TimeUnit.SECONDS);
onePerSec.subscribe(i -> System.out.println(TempInfo.fetch( "New York" )));
// onePerSec.blockingSubscribe(i -> System.out.println(TempInfo.fetch( "New York" )));

// getTemperature 메서드가 반환하는 Observable에  TempObserver 를 구독시킨다.
public static void main(String[] args) {
    // 매초마다 뉴욕 온도 방출하는 Oberserverble 생성
    Observable<TempInfo> observable = getTemperature( "New York" );
    // Observable에 가입해서 온도 출력
    observable.blockingSubscribe( new TempObserver() );
}

public static Observable<TempInfo> getTemperature(String town) {
    return Observable.create(emitter ->
                             Observable.interval(1, TimeUnit.SECONDS)
                             .subscribe(i -> {
                                 if (!emitter.isDisposed()) {
                                     // 편의상 다섯 번만
                                     if ( i >= 5 ) {
                                         emitter.onComplete();
                                     } else {
                                         try {
                                             emitter.onNext(TempInfo.fetch(town));
                                         } catch (Exception e) {
                                             emitter.onError(e);
                                         }
                                     }
                                 }}));
}

import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;
public class TempObserver implements Observer<TempInfo> {
    @Override
    public void onComplete() {
        System.out.println( "Done!" );
    }
    @Override
    public void onError( Throwable throwable ) {
        System.out.println( "Got problem: " + throwable.getMessage() );
    }
    @Override
    public void onSubscribe( Disposable disposable ) {
    }
    @Override
    public void onNext( TempInfo tempInfo ) {
        System.out.println( tempInfo );
    }
} 

// 섭씨를 화씨로
public static Observable<TempInfo> getCelsiusTemperature(String town) {
    return getTemperature( town )
        .map( temp -> new TempInfo( temp.getTown(),
                                   (temp.getTemp() - 32) * 5 / 9) );
}

// 필터에 만족하는 Observable들만 가져온다. 
// 아래는 동상에 걸릴 위험이 있을 때 알려주는 경고 시스템 사례 
public static Observable<TempInfo> getNegativeTemperature(String town) {
    return getCelsiusTemperature( town )
        .filter( temp -> temp.getTemp() < 0 ); // 온도가 섭씨 0도 이하일 때만
}

public class Main {
    public static void main(String[] args) {
        Observable<TempInfo> observable = getCelsiusTemperatures(
            "New York", "Chicago", "San Francisco" );
        observable.blockingSubscribe( new TempObserver() );
    }
}

public static Observable<TempInfo> getCelsiusTemperatures(String... towns) {
    return Observable.merge(Arrays.stream(towns)
                            .map(TempObservable::getCelsiusTemperature)
                            .collect(toList()));
}