ldproxy · cportele · Aug 29, 2026
@@ -3,7 +3,6 @@ antlr = '4.13.2'
 xsom = '20140925'
 aalto = '1.4.0'
 staxmate = '2.4.2'
-rxjava-jdbc = '0.1.4-ii.1'
 hikaricp = '7.1.0'
 postgresql = '42.7.13'
 sqlite = '3.53.4.0'
@@ -28,7 +27,6 @@ aalto = { module = "com.fasterxml:aalto-xml", version.ref = "aalto" }
 staxmate = { module = "com.fasterxml.staxmate:staxmate", version.ref = "staxmate" }
 
 # sql
-rxjava-jdbc = { module = "com.github.davidmoten:rxjava3-jdbc", version.ref = "rxjava-jdbc" }
 hikaricp = { module = "com.zaxxer:HikariCP", version.ref = "hikaricp" }
 postgresql = { module = "org.postgresql:postgresql", version.ref = "postgresql" }
 sqlite = { module = "org.xerial:sqlite-jdbc", version.ref = "sqlite" }

@@ -26,7 +26,6 @@
 import java.util.Map;
 import java.util.Optional;
 import javax.sql.DataSource;
-import org.davidmoten.rxjava3.jdbc.pool.DatabaseType;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -251,11 +250,6 @@ public List<String> getDefaultSchemas() {
     return List.of();
   }
 
-  @Override
-  public DatabaseType getRxType() {
-    return DatabaseType.OTHER;
-  }
-
   @Override
   public List<String> getSystemSchemas() {
     return List.of("public");

@@ -24,7 +24,6 @@
 import java.util.Optional;
 import java.util.Set;
 import javax.sql.DataSource;
-import org.davidmoten.rxjava3.jdbc.pool.DatabaseType;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -128,11 +127,6 @@ public Optional<String> getInitSql(ConnectionInfoSql connectionInfo) {
     return Optional.empty();
   }
 
-  @Override
-  public DatabaseType getRxType() {
-    return DatabaseType.ORACLE;
-  }
-
   @Override
   public List<String> getDefaultSchemas() {
     return List.of("public");

@@ -23,13 +23,6 @@ dependencies {
     provided project(':xtraplatform-geometries')
     provided project(':xtraplatform-strings')
 
-    embedded(libs.rxjava.jdbc) {
-        exclude module: 'rxjava'
-        exclude module: 'reactive-streams'
-        exclude module: 'commons-io'
-        exclude module: 'slf4j-api'
-        exclude group: 'com.google.code.findbugs'
-    }
     //use reactive-streams + rxjava exported from this
     embeddedImport 'de.interactive_instruments:xtraplatform-streams'
 

@@ -16,7 +16,6 @@
 import java.util.Optional;
 import javax.annotation.Nullable;
 import javax.sql.DataSource;
-import org.davidmoten.rxjava3.jdbc.pool.DatabaseType;
 import org.immutables.value.Value;
 
 @AutoMultiBind
@@ -32,8 +31,6 @@ public interface SqlDbmsAdapter {
 
   List<String> getDefaultSchemas();
 
-  DatabaseType getRxType();
-
   List<String> getSystemSchemas();
 
   List<String> getSystemTables();

@@ -20,6 +20,7 @@
 import io.reactivex.rxjava3.core.Flowable;
 import io.reactivex.rxjava3.schedulers.Schedulers;
 import java.sql.Connection;
+import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.Statement;
 import java.text.Collator;
@@ -34,7 +35,6 @@
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import javax.sql.DataSource;
-import org.davidmoten.rxjava3.jdbc.Database;
 import org.postgresql.PGConnection;
 import org.postgresql.PGNotification;
 import org.slf4j.Logger;
@@ -50,21 +50,16 @@ public class SqlClientRx implements SqlClient {
   // it exists to end.
   private static final long READ_STALL_TIMEOUT_MINUTES = 10;
 
-  // rxjava3-jdbc is used for streamed reads only; everything that needs a connection of its own
-  // (sessions, statements without a result) leases it from the pool directly
-  private final Database session;
   private final DataSource dataSource;
   private final SqlDbmsAdapter dbmsAdapter;
   private final SqlDialect dialect;
   private final Collator collator;
 
   public SqlClientRx(
-      Database session,
       DataSource dataSource,
       SqlDbmsAdapter dbmsAdapter,
       SqlDialect dialect,
       Optional<String> defaultCollation) {
-    this.session = session;
     this.dataSource = dataSource;
     this.dbmsAdapter = dbmsAdapter;
     this.dialect = dialect;
@@ -92,11 +87,19 @@ public CompletableFuture<Collection<SqlRow>> run(String query, SqlQueryOptions o
       return result;
     }
 
-    session
-        .select(query)
-        .get(resultSet -> new SqlRowVals(collator).read(resultSet, options))
-        .toList()
-        .subscribe(result::complete, result::completeExceptionally);
+    try (Connection connection = dataSource.getConnection();
+        Statement statement = connection.createStatement();
+        ResultSet resultSet = statement.executeQuery(query)) {
+      List<SqlRow> rows = new ArrayList<>();
+
+      while (resultSet.next()) {
+        rows.add(new SqlRowVals(collator).read(resultSet, options));
+      }
+
+      result.complete(rows);
+    } catch (SQLException | RuntimeException e) {
+      result.completeExceptionally(e);
+    }
 
     return result;
   }
@@ -108,31 +111,35 @@ public Reactive.Source<SqlRow> getSourceStream(String query, SqlQueryOptions opt
     }
     List<SqlRow> logBuffer = new ArrayList<>(5);
 
-    org.davidmoten.rxjava3.jdbc.ResultSetMapper<SqlRow> mapper =
-        resultSet -> {
-          SqlRow row = new SqlRowVals(collator).read(resultSet, options);
-
-          if (LOGGER.isDebugEnabled(MARKER.SQL_RESULT) && logBuffer.size() < 10) {
-            logBuffer.add(row);
-          }
-
-          return row;
-        };
-
     // A positive fetch size requires a transaction so the database driver uses a server-side cursor
     // and streams rows instead of buffering the whole result set in memory (PostgreSQL ignores the
     // fetch size with autoCommit=true).
-    // TODO encapsulating the query in a transaction is also a workaround for what appears to be a
-    //      bug in rxjava3-jdbc, see https://github.com/interactive-instruments/ldproxy/issues/1293
+    boolean streamed = options.getFetchSize() > 0;
+
+    // The connection is leased when the stream is subscribed and returned to the pool when it
+    // terminates, whether the rows were exhausted, the read failed or the consumer cancelled.
     Flowable<SqlRow> flowable =
-        options.getFetchSize() > 0
-            ? session
-                .select(query)
-                .transacted()
-                .fetchSize(options.getFetchSize())
-                .valuesOnly()
-                .get(mapper)
-            : session.select(query).get(mapper);
+        Flowable.using(
+            () -> lease(streamed),
+            connection ->
+                Flowable.generate(
+                    () -> execute(connection, query, options.getFetchSize()),
+                    (resultSet, emitter) -> {
+                      if (resultSet.next()) {
+                        SqlRow row = new SqlRowVals(collator).read(resultSet, options);
+
+                        if (LOGGER.isDebugEnabled(MARKER.SQL_RESULT) && logBuffer.size() < 10) {
+                          logBuffer.add(row);
+                        }
+
+                        emitter.onNext(row);
+                      } else {
+                        emitter.onComplete();
+                      }
+                    },
+                    SqlClientRx::close),
+            connection -> release(connection, streamed),
+            true);
 
     // TODO: prettify, see
     // https://github.com/slick/slick/blob/main/slick/src/main/scala/slick/jdbc/StatementInvoker.scala
@@ -169,20 +176,17 @@ public Reactive.Source<SqlRow> getSourceStream(String query, SqlQueryOptions opt
               });
     }
 
-    // The blocking connection provider runs connect+execute+read on the subscribing thread, so
-    // without this the whole stream is single-threaded. Subscribing on a worker thread lets several
-    // parallel-flagged queries (e.g. the concurrent single-shot value phase) run at once, each on
-    // its
-    // own connection.
+    // Lease, execute and read run on the subscribing thread, so without this the whole stream is
+    // single-threaded. Subscribing on a worker thread lets several parallel-flagged queries (e.g.
+    // the concurrent single-shot value phase) run at once, each on its own connection.
     if (options.isParallel()) {
       flowable = flowable.subscribeOn(Schedulers.io());
     }
 
-    // Safety net for a read that neither completes nor fails. A database error raised while the
-    // rows are being streamed is not delivered by the underlying library (see the issue linked
-    // above), so the stream can stall forever: no error is logged, no response is sent, and the
-    // connections the sub-query holds stay held until the client gives up. A connection lost
-    // mid-stream — a failover in a replicated cluster, for instance — looks exactly the same.
+    // Safety net for a read that neither completes nor fails: a connection lost mid-stream — a
+    // failover in a replicated cluster, for instance — can leave the driver waiting for the next
+    // row forever, so no error is logged, no response is sent, and the connections the sub-query
+    // holds stay held until the client gives up.
     // The timeout is per element, not per stream, so a slow but progressing read is unaffected
     // however long it runs in total; only a gap longer than the window ends the stream, with an
     // error that does propagate. A database-side statement_timeout is no substitute: its error
@@ -203,6 +207,80 @@ public Reactive.Source<SqlRow> getSourceStream(String query, SqlQueryOptions opt
     return Reactive.Source.publisher(flowable);
   }
 
+  private Connection lease(boolean transaction) throws SQLException {
+    Connection connection = dataSource.getConnection();
+
+    if (transaction) {
+      try {
+        connection.setAutoCommit(false);
+      } catch (SQLException e) {
+        close(connection);
+        throw e;
+      }
+    }
+
+    return connection;
+  }
+
+  private static ResultSet execute(Connection connection, String query, int fetchSize)
+      throws SQLException {
+    Statement statement = connection.createStatement();
+
+    try {
+      if (fetchSize > 0) {
+        statement.setFetchSize(fetchSize);
+      }
+
+      return statement.executeQuery(query);
+    } catch (SQLException | RuntimeException e) {
+      close(statement);
+      throw e;
+    }
+  }
+
+  /** Ends a read-only transaction, if any, and returns the connection to the pool. */
+  private static void release(Connection connection, boolean transaction) {
+    if (transaction) {
+      try {
+        // nothing to commit, and a rollback is the cheapest way to close the server-side cursor
+        connection.rollback();
+      } catch (SQLException e) {
+        LOGGER.debug("Ending the read transaction failed: {}", e.getMessage());
+      }
+      try {
+        connection.setAutoCommit(true);
+      } catch (SQLException e) {
+        LOGGER.debug("Resetting autocommit failed: {}", e.getMessage());
+      }
+    }
+
+    close(connection);
+  }
+
+  /** Closes the result set and the statement it belongs to. */
+  private static void close(ResultSet resultSet) {
+    Statement statement = null;
+
+    try {
+      statement = resultSet.getStatement();
+    } catch (SQLException e) {
+      // the result set is closed below regardless
+    }
+
+    close((AutoCloseable) resultSet);
+    close(statement);
+  }
+
+  private static void close(AutoCloseable closeable) {
+    if (Objects.nonNull(closeable)) {
+      try {
+        closeable.close();
+      } catch (Exception e) {
+        LOGGER.debug("Closing {} failed: {}", closeable.getClass().getSimpleName(), e.getMessage());
+      }
+    }
+  }
+
   @Override
   public Connection getConnection() {
     return leaseConnection();

@@ -51,7 +51,6 @@
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
-import org.davidmoten.rxjava3.jdbc.Database;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.slf4j.MDC;
@@ -80,7 +79,6 @@ public class SqlConnectorRx extends AbstractVolatilePolling implements SqlConnec
   private final AtomicInteger refCounter;
   private final boolean asyncStartup;
 
-  private Database session;
   private HikariDataSource dataSource;
   private SqlClient sqlClient;
   private Throwable connectionError;
@@ -166,10 +164,8 @@ public void start() {
     try {
       HikariConfig hikariConfig = createHikariConfig();
       this.dataSource = new HikariDataSource(hikariConfig);
-      this.session = createSession(dataSource);
       this.sqlClient =
           new SqlClientRx(
-              session,
               dataSource,
               dbmsAdapters.get(connectionInfo.getDialect()),
               dbmsAdapters.getDialect(connectionInfo.getDialect()),
@@ -192,13 +188,6 @@ public void stop() {
         // ignore
       }
     }
-    if (Objects.nonNull(session)) {
-      try {
-        session.close();
-      } catch (Throwable e) {
-        // ignore
-      }
-    }
     if (Objects.nonNull(dataSource)) {
       try {
         dataSource.close();
@@ -360,10 +349,6 @@ private HikariConfig createHikariConfig() {
     return config;
   }
 
-  private Database createSession(HikariDataSource dataSource) {
-    return Database.fromBlocking(dataSource);
-  }
-
   private static long getInitFailTimeout(ConnectionInfoSql connectionInfo) {
     return parseMs(Objects.requireNonNullElse(connectionInfo.getPool().getInitFailTimeout(), "1"));
   }

@@ -38,7 +38,6 @@
 import java.util.Objects;
 import java.util.Optional;
 import javax.sql.DataSource;
-import org.davidmoten.rxjava3.jdbc.pool.DatabaseType;
 import org.immutables.value.Value;
 import org.sqlite.SQLiteConnection;
 import org.sqlite.SQLiteDataSource;
@@ -140,11 +139,6 @@ public Optional<String> getInitSql(ConnectionInfoSql connectionInfo) {
     return Optional.of("SELECT CASE CheckGeoPackageMetaData() WHEN 1 THEN EnableGpkgMode() END;");
   }
 
-  @Override
-  public DatabaseType getRxType() {
-    return DatabaseType.SQLITE;
-  }
-
   @Override
   public List<String> getDefaultSchemas() {
     return List.of();