Update to RxJava 2

pom.xml

@@ -75,9 +75,9 @@
             <version>4.1.11.Final</version>
         </dependency>
         <dependency>
-            <groupId>io.reactivex</groupId>
+            <groupId>io.reactivex.rxjava2</groupId>
             <artifactId>rxjava</artifactId>
-            <version>1.3.0</version>
+            <version>2.1.2</version>
         </dependency>
         <dependency>
             <groupId>com.google.code.findbugs</groupId>

src/main/java/com/github/pgasync/ConnectionPool.java

@@ -14,7 +14,7 @@
 
 package com.github.pgasync;
 
-import rx.Observable;
+import io.reactivex.Single;
 
 /**
  * Pool of backend {@link Connection}s. Pools implement {@link Db} so
@@ -30,7 +30,7 @@ public interface ConnectionPool extends Db {
      * available. Connection passed to callback must be freed with
      * {@link com.github.pgasync.ConnectionPool#release(Connection)}
      */
-    Observable<Connection> getConnection();
+    Single<Connection> getConnection();
 
     /**
      * Releases a connection back to the pool.

src/main/java/com/github/pgasync/ConnectionPoolBuilder.java

@@ -17,8 +17,8 @@
 import com.github.pgasync.impl.ConnectionValidator;
 import com.github.pgasync.impl.conversion.DataConverter;
 import com.github.pgasync.impl.netty.NettyPgConnectionPool;
-import rx.Observable;
-import rx.functions.Func1;
+import io.reactivex.Single;
+import io.reactivex.functions.Function;
 
 import java.util.ArrayList;
 import java.util.Collections;
@@ -139,9 +139,9 @@ public boolean getUsePipelining() {
         public DataConverter getDataConverter() {
             return dataConverter != null ? dataConverter : new DataConverter(converters);
         }
-        public Func1<Connection,Observable<Connection>> getValidator() {
+        public Function<Connection, Single<Connection>> getValidator() {
             return validationQuery == null || validationQuery.trim().isEmpty()
-                ? Observable::just
+                ? Single::just
                 : new ConnectionValidator(validationQuery)::validate;
         }
     }

src/main/java/com/github/pgasync/Listenable.java

@@ -1,6 +1,6 @@
 package com.github.pgasync;
 
-import rx.Observable;
+import io.reactivex.Observable;
 
 /**
  * @author Antti Laisi

src/main/java/com/github/pgasync/QueryExecutor.java

@@ -1,6 +1,7 @@
 package com.github.pgasync;
 
-import rx.Observable;
+import io.reactivex.Observable;
+import io.reactivex.Single;
 
 import java.util.List;
 import java.util.function.Consumer;
@@ -15,7 +16,7 @@ public interface QueryExecutor {
     /**
      * Begins a transaction.
      */
-    Observable<Transaction> begin();
+    Single<Transaction> begin();
 
     /**
      * Executes an anonymous prepared statement. Uses native PostgreSQL syntax with $arg instead of ?
@@ -37,7 +38,7 @@ public interface QueryExecutor {
      * @param params Parameter values
      * @return Cold observable that emits a single result set.
      */
-    Observable<ResultSet> querySet(String sql, Object... params);
+    Single<ResultSet> querySet(String sql, Object... params);
 
     /**
      * Begins a transaction.
@@ -70,7 +71,7 @@ default void query(String sql, Consumer<ResultSet> onResult, Consumer<Throwable>
      * @param onResult Called when query is completed
      * @param onError Called on exception thrown
      */
-    default void query(String sql, List/*<Object>*/ params, Consumer<ResultSet> onResult, Consumer<Throwable> onError) {
+    default void query(String sql, List<?> params, Consumer<ResultSet> onResult, Consumer<Throwable> onError) {
         querySet(sql, params != null ? params.toArray() : null).subscribe(onResult::accept, onError::accept);
     }
 }

src/main/java/com/github/pgasync/Transaction.java

@@ -14,7 +14,7 @@
 
 package com.github.pgasync;
 
-import rx.Observable;
+import io.reactivex.Completable;
 
 import java.util.function.Consumer;
 
@@ -30,12 +30,12 @@ public interface Transaction extends QueryExecutor {
     /**
      * Commits a transaction
      */
-    Observable<Void> commit();
+    Completable commit();
 
     /**
      * Rollbacks a transaction.
      */
-    Observable<Void> rollback();
+    Completable rollback();
 
     /**
      * Commits a transaction.
@@ -44,7 +44,7 @@ public interface Transaction extends QueryExecutor {
      * @param onError Called on exception thrown
      */
     default void commit(Runnable onCompleted, Consumer<Throwable> onError) {
-        commit().subscribe(__ -> onCompleted.run(), onError::accept);
+        commit().subscribe(onCompleted::run, onError::accept);
     }
 
     /**
@@ -54,7 +54,7 @@ default void commit(Runnable onCompleted, Consumer<Throwable> onError) {
      * @param onError Called on exception thrown
      */
     default void rollback(Runnable onCompleted, Consumer<Throwable> onError) {
-        rollback().subscribe(__ -> onCompleted.run(), onError::accept);
+        rollback().subscribe(onCompleted::run, onError::accept);
     }
 
 }

src/main/java/com/github/pgasync/impl/ConnectionValidator.java

@@ -16,9 +16,7 @@
 
 
 import com.github.pgasync.Connection;
-import com.github.pgasync.Row;
-import rx.Observable;
-import rx.Subscriber;
+import io.reactivex.Single;
 
 /**
  * @author Antti Laisi
@@ -31,21 +29,10 @@ public ConnectionValidator(String validationQuery) {
         this.validationQuery = validationQuery;
     }
 
-    public Observable<Connection> validate(Connection connection) {
+    public Single<Connection> validate(Connection connection) {
         return connection.queryRows(validationQuery)
-                .lift(subscriber -> new Subscriber<Row>() {
-                    @Override
-                    public void onError(Throwable e) {
-                        subscriber.onError(e);
-                    }
-                    @Override
-                    public void onCompleted() {
-                        subscriber.onNext(connection);
-                        subscriber.onCompleted();
-                    }
-                    @Override
-                    public void onNext(Row row) { }
-                });
+                .ignoreElements()
+                .toSingleDefault(connection);
     }
 
 }