= Reactive PostgreSQL Client
:PREPARED_PARAMS: `$1`, `$2`, etc…​

The Reactive PostgreSQL Client is a client for PostgreSQL with a straightforward API focusing on
scalability and low overhead.

The client is reactive and non blocking, allowing to handle many database connections with a single thread.

* Event driven
* Lightweight
* Built-in connection pooling
* Prepared queries caching
* Publish / subscribe using PostgreSQL `NOTIFY/LISTEN`
* Batch and cursor
* Row streaming
* Command pipeling
* RxJava 1 and RxJava 2
* Direct memory to object without unnecessary copies
* Java 8 Date and Time
* SSL/TLS
* Unix domain socket
* HTTP/1.x CONNECT, SOCKS4a or SOCKS5 proxy support

== Usage

To use the Reactive PostgreSQL Client add the following dependency to the _dependencies_ section of your build descriptor:

* Maven (in your `pom.xml`):

[source,xml]
----
<dependency>
 <groupId>io.vertx</groupId>
 <artifactId>vertx-lang-kotlin</artifactId>
 <version>3.9.12</version>
</dependency>
----
* Gradle (in your `build.gradle` file):

[source,groovy]
----
dependencies {
 compile 'io.vertx:vertx-lang-kotlin:3.9.12'
}
----

== Getting started

Here is the simplest way to connect, query and disconnect

[source,kotlin]
----

// Connect options
var connectOptions = PgConnectOptions(
  port = 5432,
  host = "the-host",
  database = "the-db",
  user = "user",
  password = "secret")

// Pool options
var poolOptions = PoolOptions(
  maxSize = 5)

// Create the client pool
var client = PgPool.pool(connectOptions, poolOptions)

// A simple query
client.query("SELECT * FROM users WHERE id='julien'").execute({ ar ->
  if (ar.succeeded()) {
    var result = ar.result()
    println("Got ${result.size()} rows ")
  } else {
    println("Failure: ${ar.cause().getMessage()}")
  }

  // Now close the pool
  client.close()
})

----

== Connecting to PostgreSQL

Most of the time you will use a pool to connect to PostgreSQL:

[source,kotlin]
----

// Connect options
var connectOptions = PgConnectOptions(
  port = 5432,
  host = "the-host",
  database = "the-db",
  user = "user",
  password = "secret")

// Pool options
var poolOptions = PoolOptions(
  maxSize = 5)

// Create the pooled client
var client = PgPool.pool(connectOptions, poolOptions)

----

The pooled client uses a connection pool and any operation will borrow a connection from the pool
to execute the operation and release it to the pool.

If you are running with Vert.x you can pass it your Vertx instance:

[source,kotlin]
----

// Connect options
var connectOptions = PgConnectOptions(
  port = 5432,
  host = "the-host",
  database = "the-db",
  user = "user",
  password = "secret")

// Pool options
var poolOptions = PoolOptions(
  maxSize = 5)
// Create the pooled client
var client = PgPool.pool(vertx, connectOptions, poolOptions)

----

You need to release the pool when you don't need it anymore:

[source,kotlin]
----

// Close the pool and all the associated resources
pool.close()

----

When you need to execute several operations on the same connection, you need to use a client
`link:../../apidocs/io/vertx/pgclient/PgConnection.html[connection]`.

You can easily get one from the pool:

[source,kotlin]
----

// Connect options
var connectOptions = PgConnectOptions(
  port = 5432,
  host = "the-host",
  database = "the-db",
  user = "user",
  password = "secret")

// Pool options
var poolOptions = PoolOptions(
  maxSize = 5)

// Create the pooled client
var client = PgPool.pool(vertx, connectOptions, poolOptions)

// Get a connection from the pool
client.getConnection({ ar1 ->

  if (ar1.succeeded()) {

    println("Connected")

    // Obtain our connection
    var conn = ar1.result()

    // All operations execute on the same connection
    conn.query("SELECT * FROM users WHERE id='julien'").execute({ ar2 ->
      if (ar2.succeeded()) {
        conn.query("SELECT * FROM users WHERE id='emad'").execute({ ar3 ->
          // Release the connection to the pool
          conn.close()
        })
      } else {
        // Release the connection to the pool
        conn.close()
      }
    })
  } else {
    println("Could not connect: ${ar1.cause().getMessage()}")
  }
})

----

Once you are done with the connection you must close it to release it to the pool, so it can be reused.

Sometimes you want to improve performance via Unix domain socket connection, we achieve this with Vert.x Native transports.

Make sure you have added the required `netty-transport-native` dependency in your classpath and enabled the Unix domain socket option.

[source,kotlin]
----

// Connect Options
// Socket file name will be /var/run/postgresql/.s.PGSQL.5432
var connectOptions = PgConnectOptions(
  host = "/var/run/postgresql",
  port = 5432,
  database = "the-db")

// Pool options
var poolOptions = PoolOptions(
  maxSize = 5)

// Create the pooled client
var client = PgPool.pool(connectOptions, poolOptions)

// Create the pooled client with a vertx instance
// Make sure the vertx instance has enabled native transports
var client2 = PgPool.pool(vertx, connectOptions, poolOptions)

----

More information can be found in the [Vert.x documentation](https://vertx.io/docs/vertx-core/java/#_native_transports).

== Configuration

There are several alternatives for you to configure the client.

=== data object

A simple way to configure the client is to specify a `PgConnectOptions` data object.

[source,kotlin]
----

// Data object
var connectOptions = PgConnectOptions(
  port = 5432,
  host = "the-host",
  database = "the-db",
  user = "user",
  password = "secret")

// Pool Options
var poolOptions = PoolOptions(
  maxSize = 5)

// Create the pool from the data object
var pool = PgPool.pool(vertx, connectOptions, poolOptions)

pool.getConnection({ ar ->
  // Handling your connection
})

----

You can also configure the generic properties with the `setProperties` or `addProperty` methods. Note `setProperties` will override the default client properties.

For example, you can set a default schema for the connection with adding a `search_path` property.

[source,kotlin]
----
Code not translatable
----

More information about the available properties can be found in the https://www.postgresql.org/docs/current/runtime-config-client.html[PostgreSQL Manuals].

=== connection uri

Apart from configuring with a `PgConnectOptions` data object, We also provide you an alternative way to connect when you want to configure with a connection URI:

[source,kotlin]
----

// Connection URI
var connectionUri = "postgresql://dbuser:secretpassword@database.server.com:3211/mydb"

// Create the pool from the connection URI
var pool = PgPool.pool(connectionUri)

// Create the connection from the connection URI
PgConnection.connect(vertx, connectionUri, { res ->
  // Handling your connection
})

----

More information about connection string formats can be found in the https://www.postgresql.org/docs/9.6/static/libpq-connect.html#LIBPQ-CONNSTRING[PostgreSQL Manuals].

Currently the client supports the following parameter key words in connection uri

* host
* hostaddr
* port
* user
* password
* dbname
* sslmode
* properties including(application_name, fallback_application_name, search_path)

Note: configuring properties in connection URI will override the default properties.

=== environment variables

You can also use environment variables to set default connection setting values, this is useful
when you want to avoid hard-coding database connection information. You can refer to the https://www.postgresql.org/docs/9.6/static/libpq-envars.html[official documentation]
for more details. The following parameters are supported:

* `PGHOST`
* `PGHOSTADDR`
* `PGPORT`
* `PGDATABASE`
* `PGUSER`
* `PGPASSWORD`
* `PGSSLMODE`

If you don't specify a data object or a connection URI string to connect, environment variables will take precedence over them.

[source,kotlin]
----
$ PGUSER=user \
 PGHOST=the-host \
 PGPASSWORD=secret \
 PGDATABASE=the-db \
 PGPORT=5432 \
 PGSSLMODE=DISABLE
----

[source,kotlin]
----

// Create the pool from the environment variables
var pool = PgPool.pool()

// Create the connection from the environment variables
PgConnection.connect(vertx, { res ->
  // Handling your connection
})

----

=== SASL SCRAM-SHA-256 authentication mechanism.

To use the sasl SCRAM-SHA-256 authentication add the following dependency to the _dependencies_ section of your build descriptor:

* Maven (in your `pom.xml`):

[source,xml]
----
<dependency>
 <groupId>com.ongres.scram</groupId>
 <artifactId>client</artifactId>
 <version>1.0.0-beta.2</version>
</dependency>
----
* Gradle (in your `build.gradle` file):

[source,groovy]
----
dependencies {
 compile 'com.ongres.scram:client:1.0.0-beta.2'
}
----

Note that SCRAM-SHA-256-PLUS (added in Postgresql 11) is not supported.


include::queries.adoc[]

You can fetch generated keys with a 'RETURNING' clause in your query:

[source,kotlin]
----
client.preparedQuery("INSERT INTO color (color_name) VALUES (\$1), (\$2), (\$3) RETURNING color_id").execute(Tuple.of("white", "red", "blue"), { ar ->
  if (ar.succeeded()) {
    var rows = ar.result()
    println(rows.rowCount())
    for (row in rows) {
      println("generated key: ${row.getInteger("color_id")}")
    }
  } else {
    println("Failure: ${ar.cause().getMessage()}")
  }
})

----

A batch query with a `RETURNING` clause creates a `link:../../apidocs/io/vertx/sqlclient/RowSet.html[RowSet]`
containing a single for each element of the batch:

[source,kotlin]
----
client.preparedQuery("INSERT INTO color (color_name) VALUES (\$1) RETURNING color_id").executeBatch(listOf(Tuple.of("white"), Tuple.of("red"), Tuple.of("blue")), { ar ->
  if (ar.succeeded()) {
    var rows = ar.result()
    while (rows.next() != null) {
      var colorId = rows.iterator().next().getInteger("color_id")
      println("generated key: ${colorId}")

      rows = rows.next()}

  } else {
    println("Failure: ${ar.cause().getMessage()}")
  }
})

----

include::connections.adoc[]

include::transactions.adoc[]

include::cursor.adoc[]

Note: PostreSQL destroys cursors at the end of a transaction, so the cursor API shall be used
within a transaction, otherwise you will likely get the `34000` PostgreSQL error.

== PostgreSQL type mapping

Currently the client supports the following PostgreSQL types

* BOOLEAN (`java.lang.Boolean`)
* INT2 (`java.lang.Short`)
* INT4 (`java.lang.Integer`)
* INT8 (`java.lang.Long`)
* FLOAT4 (`java.lang.Float`)
* FLOAT8 (`java.lang.Double`)
* CHAR (`java.lang.String`)
* VARCHAR (`java.lang.String`)
* TEXT (`java.lang.String`)
* ENUM (`java.lang.String`)
* NAME (`java.lang.String`)
* SERIAL2 (`java.lang.Short`)
* SERIAL4 (`java.lang.Integer`)
* SERIAL8 (`java.lang.Long`)
* NUMERIC (`io.vertx.sqlclient.data.Numeric`)
* UUID (`java.util.UUID`)
* DATE (`java.time.LocalDate`)
* TIME (`java.time.LocalTime`)
* TIMETZ (`java.time.OffsetTime`)
* TIMESTAMP (`java.time.LocalDateTime`)
* TIMESTAMPTZ (`java.time.OffsetDateTime`)
* INTERVAL (`io.vertx.pgclient.data.Interval`)
* BYTEA (`io.vertx.core.buffer.Buffer`)
* JSON (`io.vertx.core.json.JsonObject`, `io.vertx.core.json.JsonArray`, `Number`, `Boolean`, `String`, `io.vertx.sqlclient.Tuple#JSON_NULL`)
* JSONB (`io.vertx.core.json.JsonObject`, `io.vertx.core.json.JsonArray`, `Number`, `Boolean`, `String`, `io.vertx.sqlclient.Tuple#JSON_NULL`)
* POINT (`io.vertx.pgclient.data.Point`)
* LINE (`io.vertx.pgclient.data.Line`)
* LSEG (`io.vertx.pgclient.data.LineSegment`)
* BOX (`io.vertx.pgclient.data.Box`)
* PATH (`io.vertx.pgclient.data.Path`)
* POLYGON (`io.vertx.pgclient.data.Polygon`)
* CIRCLE (`io.vertx.pgclient.data.Circle`)
* TSVECTOR (`java.lang.String`)
* TSQUERY (`java.lang.String`)

Tuple decoding uses the above types when storing values, it also performs on the flu conversion the actual value when possible:

[source,kotlin]
----
pool.query("SELECT 1::BIGINT \"VAL\"").execute({ ar ->
  var rowSet = ar.result()
  var row = rowSet.iterator().next()

  // Stored as java.lang.Long
  var value = row.getValue(0)

  // Convert to java.lang.Integer
  var intValue = row.getInteger(0)
})

----

Tuple encoding uses the above type mapping for encoding, unless the type is numeric in which case `java.lang.Number` is used instead:

[source,kotlin]
----
pool.query("SELECT 1::BIGINT \"VAL\"").execute({ ar ->
  var rowSet = ar.result()
  var row = rowSet.iterator().next()

  // Stored as java.lang.Long
  var value = row.getValue(0)

  // Convert to java.lang.Integer
  var intValue = row.getInteger(0)
})

----

Arrays of these types are supported.

=== Handling JSON

PostgreSQL `JSON` and `JSONB` types are represented by the following Java types:

- `String`
- `Number`
- `Boolean`
- `io.vertx.core.json.JsonObject`
- `io.vertx.core.json.JsonArray`
- `io.vertx.sqlclient.Tuple#JSON_NULL` for representing the JSON null literal

[source,kotlin]
----
Code not translatable
----

=== Handling NUMERIC

The `link:../../apidocs/io/vertx/sqlclient/data/Numeric.html[Numeric]` Java type is used to represent the PostgreSQL `NUMERIC` type.

[source,kotlin]
----
var numeric = row.get(io.vertx.sqlclient.data.Numeric.`class`, 0)
if (numeric.isNaN()) {
  // Handle NaN
} else {
  var value = numeric.bigDecimalValue()
}

----

== Handling arrays

Arrays are available on `link:../../apidocs/io/vertx/sqlclient/Tuple.html[Tuple]` and `link:../../apidocs/io/vertx/sqlclient/Row.html[Row]`:

[source,kotlin]
----
// Create a tuple with a single array
var tuple = Tuple.of(arrayOf("a", "tuple", "with", "arrays"))

// Add a string array to the tuple
tuple.addArrayOfString(arrayOf("another", "array"))

// Get the first array of string
var array = tuple.getArrayOfStrings(0)

----

== Handling Date/Time infinity

PostgreSQL defines special values to represent infinity.

The max/min constants of the corresponding type represents https://www.postgresql.org/docs/9.0/datatype-datetime.html[special value].

- `OffsetDateTime.MAX`/OffsetDateTime.MIN`
- `LocalDateTime.MAX`/LocalDateTime.MIN`
- `LocalDate.MAX`/LocalDate.MIN`

[source,kotlin]
----
client.query("SELECT 'infinity'::DATE \"LocalDate\"").execute({ ar ->
  if (ar.succeeded()) {
    var row = ar.result().iterator().next()
    println(row.getLocalDate("LocalDate") == java.time.LocalDate.MAX)
  } else {
    println("Failure: ${ar.cause().getMessage()}")
  }
})

----

== Handling custom types

Strings are used to represent custom types, both sent to and returned from Postgres.

You can read from PostgreSQL and get the custom type as a string

[source,kotlin]
----
client.preparedQuery("SELECT address, (address).city FROM address_book WHERE id=\$1").execute(Tuple.of(3), { ar ->
  if (ar.succeeded()) {
    var rows = ar.result()
    for (row in rows) {
      println("Full Address ${row.getString(0)}, City ${row.getString(1)}")
    }
  } else {
    println("Failure: ${ar.cause().getMessage()}")
  }
})

----

You can also write to PostgreSQL by providing a string

[source,kotlin]
----
client.preparedQuery("INSERT INTO address_book (id, address) VALUES (\$1, \$2)").execute(Tuple.of(3, "('Anytown', 'Second Ave', false)"), { ar ->
  if (ar.succeeded()) {
    var rows = ar.result()
    println(rows.rowCount())
  } else {
    println("Failure: ${ar.cause().getMessage()}")
  }
})

----

== Handling text search

Text search is handling using java `String`

[source,kotlin]
----
client.preparedQuery("SELECT to_tsvector( \$1 ) @@ to_tsquery( \$2 )").execute(Tuple.of("fat cats ate fat rats", "fat & rat"), { ar ->
  if (ar.succeeded()) {
    var rows = ar.result()
    for (row in rows) {
      println("Match : ${row.getBoolean(0)}")
    }
  } else {
    println("Failure: ${ar.cause().getMessage()}")
  }
})

----

`tsvector` and `tsquery` can be fetched from db using java `String`

[source,kotlin]
----
client.preparedQuery("SELECT to_tsvector( \$1 ), to_tsquery( \$2 )").execute(Tuple.of("fat cats ate fat rats", "fat & rat"), { ar ->
  if (ar.succeeded()) {
    var rows = ar.result()
    for (row in rows) {
      println("Vector : ${row.getString(0)}, query : ${row.getString(1)}")
    }
  } else {
    println("Failure: ${ar.cause().getMessage()}")
  }
})

----

== Collector queries

You can use Java collectors with the query API:

[source,kotlin]
----
Code not translatable
----

The collector processing must not keep a reference on the `link:../../apidocs/io/vertx/sqlclient/Row.html[Row]` as
there is a single row used for processing the entire set.

The Java `Collectors` provides many interesting predefined collectors, for example you can
create easily create a string directly from the row set:

[source,kotlin]
----
Code not translatable
----

== Pub/sub

PostgreSQL supports pub/sub communication channels.

You can set a `link:../../apidocs/io/vertx/pgclient/PgConnection.html#notificationHandler-io.vertx.core.Handler-[notificationHandler]` to receive
PostgreSQL notifications:

[source,kotlin]
----

connection.notificationHandler({ notification ->
  println("Received ${notification.payload} on channel ${notification.channel}")
})

connection.query("LISTEN some-channel").execute({ ar ->
  println("Subscribed to channel")
})

----

The `link:../../apidocs/io/vertx/pgclient/pubsub/PgSubscriber.html[PgSubscriber]` is a channel manager managing a single connection that
provides per channel subscription:

[source,kotlin]
----

var subscriber = PgSubscriber.subscriber(vertx, PgConnectOptions(
  port = 5432,
  host = "the-host",
  database = "the-db",
  user = "user",
  password = "secret"))

// You can set the channel before connect
subscriber.channel("channel1").handler({ payload ->
  println("Received ${payload}")
})

subscriber.connect({ ar ->
  if (ar.succeeded()) {

    // Or you can set the channel after connect
    subscriber.channel("channel2").handler({ payload ->
      println("Received ${payload}")
    })
  }
})

----

The channel name that is given to the channel method will be the exact name of the channel as held by PostgreSQL for sending
notifications.  Note this is different than the representation of the channel name in SQL, and
internally `link:../../apidocs/io/vertx/pgclient/pubsub/PgSubscriber.html[PgSubscriber]` will prepare the submitted channel name as a quoted identifier:

[source,kotlin]
----

var subscriber = PgSubscriber.subscriber(vertx, PgConnectOptions(
  port = 5432,
  host = "the-host",
  database = "the-db",
  user = "user",
  password = "secret"))

subscriber.connect({ ar ->
  if (ar.succeeded()) {
    // Complex channel name - name in PostgreSQL requires a quoted ID
    subscriber.channel("Complex.Channel.Name").handler({ payload ->
      println("Received ${payload}")
    })
    subscriber.channel("Complex.Channel.Name").subscribeHandler({ subscribed ->
      subscriber.actualConnection().query("NOTIFY \"Complex.Channel.Name\", 'msg'").execute({ notified ->
        println("Notified \"Complex.Channel.Name\"")
      })
    })

    // PostgreSQL simple ID's are forced lower-case
    subscriber.channel("simple_channel").handler({ payload ->
      println("Received ${payload}")
    })
    subscriber.channel("simple_channel").subscribeHandler({ subscribed ->
      // The following simple channel identifier is forced to lower case
      subscriber.actualConnection().query("NOTIFY Simple_CHANNEL, 'msg'").execute({ notified ->
        println("Notified simple_channel")
      })
    })

    // The following channel name is longer than the current
    // (NAMEDATALEN = 64) - 1 == 63 character limit and will be truncated
    subscriber.channel("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaabbbbb").handler({ payload ->
      println("Received ${payload}")
    })
  }
})

----
You can provide a reconnect policy as a function that takes the number of `retries` as argument and returns an `amountOfTime`
value:

* when `amountOfTime < 0`: the subscriber is closed and there is no retry
* when `amountOfTime = 0`: the subscriber retries to connect immediately
* when `amountOfTime > 0`: the subscriber retries after `amountOfTime` milliseconds

[source,kotlin]
----

var subscriber = PgSubscriber.subscriber(vertx, PgConnectOptions(
  port = 5432,
  host = "the-host",
  database = "the-db",
  user = "user",
  password = "secret"))

// Reconnect at most 10 times after 100 ms each
subscriber.reconnectPolicy({ retries ->
  if (retries < 10) {
    return 100L
  } else {
    return -1L
  }
})

----

The default policy is to not reconnect.

== Cancelling Request

PostgreSQL supports cancellation of requests in progress. You can cancel inflight requests using `link:../../apidocs/io/vertx/pgclient/PgConnection.html#cancelRequest-io.vertx.core.Handler-[cancelRequest]`. Cancelling a request opens a new connection to the server and cancels the request and then close the connection.

[source,kotlin]
----
connection.query("SELECT pg_sleep(20)").execute({ ar ->
  if (ar.succeeded()) {
    // imagine this is a long query and is still running
    println("Query success")
  } else {
    // the server will abort the current query after cancelling request
    println("Failed to query due to ${ar.cause().getMessage()}")
  }
})
connection.cancelRequest({ ar ->
  if (ar.succeeded()) {
    println("Cancelling request has been sent")
  } else {
    println("Failed to send cancelling request")
  }
})

----

> The cancellation signal might or might not have any effect — for example, if it arrives after the backend has finished processing the query, then it will have no effect. If the cancellation is effective, it results in the current command being terminated early with an error message.

More information can be found in the https://www.postgresql.org/docs/11/protocol-flow.html#id-1.10.5.7.9[official documentation].

== Using SSL/TLS

To configure the client to use SSL connection, you can configure the `link:../../apidocs/io/vertx/pgclient/PgConnectOptions.html[PgConnectOptions]`
like a Vert.x `NetClient`.
All https://www.postgresql.org/docs/current/libpq-ssl.html#LIBPQ-SSL-PROTECTION[SSL modes] are supported and you are able to configure `sslmode`. The client is in `DISABLE` SSL mode by default.
`ssl` parameter is kept as a mere shortcut for setting `sslmode`. `setSsl(true)` is equivalent to `setSslMode(VERIFY_CA)` and `setSsl(false)` is equivalent to `setSslMode(DISABLE)`.

[source,kotlin]
----

var options = PgConnectOptions(
  port = 5432,
  host = "the-host",
  database = "the-db",
  user = "user",
  password = "secret",
  sslMode = SslMode.VERIFY_CA,
  pemTrustOptions = PemTrustOptions(
    certPaths = listOf("/path/to/cert.pem")))

PgConnection.connect(vertx, options, { res ->
  if (res.succeeded()) {
    // Connected with SSL
  } else {
    println("Could not connect ${res.cause()}")
  }
})

----

More information can be found in the http://vertx.io/docs/vertx-core/java/#ssl[Vert.x documentation].

== Using a proxy

You can also configure the client to use an HTTP/1.x CONNECT, SOCKS4a or SOCKS5 proxy.

More information can be found in the http://vertx.io/docs/vertx-core/java/#_using_a_proxy_for_client_connections[Vert.x documentation].

ifeval::["kotlin" == "java"]
include::override/rxjava2.adoc[]
endif::[]