rxjava-jdbc

Efficient execution, concise code, and functional composition of database calls using JDBC and RxJava Observable.

Status: Released to Maven Central

See also rxjava2-jdbc for RxJava 2.x with non-blocking connection pools!

Release Notes

Features

• Functionally compose database queries run sequentially or in parallel
• Queries may be only partially run or indeed never run due to subscription cancellations thus improving efficiency
• Concise code
• Queries can depend on completion of other Observables and can be supplied parameters through Observables.
• Method chaining just leads the way (once you are on top of the RxJava api of course!)
• All the RxJava goodness!
• Automatically maps query result rows into typed tuples or your own classes
• CLOB and BLOB handling is simplified greatly

Maven site reports are here including javadoc.

Table of Contents

• rxjava-jdbc
  • Features
  • Todo
  • Build instructions
  • Getting started
  • Query types
  • Functional composition of JDBC calls
  • About toBlocking
  • Dependencies
  • Mixing explicit and Observable parameters
  • Passing multiple parameter sets to a query
  • Named parameters
  • Processing a ResultSet
  • Mapping
  • Explicit mapping
  • Automap
    • Automap using an interface
    • Automap using a concrete class
  • Auto mappings
  • Tuples
    • Tuple2
    • TupleN
  • Returning generated keys
  • Large objects support
    • Insert a Clob
    • Insert a Null Clob
    • Read a Clob
    • Insert a Blob
    • Insert a Null Blob
    • Read a Blob
  • Compose
  • Transactions
    • Transactions as dependency
    • onNext Transactions
  • Asynchronous queries
  • Backpressure
  • Logging
  • Database Connection Pools
  • Using a custom connection pool
  • Use a single Connection
  • Note for SQLite Users

Todo

• Callable statements

Build instructions

git clone https://github.com/davidmoten/rxjava-jdbc.git
cd rxjava-jdbc
mvn clean install

Getting started

Include this maven dependency in your pom (available in Maven Central):

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

After using RxJava on a work project and being very impressed with it (even without Java 8 lambdas!), I wondered what it could offer for JDBC usage. The answer is lots!

Here's a simple example:

Database db = Database.from(url);
List<String> names = db
        .select("select name from person where name > ? order by name")
        .parameter("ALEX")
        .getAs(String.class)
        .toList().toBlocking().single();
System.out.println(names);

output:

[FRED, JOSEPH, MARMADUKE]

Without using rxjava-jdbc:

String sql = "select name from person where name > ? order by name";
try (Connection con = nextConnection();
     PreparedStatement ps = con.prepareStatement(sql);) {
    ps.setObject(1, "ALEX");
    List<String> list = new ArrayList<String>();
    try (ResultSet rs = ps.executeQuery()) {
        while (rs.next()) {
            list.add(rs.getString(1));
        }
    }
    System.out.println(list);
} catch (SQLException e) {
    throw new RuntimeException(e);
}

Query types

The Database.select() method is used for * SQL select queries.

The Database.update() method is used for * update * insert * delete * DDL (like create table, etc)

Examples of all of the above methods are found in the sections below.

Functional composition of JDBC calls

Here's an example, wonderfully brief compared to normal JDBC usage:

import com.github.davidmoten.rx.jdbc.Database;
import rx.Observable;

// use composition to find the first person alphabetically with
// a score less than the person with the last name alphabetically
// whose name is not XAVIER. Two threads and connections will be used.

Database db = new Database(connectionProvider);
Observable<Integer> score = db
        .select("select score from person where name <> ? order by name")
        .parameter("XAVIER")
        .getAs(Integer.class)
        .last();
String name = db
        .select("select name from person where score < ? order by name")
        .parameters(score)
        .getAs(String.class)
        .first()
        .toBlocking().single();
assertEquals("FRED", name);

or alternatively using the Observable.compose() method to chain everything in one command:

String name = db
    .select("select score from person where name <> ? order by name")
    .parameter("XAVIER")
    .getAs(Integer.class)
    .last()
    .compose(db.select("select name from person where score < ? order by name")
            .parameterTransformer()
            .getAs(String.class))
    .first()
    .toBlocking().single();

About toBlocking

You'll see toBlocking() used in the examples in this page and in the unit tests but in your application code you should try to avoid using it. The most benefit from the reactive style is obtained by not leaving the monad. That is, stay in Observable land and make the most of it. Chain everything together and leave toBlocking to an endpoint or better still just subscribe with a Subscriber.

Dependencies

You can setup chains of dependencies that will determine the order of running of queries.

To indicate that a query cannot be run before one or more other Observables have been completed use the dependsOn() method. Here's an example:

Observable<Integer> insert = db
        .update("insert into person(name,score) values(?,?)")
        .parameters("JOHN", 45)
        .count()
        .map(Util.<Integer> delay(500));
int count = db
        .select("select name from person")
        .dependsOn(insert)
        .get()
        .count()
        .toBlocking().single();
assertEquals(4, count);

Note that when you pass the output of a query as a parameter to another query there is an implicit dependency established.

Mixing explicit and Observable parameters

Example:

String name= db
    .select("select name from person where name > ?  and score < ? order by name")
    .parameter("BARRY")
    .parameters(Observable.just(100))
    .getAs(String.class)
    .first()
    .toBlocking().single();
assertEquals("FRED",name);

Passing multiple parameter sets to a query

Given a sequence of parameters, each chunk of parameters will be run with the query and the results appended. In the example below there is only one parameter in the sql statement yet two parameters are specified. This causes the statement to be run twice.

List<Integer> list = 
    db.select("select score from person where name=?")
        .parameter("FRED").parameter("JOSEPH")
        .getAs(Integer.class).toList().toBlocking().single();
assertEquals(Arrays.asList(21,34),list);

Named parameters

Examples:

Observable<String> names = db
    .select("select name from person where score >= :min and score <=:max")
    .parameter("min", 24)
    .parameter("max", 26)
    .getAs(String.class);

Using a map of parameters:

Map<String, Integer> map = new HashMap<String, Integer>();
map.put("min", 24);
map.put("max", 26);
Observable<String> names = db
    .select("select name from person where score >= :min and score <=:max")
    .parameters(map)
    .getAs(String.class);

Using an Observable of maps:

Observable<String> names = db
    .select("select name from person where score >= :min and score <=:max")
    .parameters(Observable.just(map1, map2))
    .getAs(String.class);

Processing a ResultSet

Many operators in rxjava process items pushed to them asynchronously. Given this it is important that ResultSet query results are processed before being emitted to a consuming operator. This means that the select query needs to be passed a function that converts a ResultSet to a result that does not depend on an open java.sql.Connection. Use the get(), getAs(), getTuple?(), and autoMap() methods to specify this function as below.

Observable<Integer> scores = db.select("select score from person where name=?")
        .parameter("FRED")
        .getAs(Integer.class);

Mapping

A common requirement is to map the rows of a ResultSet to an object. There are two main options: explicit mapping and automap.

Explicit mapping

Using get you can map the ResultSet as you wish:

db.select("select name, score from person")
  .get( rs -> new Person(rs.getString(1), rs.getInt(2)));

Automap

automap does more for you than explicit mapping. You can provide just an annotated interface and objects will be created that implement that interface and types will be converted for you (See Auto mappings section below).

There is some reflection overhead with using auto mapping. Use your own benchmarks to determine if its important to you (the reflection overhead may not be significant compared to the network latencies involved in database calls).

The autoMap method maps result set rows to instances of the class you nominate.

If you nominate an interface then dynamic proxies (a java reflection feature) are used to build instances.

If you nominate a concrete class then the columns of the result set are mapped to parameters in the constructor (again using reflection).

Automap using an interface

Create an annotated interface (introduced in rxjava-jdbc 0.5.8):

public interface Person {

    @Column("name")
    String name();

    @Column("score")
    int score();
}

Then run

Observable<Person> persons = db
                 .select("select name, score from person order by name")
                 .autoMap(Person.class);

Easy eh!

An alternative is to annotate the interface with the indexes of the columns in the result set row:

public interface Person {

    @Index(1)
    String name();

    @Index(2)
    int score();
}

Camel cased method names will be converted to underscore by default (since 0.5.11):

public interface Address {

    @Column // maps to address_id 
    int addressId();

    @Column // maps to full_address
    String fullAddress();
}

You can also specify the sql to be run in the annotation:

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

    @Column
    String name();

    @Column
    int score();
}

Then run like this:

Observable<Person> persons = db
                 .select().autoMap(Person.class);

Automap using a concrete class

Given this class:

static class Person {
    private final String name;
    private final double score;
    private final Long dateOfBirth;
    private final Long registered;

    Person(String name, Double score, Long dateOfBirth,
            Long registered) {
            ...

Then run

Observable<Person> persons = db
                .select("select name,score,dob,registered from person order by name")
                .autoMap(Person.class);

The main requirement is that the number of columns in the select statement must match the number of columns in a constructor of Person and that the column types can be automatically mapped to the types in the constructor.

Auto mappings

The automatic mappings below of objects are used in the autoMap() method and for typed getAs() calls. * java.sql.Date,java.sql.Time,java.sql.Timestamp <==> java.util.Date * java.sql.Date,java.sql.Time,java.sql.Timestamp ==> java.lang.Long * java.sql.Blob <==> java.io.InputStream, byte[] * java.sql.Clob <==> java.io.Reader, String * java.math.BigInteger ==> Long, Integer, Decimal, Float, Short, java.math.BigDecimal * java.math.BigDecimal ==> Long, Integer, Decimal, Float, Short, java.math.BigInteger

Note that automappings do not occur to primitives so use Long instead of long.

Tuples