JDBC is so much simpler with rxjava2-jdbc.

For Rxjava 3.x see rxjava3-jdbc.

Status: published to Maven Central

Maven site reports are here including javadoc.

• How to build
• Getting started
• Database
• Support for playing with rxjava2-jdbc!
• A query example
• Asynchrony
• Nulls
  • Null parameters
• Tuple support
• Manual mapping
• Automap
  • Automapped toString
  • Automapped equals/hashCode
  • Automapped interface with default methods
  • Automap with annotated query
  • Automap with Kotlin
  • Auto mappings
• Parameters
  • Explicit anonymous parameters
  • Flowable anonymous parameters
  • Mixing explicit and Flowable parameters
  • Multiple parameters per query
  • Running a query many times that has no parameters
  • Collection parameters
• Non-blocking connection pools
  • Demonstration
• Large objects support
  • Insert a Clob
  • Insert a Null Clob
  • Read a Clob
  • Read a Null Clob
  • Insert a Blob
  • Insert a Null Blob
  • Read a Blob
• Returning generated keys
• Transactions
• Callable Statements
• Using raw JDBC
• Logging

Use maven:

The project builds fine on Java 8 to 17.

Use this maven dependency:

If you want to use the built-in test database then add the Apache Derby dependency (otherwise you’ll need the jdbc dependency for the database you want to connect to):

To start things off you need a Database instance. Given the jdbc url of your database you can create a Database object like this:

If you want to have a play with a built-in test database then do this:

To use the built-in test database you will need the Apache Derby dependency:

The test database has a few tables (see script) including Person and Address with three rows in Person and two rows in Address:

Each time you call Database.test(maxPoolSize) you will have a fresh new database to play with that is loaded with data as described above.

Let’s use a Database instance to perform a select query on the Person table and write the names to the console:

Output is

Note the use of blockingForEach. This is only for demonstration purposes. When a query is executed it is executed asynchronously on a scheduler that you can specify if desired. The default scheduler is:

While you are processing reactively you should avoid blocking calls but domain boundaries sometimes force this to happen (e.g. accumulate the results and return them as xml over the network from a web service call). Bear in mind also that if you are worried about the complexities of debugging RxJava programs then you might wish to make brief limited forays into reactive code. That’s completely fine too. What you lose in efficiency you may gain in simplicity.

The query flowables returned by the Database all run asynchronously. This is required because of the use of non-blocking connection pools. When a connection is returned to the pool and then checked-out by another query that checkout must occur on a different thread so that stack overflow does not occur. See the Non-blocking connection pools section for more details.

RxJava2 does not support streams of nulls. If you want to represent nulls in your stream then use java.util.Optional.

In the special case where a single nullable column is being returned and mapped to a class via getAs you should instead use getAsOptional:

Output:

Nulls will happily map to Tuples (see the next section) when you have two or more columns.

Null parameters

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

Database.test()
  .update("update person set date_of_birth = :dob")
  .parameter(Parameter.create("dob", null))
  .counts()
  .blockingForEach(System.out::println);

Database.test()
  .update("update person set date_of_birth = ?")
  .parameterStream(Flowable.just(Parameter.NULL))
  .counts()
  .blockingForEach(System.out::println);

When you specify more types in the getAs method they are matched to the columns in the returned result set from the query and combined into a Tuple instance. Here’s an example that returns Tuple2:

Output

Tuples are defined from Tuple2 to Tuple7 and for above that to TupleN.

You can map each row of the JDBC ResultSet to your own object using the get method:

By the way it is definitely not a good idea to hang on to a bunch of rs objects as their state will always be that of the latest read row. For this reason you immediately map it to something else either by manually mapping or automapping.

To map the result set values to an interface, first declare an interface:

In the query use the autoMap method and let’s use some of the built-in testing methods of RxJava2 to confirm we got what we expected:

If your interface method name does not exactly match the column name (underscores and case are ignored) then you can add more detail to the Column annotation:

You can also refer to the 1-based position of the column in the result set instead of its name:

In fact, you can mix use of named columns and indexed columns in automapped interfaces.

If you don’t configure things correctly these exceptions may be emitted and include extra information in the error message about the affected automap interface:

• AnnotationsNotFoundException

• ColumnIndexOutOfRangeException

• ColumnNotFoundException

• ClassCastException

• AutomappedInterfaceInaccessibleException

Automapped toString

Person[name=FRED, score=21]

Automapped equals/hashCode

Automapped interface with default methods

Automap with annotated query

@Query("select name, score from person order by name")
interface Person {
   @Column
   String name();

   @Column
   int score();
}

Database
  .test()
  .select(Person.class)
  .get()
  .map(Person::name)
  .blockingForEach(System.out::println);

FRED
JOSEPH
MARMADUKE

Database
  .test()
  .select(Person.class)
  .get(Person::name)
  .blockingForEach(System.out::println);

Automap with Kotlin

@Query("select name from person order by name")
interface Person {
    @Column("name")
    fun nm() : String
}

Auto mappings

Parameters are passed to individual queries but can also be used as a streaming source to prompt the query to be run many times.

Parameters can be named or anonymous. Named parameters are not supported natively by the JDBC specification but rxjava2-jdbc does support them.

This is sql with a named parameter:

This is sql with an anonymous parameter:

Explicit anonymous parameters

Database.test()
  .select("select score from person where name=?")
  .parameters("FRED", "JOSEPH")
  .getAs(Integer.class)
  .blockingForEach(System.out::println);

21
34

Flowable anonymous parameters

Database.test()
  .select("select score from person where name=?")
  .parameterStream(Flowable.just("FRED","JOSEPH").repeat())
  .getAs(Integer.class)
  .take(3)
  .blockingForEach(System.out::println);

21
34
21

Mixing explicit and Flowable parameters

Database.test()
  .select("select score from person where name=?")
  .parameterStream(Flowable.just("FRED","JOSEPH"))
  .parameters("FRED", "JOSEPH")
  .getAs(Integer.class)
  .blockingForEach(System.out::println);

21
34
21
34

Multiple parameters per query

Database.test()
  .select("select score from person where name=? and score=?")
  .parameterStream(Flowable.just("FRED", 21, "JOSEPH", 34).repeat())
  .getAs(Integer.class)
  .take(3)
  .blockingForEach(System.out::println);

Database.test()
  .select("select score from person where name=? and score=?")
  .parameterListStream(Flowable.just(Arrays.asList("FRED", 21), Arrays.asList("JOSEPH", 34)).repeat())
  .getAs(Integer.class)
  .take(3)
  .blockingForEach(System.out::println);

Running a query many times that has no parameters

Database.test()
  .select("select count(*) from person")
  .parameters("a", "b", "c")
  .getAs(Integer.class)
  .blockingForEach(System.out::println);

3
3
3

Collection parameters

Database.test()
  .select("select score from person where name in (?) order by score")
  .parameter(Sets.newHashSet("FRED", "JOSEPH"))
  .getAs(Integer.class)
  .blockingForEach(System.out::println);

Database.test()
  .update("update person set score=0 where name in (:names)")
  .parameter("names", Lists.newArrayList("FRED", "JOSEPH"))
  .counts()
  .blockingForEach(System.out::println);

A new exciting feature of rxjava2-jdbc is the availability of non-blocking connection pools.

In normal non-reactive database programming a couple of different threads (started by servlet calls for instance) will race for the next available connection from a pool of database connections. If no unused connection remains in the pool then the standard non-reactive approach is to block the thread until a connection becomes available.

Blocking a thread is a resource issue as each blocked thread holds onto ~0.5MB of stack and may incur context switch and memory-access delays (adds latency to thread processing) when being switched to. For example 100 blocked threads hold onto ~50MB of memory (outside of java heap).

rxjava-jdbc2 uses non-blocking JDBC connection pools by default (but is configurable to use whatever you want). What happens in practice is that for each query a subscription is made to a MemberSingle instance controlled by the NonBlockingConnectionPool object that emits connections when available to its subscribers (first in best dressed). So the definition of the processing of that query is stored on a queue to be started when a connection is available. Adding the Flowable definition of your query to the queue can be quite efficient in terms of memory use compared to the memory costs of thread per query. For example a heap dump of 1000 queued simple select statements from the person table in the test database used 429K of heap. That is 429 bytes per query.

The simplest way of creating a Database instance with a non-blocking connection pool is:

If you want to play with the in-memory built-in test database (requires Apache Derby dependency) then:

If you want more control over the behaviour of the non-blocking connection pool:

Note that the health check varies from database to database. The following databases are directly supported with DatabaseType instances:

• DB2

• Derby

• HSQLDB

• H2

• Informix

• MySQL

• Oracle

• Postgres

• Microsoft SQL Server

• SQLite

Demonstration

// create database with non-blocking connection pool
// of size 1
Database db = Database.test(1);

// start a slow query
db.select("select score from person where name=?")
  .parameter("FRED")
  .getAs(Integer.class)
   // slow things down by sleeping
  .doOnNext(x -> Thread.sleep(1000))
   // run in background thread
  .subscribeOn(Schedulers.io())
  .subscribe();

// ensure that query starts
Thread.sleep(100);

// query again while first query running
db.select("select score from person where name=?")
  .parameter("FRED")
  .getAs(Integer.class)
  .doOnNext(x -> System.out.println("emitted on " + Thread.currentThread().getName()))
  .subscribe();

System.out.println("second query submitted");

// wait for stuff to happen asynchronously
Thread.sleep(5000);

second query submitted
emitted on RxCachedThreadScheduler-1

Blobs and Clobs are straightforward to handle.

Insert a Clob

String document = ...
Flowable<Integer> count = db
  .update("insert into person_clob(name,document) values(?,?)")
  .parameters("FRED", document)
  .count();

String document = ...
Flowable<Integer> count = db
  .update("insert into person_clob(name,document) values(?,?)")
  .parameters("FRED", Database.clob(document))
  .count();

Reader reader = ...;
Flowable<Integer> count = db
  .update("insert into person_clob(name,document) values(?,?)")
  .parameters("FRED", reader)
  .count();

Insert a Null Clob

Flowable<Integer> count = db
  .update("insert into person_clob(name,document) values(?,?)")
  .parameters("FRED", Database.NULL_CLOB)
  .count();

Flowable<Integer> count = db
  .update("insert into person_clob(name,document) values(?,?)")
  .parameters("FRED", Database.clob(null))
  .count();

Read a Clob

Flowable<String> document =
  db.select("select document from person_clob")
    .getAs(String.class);

Flowable<Reader> document =
  db.select("select document from person_clob")
    .getAs(Reader.class);

Read a Null Clob

db.select("select document from person_clob where name='FRED'")
  .getAsOptional(String.class)

Insert a Blob

byte[] bytes = ...
Flowable<Integer> count = db
  .update("insert into person_blob(name,document) values(?,?)")
  .parameters("FRED", Database.blob(bytes))
  .count();

Insert a Null Blob

Flowable<Integer> count = db
  .update("insert into person_blob(name,document) values(?,?)")
  .parameters("FRED", Database.NULL_BLOB)
  .count();

Flowable<Integer> count = db
  .update("insert into person_clob(name,document) values(?,?)")
  .parameters("FRED", Database.blob(null))
  .count();

Read a Blob

Flowable<byte[]> document =
  db.select("select document from person_clob")
    .getAs(byte[].class);

Flowable<InputStream> document =
  db.select("select document from person_clob")
    .getAs(InputStream.class);

If you insert into a table that say in h2 is of type auto_increment then you don’t need to specify a value but you may want to know what value was inserted in the generated key field.

Given a table like this

This code inserts two rows into the note table and returns the two generated keys:

The returnGeneratedKeys method also supports returning multiple keys per row so the builder offers methods just like select to do explicit mapping or auto mapping.

Transactions are a critical feature of relational databases.

When we’re talking RxJava we need to consider the behaviour of individual JDBC objects when called by different threads, possibly concurrently. The approach taken by rxjava2-jdbc outside of a transaction safely uses Connection pools (in a non-blocking way). Inside a transaction we must make all calls to the database using the same Connection object so the behaviour of that Connection when called from different threads is important. Some JDBC drivers provide thread-safety on JDBC objects by synchronizing every call.

The safest approach with transactions is to perform all db interaction synchronously. Asynchronous processing within transactions was problematic in rxjava-jdbc because ThreadLocal was used to hold the Connection. Asynchronous processing with transactions is possible with rxjava2-jdbc but should be handled with care given that your JDBC driver may block or indeed suffer from race conditions that most users don’t encounter.

Let’s look at some examples. The first example uses a transaction across two select statement calls:

Output:

Note that the commit/rollback of the transaction happens automatically.

What we see above is that each emission from the select statement is wrapped with a Tx object including the terminal event (error or complete). This is so you can for instance perform an action using the same transaction.

Let’s see another example that uses the Tx object to update the database. We are going to do something a bit laborious that would normally be done in one update statement (update person set score = -1) just to demonstrate usage:

Output:

Callable statement support is a major addition to the code base as of 0.1-RC23.

Callable support is present only outside of transactions (transaction support coming later). If you’re keen for it, raise an issue. The primary impediment is the duplication of a bunch of chained builders for the transacted case.

For example:

Note above that each question mark in the call statement correponds in order with a call to in() or out(…​). Once all parameters have been defined then the input(0, 10, 20) call drives the running of the query with that input. The output Flowable is strongly typed according to the out parameters specified.

When you start specifying output ResultSet s from the call then you lose output parameter strong typing but gain ResultSet mapped strong typing as per normal select statements in rxjava2-jdbc.

Here’s an example for one in parameter and two output ResultSet s with autoMap. You can of course use getAs instead (or get):

The above example is pretty nifty in that we can zip the two result sets resulting from the call and of course the whole thing was easy to define (as opposed to normal JDBC).

You just saw autoMap used to handle an output ResultSet but getAs works too:

You can explore more examples of this in [DatabaseTest.java](rxjava2-jdbc/src/test/java/org/davidmoten/rx/jdbc/DatabaseTest.java). Search for .call.

A few nifty things in JDBC may not yet directly supported by rxjava2-jdbc but you can get acccess to the underlying Connection s from the Database object by using Database.apply or Database.member().

Here’s an example where you want to return something from a Connection (say you called a stored procedure and returned an integer):

If you don’t want to return something then use a different overload of apply:

Here are lower level versions of the above examples where you take on the responsibility of returning the connection to the pool.

and

Logging is handled by slf4j which bridges to the logging framework of your choice. Add the dependency for your logging framework as a maven dependency and you are sorted. See the test scoped log4j example in rxjava2-jdbc/pom.xml.