Reactive Streams

🙋‍♂️ API Components

---

1. Publisher (생산자)

• 데이터를 생성하고 통지한다.
• subscribe : 소비자를 등록할 때 호출

2. Subscriber (소비자)

• 통지된 데이터를 전달받아서 처리한다.
• onSubscribe : 데이터를 최초 통지할 때 호출
• onNext : 데이터를 통지할 때 마다 호출
• onError : 데이터 통지에서 에러가 발생했을 때 호출
• onComplete : 데이터 통지가 완료 되었을 때 호출

3. Subscription

• 구독 자체를 의미, 전달 받을 데이터의 개수를 요청하고 구독을 해지한다.
• request : 데이터 개수 요청
• cancel : 요청 취소

4. Processor

• Publisher와 Subscriber의 모든 기능이 있음.

🙋‍♂️ Publisher와 Subscriber간의 프로세스 흐름

---

• Subscriber(이하 소비자)는 Publisher(이하 생산자)를 구독한다.
• 생산자는 데이터를 통지할 준비가 되었음을 소비자에게 알린다.(onSubscribe)
• 소비자는 전달 받을 통지 데이터 개수를 생산자에게 요청한다(request).
• 생산자는 데이터를 생산한다. (계속 생산중)
• 생산자는 요청 받은 개수만큼 데이터를 통지한다(onNext)
• 완료 또는 에러가 발생할 때까지 데이터 생성, 통지, 요청을 계속한다.(onError , onComplete)

🙋‍♂️ Cold Publisher & Hot Publisher

---

1. Cold Publisher(차가운 생산자)

• 생산자는 소비자가 구독 할때마다 데이터를 처음부터 새로 통지한다.
• 데이터를 통지하는 새로운 타임 라인이 생성된다.
• 소비자는 구독 시점과 상관없이 통지된 데이터를 처음부터 전달 받을 수 있다.

public class ColdPublisherExample
{
    public static void main(String[] args)
    {
        Flowable<Integer> flowable = Flowable.just(1, 3, 5, 7);

        flowable.subscribe(data -> System.out.println("구독자1 " + data));
        flowable.subscribe(data -> System.out.println("구독자2 " + data));
    }
}

2. Hot Publisher(뜨거운 생산자)

• 생산자는 소비자 수와 상관없이 데이터를 한번만 통지한다.
• 데이터를 통지하는 타임 라인은 하나이다.
• 소비자는 발행된 데이터를 처음부터 전달 받는게 아니라 구독한 시점에 통지된 데이터들만 전달 받을 수 있다.

public class HotPublisherExample
{
    public static void main(String[] args)
    {
        PublishProcessor<Integer> processor = PublishProcessor.create();
        processor.subscribe(data -> System.out.println("구독자1 : " + data));
        processor.onNext(1);
        processor.onNext(3);

        processor.subscribe(data -> System.out.println("구독자1 : " + data));
        processor.onNext(5);
        processor.onNext(7);

        processor.onComplete();
    }
}