Postgres-async-driver is a non-blocking Java driver for PostgreSQL. The driver supports connection pooling, prepared statements, transactions, all standard SQL types and custom column types.
Postgres-async-driver is available on Maven Central.
<dependency>
<groupId>com.github.alaisi.pgasync</groupId>
<artifactId>postgres-async-driver</artifactId>
<version>0.9</version>
</dependency>
Querying for a set returns an rx.Observable that emits a single ResultSet.
Db db = ...;
db.querySet("select 'Hello world!' as message")
.map(result -> result.row(0).getString("message"))
.subscribe(System.out::println)
// => Hello world
Querying for rows returns an rx.Observable that emits 0-n Rows. The rows are emitted immediately as they are received from the server instead of waiting for the entire query to complete.
Db db = ...;
db.queryRows("select unnest('{ hello, world }'::text[] as message)")
.map(row -> row.getString("message"))
.subscribe(System.out::println)
// => hello
// => world
Db is a connection pool that is created with ConnectionPoolBuilder
Db db = new ConnectionPoolBuilder()
.hostname("localhost")
.port(5432)
.database("db")
.username("user")
.password("pass")
.poolSize(20)
.build();
Each connection pool will start only one IO thread used in communicating with PostgreSQL backend and executing callbacks for all connections.
Prepared statements use native PostgreSQL syntax $index
. Supported parameter types are all primitive types, String
, BigDecimal
, BigInteger
, UUID
, temporal types in java.sql
package and byte[]
.
db.querySet("insert into message(id, body) values($1, $2)", 123, "hello")
.subscribe(result -> out.printf("Inserted %d rows", result.affectedRows() ));
A transactional unit of work is started with begin()
. Queries issued to the emitted Transaction are executed in the same transaction and the tx is automatically rolled back on query failure.
db.begin()
.flatMap(tx -> tx.querySet("insert into products (name) values ($1) returning id", "saw")
.map(productsResult -> productsResult.row(0).getLong("id"))
.flatMap(id -> tx.querySet("insert into promotions (product_id) values ($1)", id))
.flatMap(promotionsResult -> tx.commit())
).subscribe(
__ -> System.out.println("Transaction committed"),
Throwable::printStackTrace);
Support for additional data types requires registering converters to ConnectionPoolBuilder
class JsonConverter implements Converter<example.Json> {
@Override
public Class<example.Json> type() {
return example.Json.class;
}
@Override
public byte[] from(example.Json json) {
return json.toBytes();
}
@Override
public example.Json to(Oid oid, byte[] value) {
return new example.Json(new String(value, UTF_8));
}
}
Db db = return new ConnectionPoolBuilder()
...
.converters(new JsonConverter())
.build();