[1장] Flowable 예제 설명 및 추가 정리

audxo112 · audxo112 · commit aa27d7bf6a24 · 2023-04-11T23:28:25.000+09:00
diff --git a/src/main/kotlin/kms/chapter01/L11_FlowableSample.kt b/src/main/kotlin/kms/chapter01/L11_FlowableSample.kt
@@ -8,42 +8,65 @@ import io.reactivex.schedulers.Schedulers
 import org.reactivestreams.Subscriber
 import org.reactivestreams.Subscription
 
-fun main() {
+private fun main() {
     val flowable = Flowable.create(
         object : FlowableOnSubscribe<String> {
+
+            // FlowableEmitter 가 Subscriber 에게 데이터를 통지한다
+            // create 의 구현을 따라가면
+            // 에러가 발생하면 catch 해서 onError 로 전달하는 부분이 존재함
+            // 단, 치명적인 에러라면 다시 Throw 를 던짐
+            // https://github.com/ReactiveX/RxJava/issues/748#issuecomment-32471495
             override fun subscribe(emitter: FlowableEmitter<String>) {
                 val strArr = arrayOf("Hello, World!", "안녕, RxJava!")
                 for (str in strArr) {
+                    // 구동 해지 상태에서 종료하지 않고 onNext 를 진행해도 데이터를 통지하지 않는다
+                    // 단, Rx 에서 해주는 것은 통지하지 않는 것이지 계속 진행은 하므로 직접 처리하는게 좋다
                     if (emitter.isCancelled) {
                         return
                     }
+                    // 데이터를 통지한다
+                    // 만약 null 을 전달하면 NullPointException 이 발생한다
                     emitter.onNext(str)
                 }
+                // onComplete 를 통지하면 그 이후엔 아무것도 통지하면 안된다
                 emitter.onComplete()
             }
         },
+        // BackpressureStrategy 에 따라 다른 Emitter 를 생성
         BackpressureStrategy.BUFFER,
     )
     flowable
+        // 데이터를 받는 측의 쓰레드를 변경할 때 사용
         .observeOn(Schedulers.computation())
+        // Flowable 는 Publisher 인터페이스를 구현했기 때문에 Subscriber 와의 상호작용을 외부에서 영향을 받지 않는다
         .subscribe(object : Subscriber<String> {
 
+            // Subscriber 가 받을 데이터의 개수를 요청 및 구독 해지할 수 있는 인터페이스
+            // onNext에서 직접 배압을 처리하기 위해서 subscription 을 멤버 변수로 저장
             private var subscription: Subscription? = null
 
             override fun onSubscribe(s: Subscription?) {
                 subscription = s
+                // 요청 데이터의 개수를 MAX 로 처리하면 onNext 에서 더 이상 요청하지 않아도 됨
+                // onSubscribe 에서 request를 호출하지 않으면 데이터르 받을 수 없다
+                // request는 onSubscribe 의 가장 마지막에서 호출 해야함
                 subscription?.request(1L)
             }
 
+            // Flowable 에서 데이터를 받으면 호출 되는 메서드
             override fun onNext(data: String?) {
                 println("${Thread.currentThread().name}: $data")
                 subscription?.request(1L)
             }
 
+            // 에러가 발생했거나 에러를 통지할 때 실행되는 메서드
+            // onError 이후에는 onNext 나 onComplete 가 실행되지 않는다
             override fun onError(error: Throwable?) {
                 error?.printStackTrace()
             }
 
+            // 모든 데이터의 통지를 끝내고 처리가 완료됐을 때 실행되는 메서드
             override fun onComplete() {
                 println("${Thread.currentThread().name}: 완료")
             }
diff --git a/src/main/kotlin/kms/chapter01/README.md b/src/main/kotlin/kms/chapter01/README.md
@@ -104,3 +104,31 @@
 >- 이미 구독하고 있던 소비자는 버퍼에 있는데이터를 통지 받음
 >- 새로 구독한 소비자는 최신 데이터를 통지 받음
 
+### Flowable 시퀀스
+
+1. Subscriber 가 Flowable을 구독한다(**Subscribe**)
+2. Flowable이 **Subscription**을 생성한다
+3. Flowable이 Subscriber에 구독 시작(**onSubscribe**)을 통지하여 Subscription을 전달
+4. Subscriber는 Subscription에 데이터를 통지하게 요청
+5. Flowable은 데이터를 Subscriber에게 통지
+6. Subscriber는 받은 데이터를 처리한다
+7. 처리한 후 Subscriber는 Subscription에게 데이터 통지를 요청
+8. Flowable은 데이터가 있다면 그 데이터를 Subscriber에게 통지
+9. Subscriber는 받은 데이터를 처리한다
+10. 처리한 후 Subscriber는 Subscription에게 데이터 통지를 요청
+11. Flowable 은 데이터가 없다면 완료(onComplete)를 통지
+12. Subscriber는 완료를 처리한다
+
+### BackpressureStrategy 종류
+
+- **BUFFER**
+  - 통지할 수 있을 때까지 모든 데이터를 버퍼링 한다
+- **DROP**
+  - 통지할 수 있을 때까지 새로 생성한 데이터를 삭제한다
+- **LATEST**
+  - 생성한 최신 데이터만 버퍼링하고 생성할 때마다 버퍼링하는 데이터를 교환한다
+- **ERROR**
+  - 버퍼 크기를 초과하면 MissingBackPressureException 에러를 통지
+- **NONE**
+  - 특정 처리를 수행하지 않는다.
+  - onBackPressure로 시작하는 메서드로 배압 모드를 설정할 때 사용
