rxjava2-jdbc

RxJava2 does not support streams of null values. To handle nullable columns, use the following strategies:

1. Single nullable column: Use getAsOptional(Class<T> type) to return a java.util.Optional.
2. Multiple columns: Nulls are supported when mapping to Tuple instances.
3. Null parameters: Use Parameter.NULL or Database.NULL_CLOB/Database.NULL_BLOB for updates.

// Using getAsOptional for a single nullable column
Database.test() 
  .select("select date_of_birth from person where name='FRED'")
  .getAsOptional(Instant.class)
  .blockingForEach(System.out::println);

// Using explicit null parameter
Database.test()
  .update("update person set date_of_birth = ?") 
  .parameter(null)
  .counts()
  .blockingForEach(System.out::println);

To interact with a database, you need a Database instance. You can create one from a JDBC URL or use a built-in test database for development and testing.

• From URL: Use Database.from(url, maxPoolSize) to connect to an existing database.
• Test Database: Use Database.test(maxPoolSize) to create a fresh, in-memory database loaded with sample data (requires Apache Derby).

// Connect to an existing database
Database db = Database.from(url, maxPoolSize);

// Use the built-in test database
Database db = Database.test(maxPoolSize);

Add the following dependency to your pom.xml to use rxjava2-jdbc. Replace VERSION_HERE with the desired version.

If you want to use the built-in test database, you must also include the Apache Derby dependency.

<dependency>
  <groupId>com.github.davidmoten</groupId>
  <artifactId>rxjava2-jdbc</artifactId>
  <version>VERSION_HERE</version>
</dependency>

<!-- Required for the built-in test database -->
<dependency>
  <groupId>org.apache.derby</groupId>
  <artifactId>derby</artifactId>
  <version>10.14.2.0</version>
</dependency>

Use the select method to execute a SQL query. You can map the results to a specific class using getAs(Class<T> type).

Note: Queries run asynchronously. For demonstration purposes, blockingForEach can be used to wait for results, but in reactive applications, you should avoid blocking calls.

Database db = Database.test();
db.select("select name from person")
  .getAs(String.class)
  .blockingForEach(System.out::println);

By default, rxjava2-jdbc uses non-blocking connection pools. This means instead of blocking a thread when no connections are available, the query is queued and executed as soon as a connection is released. This is highly efficient for memory usage.

You can customize the pool behavior using the Database.nonBlocking() builder.

Database db = Database
  .nonBlocking()
  .url(url)
  .maxIdleTime(30, TimeUnit.MINUTES)
  .healthCheck(DatabaseType.ORACLE)
  .idleTimeBeforeHealthCheck(5, TimeUnit.SECONDS)
  .createRetryInterval(30, TimeUnit.SECONDS)
  .maxPoolSize(3)
  .build();

You can automatically map rows to an interface using the autoMap(Class<T> type) method.

Requirements

• The interface must be public.
• Use the @Column annotation to map method names to column names (case and underscores are ignored by default).
• Use the @Index(int) annotation to map by 1-based column position.
• You can annotate the interface with @Query to define the SQL statement directly.

Example

interface Person {
  @Column("name")
  String fullName();

  @Index(2)
  int examScore();
}

// Usage
db.select("select name, score from person")
  .autoMap(Person.class)
  .subscribe();

Blobs and Clobs are supported via specialized methods and types:

• CLOBs: Use Database.clob(value) for nullable strings, or pass a java.io.Reader. You can read them as String or Reader.
• BLOBs: Use Database.blob(bytes) for nullable byte arrays. You can read them as byte[] or InputStream.
• Nulls: Use Database.NULL_CLOB or Database.NULL_BLOB for explicit nulls.

// Insert a Clob
String document = "some text";
db.update("insert into person_clob(name,document) values(?,?)")
  .parameters("FRED", Database.clob(document))
  .count();

// Read a Blob
Flowable<byte[]> document = db.select("select document from person_blob")
  .getAs(byte[].class);

Transactions in rxjava2-jdbc wrap emissions in a Tx object. The commit or rollback happens automatically based on the terminal event (success or error).

• transacted(): Wraps each emission in a Tx object. Use this when you need to perform subsequent database actions within the same transaction.
• transactedValuesOnly(): Emits only the values, not the Tx wrapper, which is useful for simplifying downstream processing while still maintaining the transaction context.

Database.test()
  .select("select score from person where name=?") 
  .parameters("FRED", "JOSEPH") 
  .transacted() 
  .getAs(Integer.class) 
  .blockingForEach(tx -> 
    System.out.println(tx.isComplete() ? "complete" : tx.value()));

When you specify multiple types in the getAs method, the results are matched to the columns in the ResultSet and combined into a Tuple instance (e.g., Tuple2 through Tuple7, and TupleN for more columns).

Database db = Database.test();
db.select("select name, score from person")
  .getAs(String.class, Integer.class)
  .blockingForEach(System.out::println);

Callable statement support is available outside of transactions. You define in() parameters and out() parameters (which can be typed or result sets) to build the call.

• in(): Defines input parameters.
• out(Type, Class): Defines output parameters with strong typing.
• input(values...): Drives the execution with the provided input values.
• results1(), results2(), etc.: Access multiple ResultSet outputs from the call.

// Example of a call with typed output parameters
Flowable<Tuple2<Integer,Integer>> tuples = 
  db.call("call in1out2(?,?,?)") 
    .in() 
    .out(Type.INTEGER, Integer.class) 
    .out(Type.INTEGER, Integer.class) 
    .input(0, 10, 20);

If you need to perform operations not directly supported by rxjava2-jdbc, you can access the underlying java.sql.Connection.

• apply(Function<Connection, T> fn): Use this to run a block of code with a connection and return a value. The connection is managed by the library.
• member(): Provides a lower-level way to check out a connection. Warning: You are responsible for calling member.checkin() to return the connection to the pool.

// Using apply to get a value from a connection
Single<Integer> count = db.apply(con -> con.getHoldability());

// Using member for manual management
Completable completable = db.member() 
  .doOnSuccess(member -> {
     Connection con = member.value();
     try {
       // do work
     } finally {
       member.checkin(); // MUST call this
     }
  }).ignoreElements();