[1장] 예제 11 구현

audxo112 · audxo112 · commit faade98bc9e9 · 2023-04-11T22:21:10.000+09:00
diff --git a/src/main/kotlin/kms/chapter01/L11_FlowableSample.kt b/src/main/kotlin/kms/chapter01/L11_FlowableSample.kt
@@ -0,0 +1,53 @@
+package kms.chapter01
+
+import io.reactivex.BackpressureStrategy
+import io.reactivex.Flowable
+import io.reactivex.FlowableEmitter
+import io.reactivex.FlowableOnSubscribe
+import io.reactivex.schedulers.Schedulers
+import org.reactivestreams.Subscriber
+import org.reactivestreams.Subscription
+
+fun main() {
+    val flowable = Flowable.create(
+        object : FlowableOnSubscribe<String> {
+            override fun subscribe(emitter: FlowableEmitter<String>) {
+                val strArr = arrayOf("Hello, World!", "안녕, RxJava!")
+                for (str in strArr) {
+                    if (emitter.isCancelled) {
+                        return
+                    }
+                    emitter.onNext(str)
+                }
+                emitter.onComplete()
+            }
+        },
+        BackpressureStrategy.BUFFER,
+    )
+    flowable
+        .observeOn(Schedulers.computation())
+        .subscribe(object : Subscriber<String> {
+
+            private var subscription: Subscription? = null
+
+            override fun onSubscribe(s: Subscription?) {
+                subscription = s
+                subscription?.request(1L)
+            }
+
+            override fun onNext(data: String?) {
+                println("${Thread.currentThread().name}: $data")
+                subscription?.request(1L)
+            }
+
+            override fun onError(error: Throwable?) {
+                error?.printStackTrace()
+            }
+
+            override fun onComplete() {
+                println("${Thread.currentThread().name}: 완료")
+            }
+        })
+
+    Thread.sleep(500L)
+}
