Upgrade to RfJava2 for testing reactive streams, close #1380

Motivation:

ReactiveStreamsTest#testConnectionDoesNotGetClosed randomly fails.
The server sometimes receives chunks out of order.

Modifications:

The issue seems to be a RxJava1 one.
Upgrade to RxJava2.

Result:

No more erroneous random test failures

Author: slandelle

client/pom.xml

@@ -50,12 +50,15 @@
 		<dependency>
 			<groupId>org.reactivestreams</groupId>
 			<artifactId>reactive-streams</artifactId>
-			<version>1.0.0</version>
 		</dependency>
 		<dependency>
 			<groupId>com.typesafe.netty</groupId>
 			<artifactId>netty-reactive-streams</artifactId>
-			<version>2.0.0</version>
+		</dependency>
+		<dependency>
+			<groupId>io.reactivex.rxjava2</groupId>
+			<artifactId>rxjava</artifactId>
+			<scope>test</scope>
 		</dependency>
 	</dependencies>
 </project>

client/src/test/java/org/asynchttpclient/reactivestreams/ReactiveStreamsTest.java

@@ -19,6 +19,7 @@
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 import io.netty.handler.codec.http.HttpHeaders;
+import io.reactivex.Flowable;
 
 import java.io.ByteArrayOutputStream;
 import java.io.File;
@@ -60,16 +61,12 @@
 import org.testng.annotations.BeforeClass;
 import org.testng.annotations.Test;
 
-import rx.Observable;
-import rx.RxReactiveStreams;
-
 public class ReactiveStreamsTest {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(ReactiveStreamsTest.class);
 
     public static Publisher<ByteBuf> createPublisher(final byte[] bytes, final int chunkSize) {
-        Observable<ByteBuf> observable = Observable.from(new ByteBufIterable(bytes, chunkSize));
-        return RxReactiveStreams.toPublisher(observable);
+        return Flowable.fromIterable(new ByteBufIterable(bytes, chunkSize));
     }
 
     private Tomcat tomcat;
@@ -236,11 +233,7 @@ public void testConnectionDoesNotGetClosed() throws Exception {
 
             byte[] responseBody = response.getResponseBodyAsBytes();
             responseBody = response.getResponseBodyAsBytes();
-            assertEquals(
-                Integer.valueOf(response.getHeader("X-" + CONTENT_LENGTH)).intValue(),
-                LARGE_IMAGE_BYTES.length,
-                "Server side payload length invalid"
-            );
+            assertEquals(Integer.valueOf(response.getHeader("X-" + CONTENT_LENGTH)).intValue(), LARGE_IMAGE_BYTES.length, "Server side payload length invalid");
             assertEquals(responseBody.length, LARGE_IMAGE_BYTES.length, "Client side payload length invalid");
             assertEquals(response.getHeader(CONTENT_MD5), expectedMd5, "Server side payload MD5 invalid");
             assertEquals(TestUtils.md5(responseBody), expectedMd5, "Client side payload MD5 invalid");
@@ -249,11 +242,7 @@ public void testConnectionDoesNotGetClosed() throws Exception {
             response = requestBuilder.execute().get();
             assertEquals(response.getStatusCode(), 200);
             responseBody = response.getResponseBodyAsBytes();
-            assertEquals(
-                Integer.valueOf(response.getHeader("X-" + CONTENT_LENGTH)).intValue(),
-                LARGE_IMAGE_BYTES.length,
-                "Server side payload length invalid"
-            );
+            assertEquals(Integer.valueOf(response.getHeader("X-" + CONTENT_LENGTH)).intValue(), LARGE_IMAGE_BYTES.length, "Server side payload length invalid");
             assertEquals(responseBody.length, LARGE_IMAGE_BYTES.length, "Client side payload length invalid");
 
             try {
@@ -285,9 +274,7 @@ public static void main(String[] args) throws Exception {
     @Test(groups = "standalone", expectedExceptions = ExecutionException.class)
     public void testFailingStream() throws Exception {
         try (AsyncHttpClient client = asyncHttpClient(config().setRequestTimeout(100 * 6000))) {
-            Observable<ByteBuf> failingObservable = Observable.error(new FailedStream());
-            Publisher<ByteBuf> failingPublisher = RxReactiveStreams.toPublisher(failingObservable);
-
+            Publisher<ByteBuf> failingPublisher = Flowable.error(new FailedStream());
             client.preparePut(getTargetUrl()).setBody(failingPublisher).execute().get();
         }
     }
@@ -520,7 +507,7 @@ public ByteBufIterable(byte[] payload, int chunkSize) {
         @Override
         public Iterator<ByteBuf> iterator() {
             return new Iterator<ByteBuf>() {
-                private volatile int currentIndex = 0;
+                private int currentIndex = 0;
 
                 @Override
                 public boolean hasNext() {

extras/rxjava/pom.xml

@@ -12,7 +12,6 @@
         <dependency>
             <groupId>io.reactivex</groupId>
             <artifactId>rxjava</artifactId>
-            <version>1.2.9</version>
         </dependency>
     </dependencies>
 </project>

extras/rxjava2/pom.xml

@@ -12,7 +12,6 @@
         <dependency>
             <groupId>io.reactivex.rxjava2</groupId>
             <artifactId>rxjava</artifactId>
-            <version>2.0.8</version>
         </dependency>
     </dependencies>
 </project>

pom.xml

@@ -257,11 +257,26 @@
 				<version>${netty.version}</version>
 				<optional>true</optional>
 			</dependency>
+			<dependency>
+				<groupId>org.reactivestreams</groupId>
+				<artifactId>reactive-streams</artifactId>
+				<version>${reactive-streams.version}</version>
+			</dependency>
+			<dependency>
+				<groupId>com.typesafe.netty</groupId>
+				<artifactId>netty-reactive-streams</artifactId>
+				<version>${netty-reactive-streams.version}</version>
+			</dependency>
 			<dependency>
 				<groupId>io.reactivex</groupId>
 				<artifactId>rxjava</artifactId>
 				<version>${rxjava.version}</version>
 			</dependency>
+			<dependency>
+				<groupId>io.reactivex.rxjava2</groupId>
+				<artifactId>rxjava</artifactId>
+				<version>${rxjava2.version}</version>
+			</dependency>
 		</dependencies>
 	</dependencyManagement>
 	<dependencies>
@@ -349,12 +364,6 @@
 			<version>${privilegedaccessor.version}</version>
 			<scope>test</scope>
 		</dependency>
-		<dependency>
-			<groupId>io.reactivex</groupId>
-			<artifactId>rxjava-reactive-streams</artifactId>
-			<version>${rxjava-reactive-streams.version}</version>
-			<scope>test</scope>
-		</dependency>
 		<dependency>
 			<groupId>org.powermock</groupId>
 			<artifactId>powermock-module-testng</artifactId>
@@ -375,15 +384,17 @@
 		<target.property>1.8</target.property>
 		<netty.version>4.1.13.Final</netty.version>
 		<slf4j.version>1.7.25</slf4j.version>
+		<reactive-streams.version>1.0.0</reactive-streams.version>
+		<netty-reactive-streams.version>2.0.0</netty-reactive-streams.version>
+		<rxjava.version>1.3.0</rxjava.version>
+		<rxjava2.version>2.1.2</rxjava2.version>
 		<logback.version>1.2.3</logback.version>
 		<testng.version>6.9.10</testng.version>
 		<jetty.version>9.4.6.v20170531</jetty.version>
 		<tomcat.version>8.5.14</tomcat.version>
 		<commons-io.version>2.4</commons-io.version>
 		<commons-fileupload.version>1.3</commons-fileupload.version>
 		<privilegedaccessor.version>1.2.2</privilegedaccessor.version>
-		<rxjava-reactive-streams.version>1.2.1</rxjava-reactive-streams.version>
-		<rxjava.version>1.3.0</rxjava.version>
 		<powermock.version>1.6.4</powermock.version>
 	</properties>
 </project>