Add support for RefCursor

mp911de · mp911de · commit 1e525fbf6494 · 2020-01-21T16:27:17.000+01:00
We now can decode a RefCursor from a Row and fetch the data that the cursor holds. RefCursor cursor = row.get("my_store_procedure", RefCursor.class); // later cursor.fetch().then(cursor.close()); [closes #173]
diff --git a/README.md b/README.md
@@ -15,6 +15,7 @@ This driver provides the following features:
 * Execution of prepared statements with bindings
 * Execution of batch statements without bindings
 * Read and write support for all data types except LOB types (e.g. `BLOB`, `CLOB`)
+* Fetching of `REFCURSOR` using `io.r2dbc.postgresql.api.RefCursor`
 * Extension points to register `Codec`s to handle additional PostgreSQL data types
 
 Next steps:
@@ -200,7 +201,8 @@ CREATE TABLE my_table (my_json JSON);
 ```java
 connection.createStatement("INSERT INTO my_table (my_json) VALUES($1)")
             .bind("$1", Json.of("{\"hello\": \"world\"}")).execute();
-```		
+```
+
 **Consume JSON**
 
 ```java
@@ -209,12 +211,13 @@ connection.createStatement("SELECT my_json FROM my_table")
             .flatMap(it -> it.map((row, rowMetadata) -> row.get("my_json", Json.class)))
             .map(Json::asString);
 ```
+
 **Write JSON using casting**
 
 ```java
 connection.createStatement("INSERT INTO my_table (my_json) VALUES($1::JSON)")
     .bind("$1", "{\"hello\": \"world\"}").execute();
-```		
+```
 
 **Consume JSON as scalar type**
 
@@ -233,6 +236,24 @@ The following types are supported for JSON exchange:
 * `String`
 * `InputStream` (must be closed after usage to avoid memory leaks)
 
+## Cursors
+
+The driver can consume cursors that were created by PL/pgSQL as `refcursor`. 
+Cursors are represented as `RefCursor` objects. Cursors obtained from `Result` can be used to fetch the cursor directly. 
+Since cursors are stateful, they must be closed once they are no longer in use.
+
+```java
+connection.createStatement("SELECT show_cities_multiple()").execute()
+    .flatMap(result -> result.map((row, rowMetadata) -> row.get(0, RefCursor.class)))
+    .flatMap(cursor -> {
+        Mono<PostgresResult> data = cursor.fetch()
+            .flatMap(…)
+            .then(rc.close());
+        return data;
+    });
+
+```
+
 ## Logical Decode
 
 PostgreSQL allows replication streaming and decoding persistent changes to a database's tables into useful chunks of data.
diff --git a/src/main/java/io/r2dbc/postgresql/ConnectionContext.java b/src/main/java/io/r2dbc/postgresql/ConnectionContext.java
@@ -21,7 +21,7 @@
 import io.r2dbc.postgresql.codec.Codecs;
 
 /**
- * @author Mark Paluch
+ * Value object capturing contextual connection resources such as {@link Client}, {@link Codecs} and the {@link PostgresqlConnection connection facade}.
  */
 final class ConnectionContext {
 
diff --git a/src/main/java/io/r2dbc/postgresql/PostgresqlRow.java b/src/main/java/io/r2dbc/postgresql/PostgresqlRow.java
@@ -90,12 +90,30 @@ private <T> T decode(int index, Class<T> type) {
         int readerIndex = data.readerIndex();
         try {
             RowDescription.Field field = this.fields.get(index);
-            return this.codecs.decode(data, field.getDataType(), field.getFormat(), type);
+
+            T decoded = this.context.getCodecs().decode(data, field.getDataType(), field.getFormat(), type);
+
+            return type.cast(postProcessResult(decoded));
+
         } finally {
             data.readerIndex(readerIndex);
         }
     }
 
+    @Nullable
+    private Object postProcessResult(@Nullable Object decoded) {
+
+        if (decoded instanceof RefCursor) {
+            return createCursor((RefCursor) decoded);
+        }
+
+        return decoded;
+    }
+
+    private AttachedRefCursor createCursor(RefCursor decoded) {
+        return new AttachedRefCursor(this.context, decoded.getCursorName());
+    }
+
     @Override
     public int hashCode() {
         return Objects.hash(this.fields);
@@ -163,4 +181,42 @@ private void requireNotReleased() {
         }
     }
 
+    /**
+     * Default {@link RefCursor} implementation that is attached to a {@link io.r2dbc.postgresql.api.PostgresqlConnection}.
+     */
+    static class AttachedRefCursor implements RefCursor {
+
+        private final ConnectionContext context;
+
+        private final String portal;
+
+        AttachedRefCursor(ConnectionContext context, String portal) {
+            this.context = Assert.requireNonNull(context, "connection must not be null");
+            this.portal = Assert.requireNotEmpty(portal, "portal must not be empty");
+        }
+
+        @Override
+        public String getCursorName() {
+            return this.portal;
+        }
+
+        @Override
+        public Mono<io.r2dbc.postgresql.api.PostgresqlResult> fetch() {
+            return Mono.fromDirect(this.context.getConnection().createStatement("FETCH ALL IN \"" + getCursorName() + "\"").execute());
+        }
+
+        @Override
+        public Mono<Void> close() {
+            return this.context.getConnection().createStatement("CLOSE \"" + getCursorName() + "\"").execute().flatMap(PostgresqlResult::getRowsUpdated).then();
+        }
+
+        @Override
+        public String toString() {
+            return "AttachedRefCursor{" +
+                "portal='" + this.portal + '\'' +
+                ", context=" + this.context +
+                '}';
+        }
+    }
+
 }
diff --git a/src/main/java/io/r2dbc/postgresql/api/RefCursor.java b/src/main/java/io/r2dbc/postgresql/api/RefCursor.java
@@ -0,0 +1,53 @@
+/*
+ * Copyright 2020 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.api;
+
+import io.r2dbc.spi.Closeable;
+import reactor.core.publisher.Mono;
+
+/**
+ * A ref cursor value object. Cursor objects can be attached to a {@link PostgresqlConnection} which allows interaction with the cursor object by {@link #fetch() fetching the cursor} and
+ * {@link #close() closing} it.
+ * Cursor objects are stateful resources. Obtaining a cursor typically keeps the cursor open on the database server. The cursor object can get invalidated if the transaction is closed. Cursor
+ * objects should be {@link #close() closed} if they are no longer required.
+ * Detached cursors that were instantiated without a connection may throw {@link UnsupportedOperationException} when attempting to fetch or close the cursor.
+ */
+public interface RefCursor extends Closeable {
+
+    /**
+     * Return the ref cursor name (portal name).
+     *
+     * @return the ref cursor name (portal name).
+     */
+    String getCursorName();
+
+    /**
+     * Fetch the contents of the cursor using {@code FETCH ALL IN}.
+     *
+     * @return the {@link PostgresqlResult} associated with the ref cursor.
+     */
+    Mono<PostgresqlResult> fetch();
+
+    /**
+     * Close the cursor.
+     *
+     * @return a {@link Mono} terminating with success the cursor was closed.
+     */
+    @Override
+    Mono<Void> close();
+
+}
diff --git a/src/main/java/io/r2dbc/postgresql/codec/DefaultCodecs.java b/src/main/java/io/r2dbc/postgresql/codec/DefaultCodecs.java
@@ -85,6 +85,8 @@ public DefaultCodecs(ByteBufAllocator byteBufAllocator) {
 
             new BlobCodec(byteBufAllocator),
             new ClobCodec(byteBufAllocator),
+            RefCursorCodec.INSTANCE,
+            RefCursorNameCodec.INSTANCE,
 
             new ShortArrayCodec(byteBufAllocator),
             new StringArrayCodec(byteBufAllocator),
diff --git a/src/main/java/io/r2dbc/postgresql/codec/RefCursorCodec.java b/src/main/java/io/r2dbc/postgresql/codec/RefCursorCodec.java
@@ -0,0 +1,96 @@
+/*
+ * Copyright 2017-2020 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.codec;
+
+import io.netty.buffer.ByteBuf;
+import io.r2dbc.postgresql.api.PostgresqlResult;
+import io.r2dbc.postgresql.api.RefCursor;
+import io.r2dbc.postgresql.client.Parameter;
+import io.r2dbc.postgresql.message.Format;
+import io.r2dbc.postgresql.type.PostgresqlObjectId;
+import io.r2dbc.postgresql.util.Assert;
+import io.r2dbc.postgresql.util.ByteBufUtils;
+import reactor.core.publisher.Mono;
+import reactor.util.annotation.Nullable;
+
+import static io.r2dbc.postgresql.type.PostgresqlObjectId.REF_CURSOR;
+
+final class RefCursorCodec extends AbstractCodec<RefCursor> {
+
+    static final RefCursorCodec INSTANCE = new RefCursorCodec();
+
+    RefCursorCodec() {
+        super(RefCursor.class);
+    }
+
+    @Override
+    public Parameter encodeNull() {
+        throw new UnsupportedOperationException("RefCursor cannot be encoded");
+    }
+
+    @Override
+    boolean doCanDecode(PostgresqlObjectId type, Format format) {
+        Assert.requireNonNull(format, "format must not be null");
+        Assert.requireNonNull(type, "type must not be null");
+
+        return REF_CURSOR == type;
+    }
+
+    @Override
+    RefCursor doDecode(ByteBuf buffer, PostgresqlObjectId dataType, @Nullable Format format, @Nullable Class<? extends RefCursor> type) {
+        Assert.requireNonNull(buffer, "byteBuf must not be null");
+
+        return new SimpleRefCursor(ByteBufUtils.decode(buffer));
+    }
+
+    @Override
+    Parameter doEncode(RefCursor value) {
+        throw new UnsupportedOperationException("RefCursor cannot be encoded");
+    }
+
+    static class SimpleRefCursor implements RefCursor {
+
+        private final String portal;
+
+        SimpleRefCursor(String portal) {
+            this.portal = portal;
+        }
+
+        @Override
+        public String getCursorName() {
+            return this.portal;
+        }
+
+        @Override
+        public Mono<PostgresqlResult> fetch() {
+            throw new UnsupportedOperationException("Stateless RefCursor does not support fetch()");
+        }
+
+        @Override
+        public Mono<Void> close() {
+            throw new UnsupportedOperationException("Stateless RefCursor does not support close()");
+        }
+
+        @Override
+        public String toString() {
+            return "SimpleRefCursor{" +
+                "portal='" + this.portal + '\'' +
+                '}';
+        }
+    }
+
+}
diff --git a/src/main/java/io/r2dbc/postgresql/codec/RefCursorNameCodec.java b/src/main/java/io/r2dbc/postgresql/codec/RefCursorNameCodec.java
@@ -0,0 +1,62 @@
+/*
+ * Copyright 2017-2020 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.codec;
+
+import io.netty.buffer.ByteBuf;
+import io.r2dbc.postgresql.client.Parameter;
+import io.r2dbc.postgresql.message.Format;
+import io.r2dbc.postgresql.type.PostgresqlObjectId;
+import io.r2dbc.postgresql.util.Assert;
+import io.r2dbc.postgresql.util.ByteBufUtils;
+import reactor.util.annotation.Nullable;
+
+import static io.r2dbc.postgresql.type.PostgresqlObjectId.REF_CURSOR;
+
+final class RefCursorNameCodec extends AbstractCodec<String> {
+
+    static final RefCursorNameCodec INSTANCE = new RefCursorNameCodec();
+
+    RefCursorNameCodec() {
+        super(String.class);
+    }
+
+    @Override
+    public Parameter encodeNull() {
+        throw new UnsupportedOperationException("Cannot encode RefCursor");
+    }
+
+    @Override
+    boolean doCanDecode(PostgresqlObjectId type, Format format) {
+        Assert.requireNonNull(format, "format must not be null");
+        Assert.requireNonNull(type, "type must not be null");
+
+        return REF_CURSOR == type;
+    }
+
+    @Override
+    String doDecode(ByteBuf buffer, PostgresqlObjectId dataType, @Nullable Format format, @Nullable Class<? extends String> type) {
+        Assert.requireNonNull(buffer, "byteBuf must not be null");
+
+        return ByteBufUtils.decode(buffer);
+    }
+
+    @Override
+    Parameter doEncode(String value) {
+        throw new UnsupportedOperationException("Cannot encode RefCursor");
+    }
+
+}
diff --git a/src/main/java/io/r2dbc/postgresql/codec/StringCodec.java b/src/main/java/io/r2dbc/postgresql/codec/StringCodec.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2019 the original author or authors.
+ * Copyright 2017-2020 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.
diff --git a/src/test/java/io/r2dbc/postgresql/AbstractIntegrationTests.java b/src/test/java/io/r2dbc/postgresql/AbstractIntegrationTests.java
@@ -31,9 +31,9 @@ public abstract class AbstractIntegrationTests {
     @RegisterExtension
     public static final PostgresqlServerExtension SERVER = new PostgresqlServerExtension();
 
-    PostgresqlConnectionFactory connectionFactory;
+    public PostgresqlConnectionFactory connectionFactory;
 
-    PostgresqlConnection connection;
+    public PostgresqlConnection connection;
 
     /**
      * Entry-point to obtain a {@link PostgresqlConnectionFactory}.
diff --git a/src/test/java/io/r2dbc/postgresql/codec/RefCursorCodecTest.java b/src/test/java/io/r2dbc/postgresql/codec/RefCursorCodecTest.java
diff --git a/src/test/java/io/r2dbc/postgresql/feature/RefCursorIntegrationTests.java b/src/test/java/io/r2dbc/postgresql/feature/RefCursorIntegrationTests.java

Original file line number	Diff line number	Diff line change
`@@ -1,5 +1,5 @@`
`1`	`1`	`/*`
`2`		`- * Copyright 2017-2019 the original author or authors.`
	`2`	`+ * Copyright 2017-2020 the original author or authors.`
`3`	`3`	`*`
`4`	`4`	`* Licensed under the Apache License, Version 2.0 (the "License");`
`5`	`5`	`* you may not use this file except in compliance with the License.`