Added SubscribeOnLifecycle

Author: li2

.idea/misc.xml

@@ -5,15 +5,10 @@
 @file:Suppress("unused")
 package me.li2.android.common.rx
 
-import androidx.lifecycle.Lifecycle
-import androidx.lifecycle.LifecycleObserver
-import androidx.lifecycle.OnLifecycleEvent
 import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers
 import io.reactivex.rxjava3.core.Completable
-import io.reactivex.rxjava3.core.Flowable
 import io.reactivex.rxjava3.core.Observable
 import io.reactivex.rxjava3.core.Single
-import io.reactivex.rxjava3.disposables.Disposable
 import io.reactivex.rxjava3.schedulers.Schedulers
 import java.util.concurrent.TimeUnit
 
@@ -38,42 +33,3 @@ fun <T> Single<T>.forUi(): Single<T> =
 fun Completable.forUi(): Completable =
         this.observeOn(AndroidSchedulers.mainThread())
                 .subscribeOn(Schedulers.io())
-
-/**
- * Subscribe on lifecycle onStart and dispose on lifecycle onStop.
- */
-fun <T> Observable<T>.subscribeOnLifecycle(lifecycle: Lifecycle, block: (T) -> Unit) {
-    val lifecycleObserver: LifecycleObserver = object : LifecycleObserver {
-        private var subscription: Disposable? = null
-
-        @OnLifecycleEvent(Lifecycle.Event.ON_START)
-        fun onStart() {
-            subscription = subscribe(block)
-        }
-
-        @OnLifecycleEvent(Lifecycle.Event.ON_STOP)
-        fun onStop() {
-            subscription?.dispose()
-        }
-    }
-
-    lifecycle.addObserver(lifecycleObserver)
-}
-
-fun <T> Flowable<T>.subscribeOnLifecycle(lifecycle: Lifecycle, block: (T) -> Unit) {
-    val lifecycleObserver: LifecycleObserver = object : LifecycleObserver {
-        private var subscription: Disposable? = null
-
-        @OnLifecycleEvent(Lifecycle.Event.ON_START)
-        fun onStart() {
-            subscription = subscribe(block)
-        }
-
-        @OnLifecycleEvent(Lifecycle.Event.ON_STOP)
-        fun onStop() {
-            subscription?.dispose()
-        }
-    }
-
-    lifecycle.addObserver(lifecycleObserver)
-}

common/src/main/java/me/li2/android/common/rx/ObservableExt.kt

@@ -11,6 +11,7 @@ import android.bluetooth.BluetoothAdapter.*
 import android.content.Context
 import androidx.fragment.app.FragmentActivity
 import io.reactivex.rxjava3.core.Observable
+import me.li2.android.common.framework.onBroadcast
 import me.li2.android.common.logic.orFalse
 
 /**

common/src/main/java/me/li2/android/common/rx/RxBluetooth.kt

@@ -0,0 +1,61 @@
+package me.li2.android.common.rx
+
+import androidx.lifecycle.Lifecycle
+import androidx.lifecycle.LifecycleObserver
+import androidx.lifecycle.OnLifecycleEvent
+import io.reactivex.rxjava3.core.Flowable
+import io.reactivex.rxjava3.core.Observable
+import io.reactivex.rxjava3.disposables.Disposable
+
+/*
+ * Created by Weiyi Li on 12/05/21.
+ * https://github.com/li2
+ */
+
+/**
+ * Subscribe on [Lifecycle.Event.ON_START] and dispose on [Lifecycle.Event.ON_STOP] automatically.
+ *
+ * @param lifecycle an instance of [androidx.lifecycle.LifecycleRegistry], for example, [androidx.activity.ComponentActivity.getLifecycle]
+ * @param onNext the consumer to accept emissions from the current Observable.
+ */
+fun <T> Observable<T>.subscribeOnLifecycle(lifecycle: Lifecycle, onNext: (T) -> Unit) {
+    val lifecycleObserver: LifecycleObserver = object : LifecycleObserver {
+        private var subscription: Disposable? = null
+
+        @OnLifecycleEvent(Lifecycle.Event.ON_START)
+        fun subscribe() {
+            subscription = subscribe(onNext)
+        }
+
+        @OnLifecycleEvent(Lifecycle.Event.ON_STOP)
+        fun dispose() {
+            subscription?.dispose()
+        }
+    }
+
+    lifecycle.addObserver(lifecycleObserver)
+}
+
+/**
+ * Subscribe on [Lifecycle.Event.ON_START] and dispose on [Lifecycle.Event.ON_STOP] automatically.
+ *
+ * @param lifecycle an instance of [androidx.lifecycle.LifecycleRegistry], for example, [androidx.activity.ComponentActivity.getLifecycle]
+ * @param onNext the consumer to accept emissions from the current Flowable.
+ */
+fun <T> Flowable<T>.subscribeOnLifecycle(lifecycle: Lifecycle, onNext: (T) -> Unit) {
+    val lifecycleObserver: LifecycleObserver = object : LifecycleObserver {
+        private var subscription: Disposable? = null
+
+        @OnLifecycleEvent(Lifecycle.Event.ON_START)
+        fun subscribe() {
+            subscription = subscribe(onNext)
+        }
+
+        @OnLifecycleEvent(Lifecycle.Event.ON_STOP)
+        fun dispose() {
+            subscription?.dispose()
+        }
+    }
+
+    lifecycle.addObserver(lifecycleObserver)
+}