Add Logical Decoding API

We now support consumption of replication streams through
logical and physical decoding. The API exposes functionality
to create replication slots and to consume a replication stream.

ReplicationSlotRequest request = ReplicationSlotRequest.logical().slotName("my_slot").outputPlugin("test_decoding").temporary().build();
Mono<ReplicationSlot> replicationSlot = replicationConnection.createSlot(request);
Mono<ReplicationStream> replicationStream = replicationConnection.startReplication(replicationRequest);

// Later

Flux<…> replicationMessages = replicationStream.map(byteBuf -> …);
Mono<Void> close = replicationStream.close();

[resolves #18][resolves #179]

Author: mp911de

README.md

@@ -10,12 +10,13 @@ This driver provides the following features:
 * TLS
 * Explicit transactions
 * Notifications
+* Logical Decode
 * Binary data transfer
 * Execution of prepared statements with bindings
 * Execution of batch statements without bindings
 * Binary data transfer
 * Read and write support for all data types except LOB types (e.g. `BLOB`, `CLOB`)
-* Extension points to register `Codec`s to handle additional Postgres data types
+* Extension points to register `Codec`s to handle additional PostgreSQL data types
 
 Next steps:
 
@@ -160,7 +161,7 @@ If you'd rather like the latest snapshots of the upcoming major version, use our
 
 ## Listen/Notify
 
-Listen and Notify provide a simple form of signal or inter-process communication mechanism for processes accessing the same Postgres database.
+Listen and Notify provide a simple form of signal or inter-process communication mechanism for processes accessing the same PostgreSQL database.
 For Listen/Notify, two actors are involved: The sender (notify) and the receiver (listen). The following example uses two connections
 to illustrate how they work together:
 
@@ -184,7 +185,7 @@ The second connection broadcasts a notification to the `mymessage` channel upon
 
 ## JSON/JSONB support
 
-Postgres supports JSON by storing values in `JSON`/`JSONB` columns. These values can be consumed and written using the regular R2DBC SPI and by using driver-specific extensions with the `io.r2dbc.postgresql.codec.Json` type.
+PostgreSQL supports JSON by storing values in `JSON`/`JSONB` columns. These values can be consumed and written using the regular R2DBC SPI and by using driver-specific extensions with the `io.r2dbc.postgresql.codec.Json` type.
 
 You can choose from two approaches:
 
@@ -238,6 +239,47 @@ The following types are supported for JSON exchange:
 * `String`
 * `InputStream` (must be closed after usage to avoid memory leaks)
 
+## Logical Decode
+
+PostgreSQL allows replication streaming and decoding persistent changes to a database's tables into useful chunks of data.
+In PostgreSQL, logical decoding is implemented by decoding the contents of the write-ahead log, which describe changes on a storage level, into an application-specific form such as a stream of tuples or SQL statements.
+
+Consuming the replication stream is a four-step process:
+
+1. Obtain a replication connection via `PostgresqlConnectionFactory.replication()`.
+2. Create a replication slot (physical/logical).
+3. Initiate replication using the replication slot.
+4. Once the replication stream is set up, you can consume and map the binary data using `ReplicationStream.map(…)`.
+
+On application shutdown, `close()` the `ReplicationStream`.
+
+Note that a connection is busy once the replication is active and a connection can have at most one active replication stream.  
+
+```java
+
+Mono<PostgresqlReplicationConnection> replicationMono = connectionFactory.replication();
+
+// later:
+ReplicationSlotRequest request = ReplicationSlotRequest.logical()
+                                        .slotName("my_slot")
+                                        .outputPlugin("test_decoding")
+                                        .temporary()
+                                        .build();
+Mono<ReplicationSlot> createSlot = replicationConnection.createSlot(request);
+
+ReplicationRequest replicationRequest = ReplicationRequest.logical()
+                                        .slotName("my_slot")
+                                        .startPosition(LogSequenceNumber.valueOf(0))
+                                        .slotOption("skip-empty-xacts", true)
+                                        .slotOption("include-xids", false)
+                                        .build();
+
+Flux<T> replicationStream = replicationConnection.startReplication(replicationRequest).flatMapMany(it -> {
+    return it.map(byteBuf -> {…})
+        .doOnError(t -> it.close().subscribe());
+});
+```
+
 ## Data Type Mapping
 
 This reference table shows the type mapping between [PostgreSQL][p] and Java data types:
@@ -360,7 +402,7 @@ Support for the following single-dimensional arrays (read and write):
 ## Extension mechanism
 This driver accepts the following extensions:
 
-* `CodecRegistrar` to contribute `Codec`s for Postgres ObjectIDs. 
+* `CodecRegistrar` to contribute `Codec`s for PostgreSQL ObjectIDs. 
 
 Extensions can be registered programmatically using `PostgresConnectionConfiguration` or discovered using Java's `ServiceLoader` mechanism (from `META-INF/services/io.r2dbc.postgresql.extension.Extension`).
 

src/main/java/io/r2dbc/postgresql/DefaultPostgresqlReplicationConnection.java

@@ -0,0 +1,113 @@
+/*
+ * Copyright 2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.r2dbc.postgresql;
+
+import io.r2dbc.postgresql.api.PostgresqlConnectionMetadata;
+import io.r2dbc.postgresql.client.Client;
+import io.r2dbc.postgresql.message.backend.BackendMessage;
+import io.r2dbc.postgresql.message.backend.EmptyQueryResponse;
+import io.r2dbc.postgresql.message.backend.ErrorResponse;
+import io.r2dbc.postgresql.message.backend.ReadyForQuery;
+import io.r2dbc.postgresql.message.frontend.FrontendMessage;
+import io.r2dbc.postgresql.message.frontend.Query;
+import io.r2dbc.postgresql.replication.LogSequenceNumber;
+import io.r2dbc.postgresql.replication.ReplicationRequest;
+import io.r2dbc.postgresql.replication.ReplicationSlot;
+import io.r2dbc.postgresql.replication.ReplicationSlotRequest;
+import io.r2dbc.postgresql.replication.ReplicationStream;
+import io.r2dbc.postgresql.util.Assert;
+import io.r2dbc.spi.Row;
+import reactor.core.publisher.EmitterProcessor;
+import reactor.core.publisher.Mono;
+
+import java.util.function.Predicate;
+
+import static io.r2dbc.postgresql.util.PredicateUtils.or;
+
+/**
+ * Postgres replication connection.
+ */
+final class DefaultPostgresqlReplicationConnection implements io.r2dbc.postgresql.api.PostgresqlReplicationConnection {
+
+    private static final Predicate<BackendMessage> WINDOW_UNTIL = or(ReadyForQuery.class::isInstance, EmptyQueryResponse.class::isInstance, ErrorResponse.class::isInstance);
+
+    private final PostgresqlConnection delegate;
+
+    private final Client client;
+
+    DefaultPostgresqlReplicationConnection(PostgresqlConnection delegate) {
+        this.delegate = delegate;
+        this.client = delegate.getClient();
+    }
+
+    @Override
+    public Mono<Void> close() {
+        return this.delegate.close();
+    }
+
+    @Override
+    public Mono<ReplicationSlot> createSlot(ReplicationSlotRequest request) {
+
+        Assert.requireNonNull(request, "request must not be null");
+
+        return this.delegate.createStatement(request.asSQL()).execute().flatMap(it -> {
+
+            return it.map((row, rowMetadata) -> getReplicationSlot(request, row));
+        }).last();
+    }
+
+    private static ReplicationSlot getReplicationSlot(ReplicationSlotRequest request, Row row) {
+        return new ReplicationSlot(
+            getString(row, "slot_name"),
+            request.getReplicationType(),
+            LogSequenceNumber.valueOf(getString(row, "consistent_point")),
+            row.get("snapshot_name", String.class),
+            row.get("output_plugin", String.class));
+    }
+
+    @Override
+    public Mono<ReplicationStream> startReplication(ReplicationRequest request) {
+
+        Assert.requireNonNull(request, "request must not be null");
+
+        String sql = request.asSQL();
+        ExceptionFactory exceptionFactory = ExceptionFactory.withSql(sql);
+
+        EmitterProcessor<FrontendMessage> requestProcessor = EmitterProcessor.create();
+
+        return Mono.fromDirect(this.client.exchange(requestProcessor.startWith(new Query(sql)))
+            .handle(exceptionFactory::handleErrorResponse)
+            .windowUntil(WINDOW_UNTIL)
+            .map(messages -> {
+                return new PostgresReplicationStream(this.client.getByteBufAllocator(), request, requestProcessor, messages);
+            }));
+    }
+
+    @Override
+    public PostgresqlConnectionMetadata getMetadata() {
+        return this.delegate.getMetadata();
+    }
+
+    private static String getString(Row row, String column) {
+        String value = row.get(column, String.class);
+        if (value == null) {
+            throw new IllegalStateException(String.format("No value found for column %s", column));
+        }
+        return value;
+    }
+
+}

src/main/java/io/r2dbc/postgresql/KeepAliveMessage.java

@@ -0,0 +1,71 @@
+/*
+ * Copyright 2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.r2dbc.postgresql;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+import io.r2dbc.postgresql.message.frontend.CopyData;
+import io.r2dbc.postgresql.replication.LogSequenceNumber;
+
+/**
+ * Nested message for a Replication Stream using {@link CopyData} as carrier.
+ */
+final class KeepAliveMessage {
+
+    private static final byte KEEP_ALIVE_REPLY = 'r';
+
+    private static final int NO_REPLY_REQUIRED = 0;
+
+    private static final int REPLY_REQUIRED = 1;
+
+    private final LogSequenceNumber received;
+
+    private final LogSequenceNumber flushed;
+
+    private final LogSequenceNumber applied;
+
+    private final long systemClock;
+
+    private boolean replyRequired;
+
+    KeepAliveMessage(LogSequenceNumber received, LogSequenceNumber flushed, LogSequenceNumber applied, long systemClock, boolean replyRequired) {
+        this.received = received;
+        this.flushed = flushed;
+        this.applied = applied;
+        this.systemClock = systemClock;
+        this.replyRequired = replyRequired;
+    }
+
+    public ByteBuf encode(ByteBufAllocator allocator) {
+
+        ByteBuf out = allocator.buffer(34);
+
+        out.writeByte(KEEP_ALIVE_REPLY);
+        out.writeLong(this.received.asLong());
+        out.writeLong(this.flushed.asLong());
+        out.writeLong(this.applied.asLong());
+        out.writeLong(this.systemClock);
+
+        if (this.replyRequired) {
+            out.writeByte((byte) REPLY_REQUIRED);
+        } else {
+            out.writeByte(this.received == LogSequenceNumber.INVALID_LSN ? (byte) REPLY_REQUIRED : (byte) NO_REPLY_REQUIRED);
+        }
+
+        return out;
+    }
+}