RxJAVA - Subject 간단 정리

* 개인공부를 하며 정리한 내용이므로, 부정확하거나 사실과는 다른 내용이 있을수도 있습니다.

  해당사항에 대해 의견을 주시면 겸허히 수정하도록 하겠습니다.

---

Observable과 Observer

 Observable을 Observer가 받아서 '반응'합니다.

Observable을 단독으로 쓰진 않고, Observer이자 Observable인 subject를 주로 사용합니다.

 

Cold / Hot Observer

Hot : 생성하자마자 바로 흘려보냄

• 구독자의 존재여부 상관 x
• 구독자를 선택적으로 고려 가능
• 구독 시점으로부터 발행하는 값을 받는게 기본
• 마우스, 키보드, 시스템 이벤트 등의 주로 사용 (멀티캐스팅도 포함)

Cold : 요청하고 결과를 받아옴.

• 일반적인 옵저버
• 구독하지 않다면, 데이터를 흘리지도 않음.
• 요청시에만 값을 받는다.

 

 

Subject

Onservable와 Subject 구조 파해치기 https://sujinnaljin.medium.com/rxswift-subject-99b401e5d2e5

 

 

 

 

publishSubject

 초기값이 없으며, 구독 이후 배출한 항목들만 옵저버에게 발행한다.

단, 생성시점과 구독시점의 차이가 있다면, 그 차이기간동안의 발행된 항목은 받지 못할 수 있다.

따라서, 만약 값을 모두 받아야 한다면 Create를 명시적으로 사용해 명시적으로 cold observable을 사용하거나, ReplaySubject를 사용해야 한다

 

 

AsyncSubject

동작이 완료된 후에 동작하며, 마지막 값만 출력한다.

 

 

BehaviorSubject

초기값이 존재하는 Observable이다. 구독 이후 값을 발행하며, 이후 발행된 항목을 계속 발행한다.

 

 

 

 

RePlaySubject

구독시점과 관계없이 Observable이 발행한 모든 항목을 발행해준다.

단, 버퍼의 크기가 매우 커지거나, 첫 발행이후 시간이 오래 경과한다면 항목들을 잃을 수 있다.

멀티스레드 환경에서 이는 모호함ㅇ과 observavle 계약 위반을 유발할수있기 때문에, on 또는 onNext와 같이 비순차적인 호출을 유발시키는 메를 사용하지 않아야 한다.

• observable Contract : http://reactivex.io/documentation/ko/contract.html

 

---

Example

 

//CallManager.java
public static BehaviorSubject<Integer> state = BehaviorSubject.create();

//IncallService에 대한 변경사항을 알리는 콜백을 정의
    private Object callback = new Call.Callback(){
        @Override
        public void onStateChanged(Call call, int newState) {
            super.onStateChanged(call, newState);
            //데이터의발행을 알림
            state.onNext(newState);
        }
    };

통화 상태에 대한 정보(callback)이 변경될때마다 state를 현재 값으로 갱신합니다.

 

 

//CallAcitivity.java

@Override
    protected void onStart() {
        super.onStart();
//        assert updateUi(-1) != null;
        disposable.add(
                CallManager.state
                        //onNext 이벤트 처리 (발행에 대한 처리)
                        .subscribe(new Consumer<Integer>() {
                            @Override
                            public void accept(Integer integer) throws Throwable {
                                updateUi(integer);
                            }
                        }));


        disposable.add(
                CallManager.state
                        .filter(new Predicate<Integer>() {
                            @Override
                            public boolean test(Integer integer) throws Exception {
                                return integer == Call.STATE_DISCONNECTED;
                            }
                        })
                        .delay(1, TimeUnit.SECONDS)
                        .firstElement()
                        .subscribe(new Consumer<Integer>() {
                            @Override
                            public void accept(Integer integer)throws Throwable{
                                finish();
                            }
                        }));
    }

disposable을 이용해 state를 구독받아 알맞은 작업을 처리하는 코드입니다.