Fix #653 - Implement converters for RX Java 3

documentation/src/main/jekyll/guides/converters.adoc

@@ -22,12 +22,19 @@ Pick the right one and add it to your project:
   <version>{{ site.data.versions.mutiny_version }}</version>
 </dependency>
 
-<!-- Mutiny <-> RX Java -->
+<!-- Mutiny <-> RX Java 2 -->
 <dependency>
   <groupId>io.smallrye.reactive</groupId>
   <artifactId>mutiny-rxjava</artifactId>
   <version>{{ site.data.versions.mutiny_version }}</version>
 </dependency>
+
+<!-- Mutiny <-> RX Java 3 -->
+<dependency>
+  <groupId>io.smallrye.reactive</groupId>
+  <artifactId>mutiny-rxjava3</artifactId>
+  <version>{{ site.data.versions.mutiny_version }}</version>
+</dependency>
 ----
 
 == Integration with Project Reactor
@@ -97,17 +104,21 @@ include::{include_dir}/ReactorTest.java[tag=uni-export]
 include::{include_dir}/ReactorTest.java[tag=multi-export]
 ----
 
-== Integration with RX Java
+== Integration with RX Java 2 and RX Java 3
 
 RxJava is another popular reactive programming library.
 It offers 5 types: `Completable` (no item), `Single` (one item), `Maybe` (0 or 1 item), `Observable` (multiple items), `Flowable` (multiple items, implements Reactive Stream `Publisher`).
 
-To use the RxJava <-> Mutiny converter, add the following imports to your class:
+To use the RxJava <-> Mutiny converters, add the following imports to your class:
 
 [source, java, indent=0]
 ----
+// RX Java 2 - provided by mutiny-rxjava
 import io.smallrye.mutiny.converters.multi.MultiRxConverters;
 import io.smallrye.mutiny.converters.uni.UniRxConverters;
+// RX Java 3 - provided by mutiny-rxjava3
+import io.smallrye.mutiny.converters.multi.MultiRx3Converters;
+import io.smallrye.mutiny.converters.uni.UniRx3Converters;
 ----
 
 === Converting an Observable or a Flowable into a Multi

pom.xml

@@ -64,6 +64,7 @@
         <module>test-utils</module>
         <module>implementation</module>
         <module>rxjava</module>
+        <module>rxjava3</module>
         <module>reactor</module>
         <module>reactive-streams-operators</module>
         <module>documentation</module>
@@ -76,6 +77,7 @@
 
     <properties>
         <rxjava2.version>2.2.21</rxjava2.version>
+        <rxjava3.version>3.1.0</rxjava3.version>
         <reactor-core.version>3.4.9</reactor-core.version>
         <microprofile-reactive-streams.version>2.0</microprofile-reactive-streams.version>
         <microprofile-context-propagation.version>1.2</microprofile-context-propagation.version>
@@ -126,6 +128,11 @@
                 <artifactId>rxjava</artifactId>
                 <version>${rxjava2.version}</version>
             </dependency>
+            <dependency>
+                <groupId>io.reactivex.rxjava3</groupId>
+                <artifactId>rxjava</artifactId>
+                <version>${rxjava3.version}</version>
+            </dependency>
             <dependency>
                 <groupId>io.projectreactor</groupId>
                 <artifactId>reactor-core</artifactId>

rxjava3/pom.xml

@@ -0,0 +1,62 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>io.smallrye.reactive</groupId>
+        <artifactId>mutiny-project</artifactId>
+        <version>999-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>mutiny-rxjava3</artifactId>
+    <name>SmallRye Mutiny - Integration with RxJava 3</name>
+
+    <properties>
+        <revapi.skip>false</revapi.skip>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>io.smallrye.reactive</groupId>
+            <artifactId>mutiny</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>io.reactivex.rxjava3</groupId>
+            <artifactId>rxjava</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>io.smallrye.reactive</groupId>
+            <artifactId>mutiny-test-utils</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-api</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-engine</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-javadoc-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <id>attach-javadocs</id>
+                        <goals>
+                            <goal>jar</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+</project>

rxjava3/revapi.json

@@ -0,0 +1,43 @@
+[
+  {
+    "extension": "revapi.java",
+    "id": "java",
+    "configuration": {
+      "missing-classes": {
+        "behavior": "report",
+        "ignoreMissingAnnotations": false
+      },
+      "filter": {
+        "packages": {
+          "regex": true,
+          "include": [
+            "io\\.smallrye\\.mutiny\\.converters\\.*",  "io\\.smallrye\\.mutiny\\.converters\\..*"
+          ]
+        }
+      }
+    }
+  },
+
+  {
+    "extension": "revapi.differences",
+    "id": "breaking-changes",
+    "configuration": {
+      "criticality": "highlight",
+      "differences": [
+
+      ]
+    }
+  },
+
+  {
+    "extension": "revapi.reporter.json",
+    "configuration": {
+      "minSeverity": "POTENTIALLY_BREAKING",
+      "minCriticality": "documented",
+      "output": "target/compatibility.json",
+      "indent": true,
+      "append": false,
+      "keepEmptyFile": true
+    }
+  }
+]

rxjava3/src/main/java/io/smallrye/mutiny/converters/multi/FromCompletable.java

@@ -0,0 +1,18 @@
+package io.smallrye.mutiny.converters.multi;
+
+import io.reactivex.rxjava3.core.Completable;
+import io.smallrye.mutiny.Multi;
+import io.smallrye.mutiny.converters.MultiConverter;
+
+public class FromCompletable implements MultiConverter<Completable, Void> {
+    public static final FromCompletable INSTANCE = new FromCompletable();
+
+    private FromCompletable() {
+        // Avoid direct instantiation
+    }
+
+    @Override
+    public Multi<Void> from(Completable instance) {
+        return Multi.createFrom().publisher(instance.toFlowable());
+    }
+}

rxjava3/src/main/java/io/smallrye/mutiny/converters/multi/FromFlowable.java

@@ -0,0 +1,20 @@
+package io.smallrye.mutiny.converters.multi;
+
+import io.reactivex.rxjava3.core.Flowable;
+import io.smallrye.mutiny.Multi;
+import io.smallrye.mutiny.converters.MultiConverter;
+
+@SuppressWarnings("rawtypes")
+public class FromFlowable<T> implements MultiConverter<Flowable<T>, T> {
+
+    public static final FromFlowable INSTANCE = new FromFlowable();
+
+    private FromFlowable() {
+        // Avoid direct instantiation
+    }
+
+    @Override
+    public Multi<T> from(Flowable<T> instance) {
+        return Multi.createFrom().publisher(instance);
+    }
+}

rxjava3/src/main/java/io/smallrye/mutiny/converters/multi/FromMaybe.java

@@ -0,0 +1,34 @@
+package io.smallrye.mutiny.converters.multi;
+
+import io.reactivex.rxjava3.core.Maybe;
+import io.reactivex.rxjava3.disposables.Disposable;
+import io.smallrye.mutiny.Multi;
+import io.smallrye.mutiny.converters.MultiConverter;
+
+public class FromMaybe<T> implements MultiConverter<Maybe<T>, T> {
+
+    public static final FromMaybe INSTANCE = new FromMaybe();
+
+    private FromMaybe() {
+        // Avoid direct instantiation
+    }
+
+    @Override
+    public Multi<T> from(Maybe<T> instance) {
+        return Multi.createFrom().emitter(sink -> {
+            Disposable disposable = instance.subscribe(
+                    item -> {
+                        sink.emit(item);
+                        sink.complete();
+                    },
+                    sink::fail,
+                    sink::complete);
+
+            sink.onTermination(() -> {
+                if (!disposable.isDisposed()) {
+                    disposable.dispose();
+                }
+            });
+        });
+    }
+}

rxjava3/src/main/java/io/smallrye/mutiny/converters/multi/FromObservable.java

@@ -0,0 +1,21 @@
+package io.smallrye.mutiny.converters.multi;
+
+import io.reactivex.rxjava3.core.BackpressureStrategy;
+import io.reactivex.rxjava3.core.Observable;
+import io.smallrye.mutiny.Multi;
+import io.smallrye.mutiny.converters.MultiConverter;
+
+@SuppressWarnings("rawtypes")
+public class FromObservable<T> implements MultiConverter<Observable<T>, T> {
+
+    public static final FromObservable INSTANCE = new FromObservable();
+
+    private FromObservable() {
+        // Avoid direct instantiation
+    }
+
+    @Override
+    public Multi<T> from(Observable<T> instance) {
+        return Multi.createFrom().publisher(instance.toFlowable(BackpressureStrategy.BUFFER));
+    }
+}

rxjava3/src/main/java/io/smallrye/mutiny/converters/multi/FromSingle.java

@@ -0,0 +1,20 @@
+package io.smallrye.mutiny.converters.multi;
+
+import io.reactivex.rxjava3.core.Single;
+import io.smallrye.mutiny.Multi;
+import io.smallrye.mutiny.converters.MultiConverter;
+
+@SuppressWarnings("rawtypes")
+public class FromSingle<T> implements MultiConverter<Single<T>, T> {
+
+    public static final FromSingle INSTANCE = new FromSingle();
+
+    private FromSingle() {
+        // Avoid direct instantiation
+    }
+
+    @Override
+    public Multi<T> from(Single<T> instance) {
+        return Multi.createFrom().publisher(instance.toFlowable());
+    }
+}

rxjava3/src/main/java/io/smallrye/mutiny/converters/multi/MultiRx3Converters.java

@@ -0,0 +1,58 @@
+package io.smallrye.mutiny.converters.multi;
+
+public class MultiRx3Converters {
+
+    private MultiRx3Converters() {
+        // Avoid direct instantiation
+    }
+
+    public static FromCompletable fromCompletable() {
+        return FromCompletable.INSTANCE;
+    }
+
+    @SuppressWarnings("unchecked")
+    public static <T> FromSingle<T> fromSingle() {
+        return FromSingle.INSTANCE;
+    }
+
+    @SuppressWarnings("unchecked")
+    public static <T> FromMaybe<T> fromMaybe() {
+        return FromMaybe.INSTANCE;
+    }
+
+    @SuppressWarnings("unchecked")
+    public static <T> FromFlowable<T> fromFlowable() {
+        return FromFlowable.INSTANCE;
+    }
+
+    @SuppressWarnings("unchecked")
+    public static <T> FromObservable<T> fromObservable() {
+        return FromObservable.INSTANCE;
+    }
+
+    @SuppressWarnings("unchecked")
+    public static <T> ToSingle<T> toSingle() {
+        return ToSingle.INSTANCE;
+    }
+
+    @SuppressWarnings("unchecked")
+    public static <T> ToCompletable<T> toCompletable() {
+        return ToCompletable.INSTANCE;
+    }
+
+    @SuppressWarnings("unchecked")
+    public static <T> ToMaybe<T> toMaybe() {
+        return ToMaybe.INSTANCE;
+    }
+
+    @SuppressWarnings("unchecked")
+    public static <T> ToFlowable<T> toFlowable() {
+        return ToFlowable.INSTANCE;
+    }
+
+    @SuppressWarnings("unchecked")
+    public static <T> ToObservable<T> toObservable() {
+        return ToObservable.INSTANCE;
+    }
+
+}

rxjava3/src/main/java/io/smallrye/mutiny/converters/multi/ToCompletable.java

@@ -0,0 +1,21 @@
+package io.smallrye.mutiny.converters.multi;
+
+import java.util.function.Function;
+
+import io.reactivex.rxjava3.core.Completable;
+import io.smallrye.mutiny.Multi;
+
+@SuppressWarnings("rawtypes")
+public class ToCompletable<T> implements Function<Multi<T>, Completable> {
+
+    public static final ToCompletable INSTANCE = new ToCompletable();
+
+    private ToCompletable() {
+        // Avoid direct instantiation
+    }
+
+    @Override
+    public Completable apply(Multi<T> multi) {
+        return Completable.fromPublisher(multi.onItem().ignore());
+    }
+}