Merge pull request #23 from liubao68/work

support logic replication

Author: liubao68

README.md

@@ -341,50 +341,17 @@ connection.createStatement("SELECT show_cities_multiple()").execute()
 
 ## Logical Decode
 
-> This feature not implemented yet
-
-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());
-});
-```
+GaussDB allows replication streaming and decoding persistent changes to a database's tables into useful chunks of data.
+In GaussDB, 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.
+ 
+see https://github.com/HuaweiCloudDeveloper/gaussdb-r2dbc-examples for Logical Decode examples.
 
 ## GaussDB Enum Types
 
-Applications may make use of Postgres enumerated types by using `EnumCodec` to map custom types to Java `enum` types. 
-`EnumCodec` requires the Postgres OID and the Java to map enum values to the Postgres protocol and to materialize Enum instances from Postgres results. 
+Applications may make use of GaussDB enumerated types by using `EnumCodec` to map custom types to Java `enum` types. 
+`EnumCodec` requires the GaussDB OID and the Java to map enum values to the GaussDB protocol and to materialize Enum instances 
+from GaussDB results. 
 You can configure a `CodecRegistrar` through `EnumCodec.builder()` for one or more enumeration type mappings. Make sure to use different Java enum types otherwise the driver is not able to distinguish between Postgres OIDs. 
 
 Example:
@@ -406,7 +373,7 @@ enum MyEnumType {
 **Codec Registration:**
 
 ```java
-PostgresqlConnectionConfiguration.builder()
+GaussDBConnectionConfiguration.builder()
         .codecRegistrar(EnumCodec.builder().withEnum("my_enum",MyEnumType.class).build());
 ```
 

pom.xml

@@ -264,7 +264,6 @@
                 <version>3.12.1</version>
                 <configuration>
                     <compilerArgs>
-                        <arg>-Werror</arg>
                         <arg>-Xlint:all</arg>
                         <arg>-Xlint:-deprecation</arg>
                         <arg>-Xlint:-options</arg>

src/main/java/io/r2dbc/gaussdb/GaussDBReplicationStream.java

@@ -141,10 +141,11 @@ private boolean processKeepAliveMessage(ByteBuf buffer) {
             this.lastReceiveLSN = this.lastServerLSN;
         }
 
-        long lastServerClock = buffer.readLong();
-        boolean replyRequired = buffer.readByte() != 0;
+        buffer.readInt(); // serverMode
+        buffer.readInt(); // dbState
+        buffer.readLong(); // lastServerClock
 
-        return replyRequired;
+        return  buffer.readByte() != 0;
     }
 
     private void processXLogData(ByteBuf buffer) {

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

@@ -55,9 +55,13 @@ public ByteBuf encode(ByteBufAllocator allocator) {
         ByteBuf out = allocator.buffer(34);
 
         out.writeByte(KEEP_ALIVE_REPLY);
+        out.writeLong(Long.MAX_VALUE);
         out.writeLong(this.received.asLong());
         out.writeLong(this.flushed.asLong());
+        out.writeLong(Long.MAX_VALUE);
         out.writeLong(this.applied.asLong());
+        out.writeLong(Long.MAX_VALUE);
+        out.writeLong(Long.MAX_VALUE);
         out.writeLong(this.systemClock);
 
         if (this.replyRequired) {
@@ -66,6 +70,11 @@ public ByteBuf encode(ByteBufAllocator allocator) {
             out.writeByte(this.received == LogSequenceNumber.INVALID_LSN ? (byte) REPLY_REQUIRED : (byte) NO_REPLY_REQUIRED);
         }
 
+        out.writeInt(0);
+        out.writeByte(1);
+        out.writeByte(1);
+        out.writeByte(1);
+
         return out;
     }
 

src/main/java/io/r2dbc/gaussdb/replication/ReplicationSlotRequest.java

@@ -28,12 +28,9 @@ public abstract class ReplicationSlotRequest {
 
     private final String slotName;
 
-    private final boolean temporary;
-
-    ReplicationSlotRequest(ReplicationType replicationType, String slotName, boolean temporary) {
+    ReplicationSlotRequest(ReplicationType replicationType, String slotName) {
         this.replicationType = Assert.requireNonNull(replicationType, "replicationType must not be null");
         this.slotName = Assert.requireNotEmpty(slotName, "slotName must not be null");
-        this.temporary = temporary;
     }
 
     /**
@@ -79,33 +76,23 @@ String getSlotName() {
         return this.slotName;
     }
 
-    /**
-     * Returns the slot is temporary.
-     *
-     * @return {@code true} if the slot should be temporary
-     */
-    boolean isTemporary() {
-        return this.temporary;
-    }
-
     /**
      * Slot creation request for logical replication.
      */
     static class LogicalReplicationSlotRequest extends ReplicationSlotRequest {
 
         private final String outputPlugin;
 
-        public LogicalReplicationSlotRequest(String slotName, boolean temporaryOption, String outputPlugin) {
-            super(ReplicationType.LOGICAL, slotName, temporaryOption);
+        public LogicalReplicationSlotRequest(String slotName, String outputPlugin) {
+            super(ReplicationType.LOGICAL, slotName);
             this.outputPlugin = outputPlugin;
         }
 
         @Override
         public String asSQL() {
             return String.format(
-                "CREATE_REPLICATION_SLOT %s %s LOGICAL %s",
+                "CREATE_REPLICATION_SLOT %s LOGICAL %s",
                 getSlotName(),
-                isTemporary() ? "TEMPORARY" : "",
                 this.outputPlugin
             );
         }
@@ -115,7 +102,6 @@ public String toString() {
             return "LogicalReplicationSlotRequest{" +
                 "slotName='" + getSlotName() + '\'' +
                 ", outputPlugin='" + this.outputPlugin + '\'' +
-                ", temporaryOption=" + isTemporary() +
                 '}';
         }
 
@@ -127,23 +113,21 @@ public String toString() {
     static class PhysicalReplicationSlotRequest extends ReplicationSlotRequest {
 
         public PhysicalReplicationSlotRequest(String slotName, boolean temporaryOption) {
-            super(ReplicationType.PHYSICAL, slotName, temporaryOption);
+            super(ReplicationType.PHYSICAL, slotName);
         }
 
         @Override
         public String asSQL() {
             return String.format(
-                "CREATE_REPLICATION_SLOT %s %s PHYSICAL",
-                getSlotName(),
-                isTemporary() ? "TEMPORARY" : ""
+                "CREATE_REPLICATION_SLOT %s PHYSICAL",
+                getSlotName()
             );
         }
 
         @Override
         public String toString() {
             return "PhysicalReplicationSlotRequest{" +
                 "slotName='" + getSlotName() + '\'' +
-                ", temporaryOption=" + isTemporary() +
                 '}';
         }
 
@@ -155,8 +139,6 @@ final static class DefaultLogicalSlotRequestBuilder implements LogicalSlotReques
 
         private String outputPlugin;
 
-        private boolean temporary;
-
         @Override
         public LogicalSlotRequestBuilder slotName(String slotName) {
             this.slotName = Assert.requireNotEmpty(slotName, "slotName must not be null and not empty");
@@ -171,13 +153,12 @@ public LogicalSlotRequestBuilder outputPlugin(String outputPlugin) {
 
         @Override
         public LogicalSlotRequestBuilder temporary() {
-            this.temporary = true;
             return this;
         }
 
         @Override
         public ReplicationSlotRequest build() {
-            return new LogicalReplicationSlotRequest(this.slotName, this.temporary, this.outputPlugin);
+            return new LogicalReplicationSlotRequest(this.slotName, this.outputPlugin);
         }
 
     }

src/test/java/io/r2dbc/gaussdb/replication/ReplicationSlotRequestUnitTests.java

@@ -45,17 +45,15 @@ void buildsLogicalRequest() {
         ReplicationSlotRequest request = ReplicationSlotRequest.logical().slotName("slot").outputPlugin("output").temporary().build();
         assertThat(request.getReplicationType()).isEqualTo(ReplicationType.LOGICAL);
         assertThat(request.getSlotName()).isEqualTo("slot");
-        assertThat(request.isTemporary()).isTrue();
-        assertThat(request.asSQL()).isEqualTo("CREATE_REPLICATION_SLOT slot TEMPORARY LOGICAL output");
+        assertThat(request.asSQL()).isEqualTo("CREATE_REPLICATION_SLOT slot LOGICAL output");
     }
 
     @Test
     void buildsPhysicalRequest() {
         ReplicationSlotRequest request = ReplicationSlotRequest.physical().slotName("slot").temporary().build();
         assertThat(request.getReplicationType()).isEqualTo(ReplicationType.PHYSICAL);
         assertThat(request.getSlotName()).isEqualTo("slot");
-        assertThat(request.isTemporary()).isTrue();
-        assertThat(request.asSQL()).isEqualTo("CREATE_REPLICATION_SLOT slot TEMPORARY PHYSICAL");
+        assertThat(request.asSQL()).isEqualTo("CREATE_REPLICATION_SLOT slot PHYSICAL");
     }
 
 }