Reactive Programming - 비동기적으로 Data Stream을 다루는 방법

import io.reactivex.Observable;

public class RxJavaExample {
    public static void main(String[] args) {
        // Observable 생성
        Observable<String> observable = Observable.create(emitter -> {
            emitter.onNext("Hello");
            emitter.onNext("RxJava");
            emitter.onComplete();
        });

        // Observer 생성 및 구독
        observable.subscribe(
            item -> System.out.println("Next : " + item),    // onNext
            error -> System.err.println("Error : " + error),    // onError
            () -> System.out.println("Completed")    // onComplete
        );
    }
}

import io.reactivex.Observable;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;

public class RxJavaBasicExample {
    public static void main(String[] args) {
        // Observable 생성
        Observable<String> observable = Observable.create(emitter -> {
            emitter.onNext("Hello");
            emitter.onNext("RxJava");
            emitter.onComplete();
        });

        // Observer 생성
        Observer<String> observer = new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                System.out.println("Subscribed");
            }

            @Override
            public void onNext(String s) {
                System.out.println("Next : " + s);
            }

            @Override
            public void onError(Throwable e) {
                System.out.println("Error : " + e.getMessage());
            }

            @Override
            public void onComplete() {
                System.out.println("Completed");
            }
        };

        // Observable 구독
        observable.subscribe(observer);
    }
}

import io.reactivex.Observable;
import io.reactivex.schedulers.Schedulers;
import io.reactivex.observers.DisposableObserver;

public class RxJavaOperatorExample {
    public static void main(String[] args) {
        Observable<Integer> observable = Observable.fromArray(1, 2, 3, 4, 5)
                .filter(num -> num % 2 == 0)    // 짝수 filtering
                .map(num -> num * 2)    // 각 값을 2배로 변환
                .subscribeOn(Schedulers.io())    // IO scheduler에서 실행
                .observeOn(Schedulers.single());    // 단일 thread에서 관찰

        observable.subscribeWith(new DisposableObserver<Integer>() {
            @Override
            public void onNext(Integer integer) {
                System.out.println("Next : " + integer);
            }

            @Override
            public void onError(Throwable e) {
                System.out.println("Error : " + e.getMessage());
            }

            @Override
            public void onComplete() {
                System.out.println("Completed");
            }
        });

        // 잠시 대기하여 비동기 작업이 완료되도록 함
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

import io.reactivex.Observable;
import io.reactivex.schedulers.Schedulers;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.observers.DisposableObserver;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.Response;

public class RxJavaNetworkExample {
    public static void main(String[] args) {
        OkHttpClient client = new OkHttpClient();

        Observable<String> observable = Observable.create(emitter -> {
            try {
                Request request = new Request.Builder()
                        .url("https://api.github.com")
                        .build();
                Response response = client.newCall(request).execute();
                if (response.isSuccessful()) {
                    emitter.onNext(response.body().string());
                    emitter.onComplete();
                } else {
                    emitter.onError(new Exception("Network request failed"));
                }
            } catch (Exception e) {
                emitter.onError(e);
            }
        });

        observable.subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribeWith(new DisposableObserver<String>() {
                    @Override
                    public void onNext(String response) {
                        System.out.println("Response : " + response);
                    }

                    @Override
                    public void onError(Throwable e) {
                        System.out.println("Error : " + e.getMessage());
                    }

                    @Override
                    public void onComplete() {
                        System.out.println("Completed");
                    }
                });

        // 잠시 대기하여 비동기 작업이 완료되도록 함
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

import io.reactivex.rxjava3.core.Observable;
import java.util.Timer;
import java.util.TimerTask;

public class StockPriceFetcher {
    public static Observable<Double> getStockPriceStream() {
        return Observable.create(emitter -> {
            Timer timer = new Timer();
            timer.scheduleAtFixedRate(new TimerTask() {
                @Override
                public void run() {
                    try {
                        // 주식 가격을 가져오는 비동기 작업
                        double stockPrice = fetchStockPrice();
                        emitter.onNext(stockPrice);
                    } catch (Exception e) {
                        emitter.onError(e);
                    }
                }
            }, 0, 1000);    // 1초마다 주식 가격 update
        });
    }

    private static double fetchStockPrice() {
        // 여기서는 예시로 random 값을 가져오지만, 실제로는 외부 API를 호출하는 등의 작업이 필요함
        return 100 + Math.random() * 10;
    }
}

public class StockPriceUI {
    public static void main(String[] args) {
        Observable<Double> stockPriceStream = StockPriceFetcher.getStockPriceStream();

        stockPriceStream.subscribe(
            price -> updateStockPriceUI(price),
            error -> System.err.println("Error : " + error),
            () -> System.out.println("Stock price stream completed")
        );

        // 계속 실행되도록 main thread 대기시키기
        try {
            Thread.sleep(Long.MAX_VALUE);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private static void updateStockPriceUI(Double price) {
        // 실제로는 UI를 update하는 code가 필요함
        System.out.println("Updated Stock Price : " + price);
    }
}