[#289] Added interop-rx3 for Rxjava3 support

CHANGELOG.md

@@ -2,7 +2,7 @@
 
 ## Pending changes
 
-–
+- [#289](https://github.com/bumble-tech/appyx/issues/289) – **Added**: Introduced `interop-rx3` for RxJava 3 support. This has identical functionality to `interop-rx2`. 
 
 ---
 

documentation/releases/downloads.md

@@ -38,8 +38,9 @@ dependencies {
 
 ```groovy
 dependencies {
-    // Optional support for RxJava 2
+    // Optional support for RxJava 2/3
     implementation "com.bumble.appyx:interop-rx2:$version"
+    implementation "com.bumble.appyx:interop-rx3:$version"
 
     // Optional interoperability layer between Appyx and badoo/RIBs
     // You have to add https://jitpack.io repository to use it because badoo/RIBs is hosted there

gradle/libs.versions.toml

@@ -13,8 +13,6 @@ ribs = "0.36.1"
 mvicore = "1.2.6"
 coroutines = "1.6.4"
 kotlin = "1.7.10"
-rxjava2 = "2.2.21"
-rxandroid = "2.1.1"
 junit5 = "5.8.2"
 detekt = "1.21.0"
 dependencyAnalysis = "1.13.1"
@@ -57,11 +55,13 @@ ribs-base-test = { module = "com.github.badoo.RIBs:rib-base-test", version.ref =
 ribs-base-test-activity = { module = "com.github.badoo.RIBs:rib-base-test-activity", version.ref = "ribs" }
 ribs-base-test-rx2 = { module = "com.github.badoo.RIBs:rib-base-test-rx2", version.ref = "ribs" }
 ribs-compose = { module = "com.github.badoo.RIBs:rib-compose", version.ref = "ribs" }
-ribs-rx = { module = "com.github.badoo.RIBs:rib-rx2", version.ref = "ribs" }
 
-rxjava2 = { module = "io.reactivex.rxjava2:rxjava", version.ref = "rxjava2" }
-rxandroid = { module = "io.reactivex.rxjava2:rxandroid", version.ref = "rxandroid" }
-rxrelay = "com.jakewharton.rxrelay2:rxrelay:2.1.1"
+rxjava2 = "io.reactivex.rxjava2:rxjava:2.2.21"
+rxjava3 = "io.reactivex.rxjava3:rxjava:3.1.5"
+rxandroid2 = "io.reactivex.rxjava2:rxandroid:2.1.1"
+rxandroid3 = "io.reactivex.rxjava3:rxandroid:3.0.2"
+rxrelay2 = "com.jakewharton.rxrelay2:rxrelay:2.1.1"
+rxrelay3 = "com.jakewharton.rxrelay3:rxrelay:3.0.1"
 
 kotlin-test = { module = "org.jetbrains.kotlin:kotlin-test", version.ref = "kotlin" }
 junit-api = { module = "org.junit.jupiter:junit-jupiter-api", version.ref = "junit5" }

libraries/interop-rx2/build.gradle.kts

@@ -26,7 +26,7 @@ android {
 dependencies {
     api(project(":libraries:core"))
     api(libs.rxjava2)
-    api(libs.rxrelay)
+    api(libs.rxrelay2)
 
     implementation(libs.kotlin.coroutines.rx2)
     implementation(libs.androidx.lifecycle.java8)

libraries/interop-rx3/build.gradle.kts

@@ -0,0 +1,35 @@
+plugins {
+    id("com.android.library")
+    id("kotlin-android")
+    id("appyx-publish-android")
+    id("appyx-lint")
+    id("appyx-detekt")
+}
+
+android {
+    namespace = "com.bumble.appyx.interop.rx3"
+    compileSdk = libs.versions.androidCompileSdk.get().toInt()
+
+    defaultConfig {
+        minSdk = libs.versions.androidMinSdk.get().toInt()
+        targetSdk = libs.versions.androidTargetSdk.get().toInt()
+
+        testInstrumentationRunner = "androidx.test.runner.AndroidJUnitRunner"
+    }
+    testOptions {
+        unitTests.all {
+            it.useJUnitPlatform()
+        }
+    }
+}
+
+dependencies {
+    api(project(":libraries:core"))
+    api(libs.rxjava3)
+    api(libs.rxrelay3)
+
+    implementation(libs.androidx.lifecycle.java8)
+
+    testImplementation(libs.junit.api)
+    testRuntimeOnly(libs.junit.engine)
+}

libraries/interop-rx3/lint-baseline.xml

@@ -0,0 +1,4 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<issues format="6" by="lint 7.3.1" type="baseline" client="gradle" dependencies="false" name="AGP (7.3.1)" variant="all" version="7.3.1">
+
+</issues>

libraries/interop-rx3/src/main/kotlin/com/bumble/appyx/interop/rx3/connectable/Connectable.kt

@@ -0,0 +1,9 @@
+package com.bumble.appyx.interop.rx3.connectable
+
+import com.bumble.appyx.core.plugin.NodeLifecycleAware
+import com.jakewharton.rxrelay3.Relay
+
+interface Connectable<Input, Output> : NodeLifecycleAware {
+    val input: Relay<Input>
+    val output: Relay<Output>
+}

libraries/interop-rx3/src/main/kotlin/com/bumble/appyx/interop/rx3/connectable/NodeConnector.kt

@@ -0,0 +1,60 @@
+package com.bumble.appyx.interop.rx3.connectable
+
+import androidx.lifecycle.Lifecycle
+import com.bumble.appyx.core.lifecycle.subscribe
+import com.jakewharton.rxrelay3.PublishRelay
+import com.jakewharton.rxrelay3.Relay
+import io.reactivex.rxjava3.core.Observer
+
+class NodeConnector<Input, Output : Any>(
+    override val input: Relay<Input> = PublishRelay.create(),
+) : Connectable<Input, Output> {
+
+    private val intake: Relay<Output> = PublishRelay.create()
+    private val exhaust: Relay<Output> = PublishRelay.create()
+    private var isFlushed = false
+    private val outputCache = mutableListOf<Output>()
+
+    override val output: Relay<Output> = object : Relay<Output>() {
+
+        override fun subscribeActual(observer: Observer<in Output>) {
+            exhaust.subscribe(observer)
+        }
+
+        override fun accept(value: Output) {
+            intake.accept(value)
+        }
+
+        override fun hasObservers() = exhaust.hasObservers()
+
+    }
+
+    override fun onCreate(lifecycle: Lifecycle) {
+        lifecycle.subscribe(onCreate = { flushOutputCache() })
+    }
+
+    private val cacheSubscription = intake.subscribe {
+        synchronized(this) {
+            if (!isFlushed) {
+                outputCache.add(it)
+            } else {
+                exhaust.accept(it)
+                switchToExhaust()
+            }
+        }
+    }
+
+    private fun flushOutputCache() {
+        synchronized(this) {
+            if (isFlushed) error("Already flushed")
+            isFlushed = true
+            outputCache.forEach { exhaust.accept(it) }
+            outputCache.clear()
+        }
+    }
+
+    private fun switchToExhaust() {
+        intake.subscribe { exhaust.accept(it) }
+        cacheSubscription.dispose()
+    }
+}

libraries/interop-rx3/src/main/kotlin/com/bumble/appyx/interop/rx3/plugin/DisposeOnDestroy.kt

@@ -0,0 +1,17 @@
+package com.bumble.appyx.interop.rx3.plugin
+
+import com.bumble.appyx.core.plugin.Destroyable
+import com.bumble.appyx.core.plugin.Plugin
+import io.reactivex.rxjava3.disposables.CompositeDisposable
+import io.reactivex.rxjava3.disposables.Disposable
+
+private class DisposeOnDestroy(disposables: List<Disposable>) : Destroyable {
+    private val disposable = CompositeDisposable(disposables)
+
+    override fun destroy() {
+        disposable.dispose()
+    }
+}
+
+fun disposeOnDestroyPlugin(vararg disposables: Disposable): Plugin =
+    DisposeOnDestroy(disposables.toList())