Realm 개발자 김용욱님 RxJava의 PublishSubject 내부 구조를 GDG Korea 안드로이드 Rx 세미나에서 공유한 내용입니다.

RxJava의 PublishSubject 알아보기

RxJava의 Subject에 대한 궁금증이 많아 대표적인 Subject인 PublishSubject부터 살펴보았습니다.

Subject<T, R>

Observable<R>를 상속받고 Observer`를 구현하는 추상 클래스입니다.

이는 두가지 메서드 hasObservers와 toSerialized를 가지고 있습니다. 전자는 서크스크라이브되고 있는지 여부이고 후자는 SerializedSubject로 래핑을 해서 여러 스레드에서 onNext 등을 호출해도 안전하게 합니다.

SerializedSubject

생성자는 서브스크라이브때 사용될 OnSubscribe<R> 정의하며 서브스크라이브의 onStart()를 호출하며 RxJavaHooks에 등록된 콜백을 통해 onObservableStart와 onObservableReturn등의 훅을 연동시킵니다.

Subject를 Subject 타입인 actual과 SerializedObserver 타입인 observer에 같이 넣어두고 SerailizedSubject가 오버라이드 해야할 기능들은 actual과 osberver를 위임하여 수행합니다.

SerializedObserver

직렬화된 실행은 SerializedObserver가 담당하며 서브스크라이브에 onNext, onError, onCompleted를 전달합니다. 내부에 리스트를 가지고 있어서 직렬화를 합니다. 단일 리스트를 가지고 있기 때문에 상태를 관리하기 위해 빈 Object인 ON_COMPLETED_SENTINEL, ON_NEXT_NULL_SENTINEL과 에러를 위한 래퍼 클래스 OnErrorSentinel 등을 사용합니다.

PublishSubject

PublishSubject에 대해 onNext 요청이 들어오면 SubjectSubscriptionManager 타입인 state를 통해 SubjectObserver를 거쳐 onNext를 호출하며 onError나 onCompleted의 경우 state의 여부를 확인하여 emitNext를 통해 호출합니다. 리스트에 있는 항목을 먼저 수행할지 늦게 수행할지 여부에 따라 emitNext와 emitFirst가 존재합니다.

SubjectObserver

---

발표 슬라이드는 아래를 참고해주세요