Add support for Postgres enumerated types

README.md

@@ -201,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
@@ -215,7 +216,7 @@ connection.createStatement("SELECT my_json FROM my_table")
 ```java
 connection.createStatement("INSERT INTO my_table (my_json) VALUES($1::JSON)")
     .bind("$1", "{\"hello\": \"world\"}").execute();
-```		
+```
 
 **Consume JSON as scalar type**
 
@@ -251,7 +252,6 @@ 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:
@@ -275,6 +275,35 @@ Flux<T> replicationStream = replicationConnection.startReplication(replicationRe
 });
 ```
 
+## Postgres 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. 
+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:
+
+**SQL:**
+
+```sql
+CREATE TYPE my_enum AS ENUM ('FIRST', 'SECOND');
+``` 
+
+**Java Model:**
+
+```java
+enum MyEnumType {
+  FIRST, SECOND;
+}
+```
+
+**Codec Registration:**
+
+```java
+PostgresqlConnectionConfiguration.builder()
+  .codecRegistrar(EnumCodec.builder().withEnum("my_enum", MyEnumType.class).build());
+```
+
 ## Data Type Mapping
 
 This reference table shows the type mapping between [PostgreSQL][p] and Java data types:
@@ -293,6 +322,7 @@ This reference table shows the type mapping between [PostgreSQL][p] and Java dat
 | [`circle`][psql-circle-ref]                     | Not yet supported.|
 | [`date`][psql-date-ref]                         | [`LocalDate`][java-ld-ref]|
 | [`double precision`][psql-floating-point-ref]   | [**`Double`**][java-double-ref], [`Float`][java-float-ref], [`Boolean`][java-boolean-ref], [`Byte`][java-byte-ref], [`Short`][java-short-ref], [`Integer`][java-integer-ref], [`Long`][java-long-ref], [`BigDecimal`][java-bigdecimal-ref], [`BigInteger`][java-biginteger-ref]|
+| [enumerated types][psql-enum-ref]               | Client code `Enum` types through `EnumCodec`|
 | [`hstore`][psql-hstore-ref]                     | [**`Map`**][java-map-ref]|
 | [`inet`][psql-inet-ref]                         | [**`InetAddress`**][java-inet-ref]|
 | [`integer`][psql-integer-ref]                   | [**`Integer`**][java-integer-ref], [`Boolean`][java-boolean-ref], [`Byte`][java-byte-ref], [`Short`][java-short-ref], [`Long`][java-long-ref], [`BigDecimal`][java-bigdecimal-ref], [`BigInteger`][java-biginteger-ref]|
@@ -345,6 +375,7 @@ Support for the following single-dimensional arrays (read and write):
 [psql-circle-ref]: https://www.postgresql.org/docs/current/datatype-geometric.html#DATATYPE-CIRCLE
 [psql-date-ref]: https://www.postgresql.org/docs/current/datatype-datetime.html
 [psql-floating-point-ref]: https://www.postgresql.org/docs/current/datatype-numeric.html#DATATYPE-FLOAT
+[psql-enum-ref]: https://www.postgresql.org/docs/current/datatype-enum.html
 [psql-hstore-ref]: https://www.postgresql.org/docs/current/hstore.html
 [psql-inet-ref]: https://www.postgresql.org/docs/current/datatype-net-types.html#DATATYPE-INET
 [psql-integer-ref]: https://www.postgresql.org/docs/current/datatype-numeric.html#DATATYPE-INT
@@ -408,7 +439,7 @@ This driver accepts the following extensions:
 
 Extensions can be registered programmatically using `PostgresConnectionConfiguration` or discovered using Java's `ServiceLoader` mechanism (from `META-INF/services/io.r2dbc.postgresql.extension.Extension`).
 
-The driver ships with built-in dynamic codecs (e.g. `hstore`) that are registered during the connection handshake depending on their availability while connecting. Note that Postgres extensions registered after a connection was established require a reconnect to initialize the codec.
+The driver ships with built-in dynamic codecs (e.g. `hstore`) that are registered during the connection handshake depending on their availability while connecting. Note that Postgres extensions registered after a connection was established require a reconnect to initialize the codec. 
 
 ## Logging
 If SL4J is on the classpath, it will be used. Otherwise, there are two possible fallbacks: Console or `java.util.logging.Logger`). By default, the Console fallback is used. To use the JDK loggers, set the `reactor.logging.fallback` System property to `JDK`.

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

@@ -370,6 +370,7 @@ public Builder connectTimeout(@Nullable Duration connectTimeout) {
 
         /**
          * Registers a {@link CodecRegistrar} that can contribute extension {@link Codec}s.
+         * Calling this method adds a {@link CodecRegistrar} and does not replace existing {@link Extension}s.
          *
          * @param codecRegistrar registrar to contribute codecs
          * @return this {@link Builder}
@@ -400,6 +401,7 @@ public Builder enableSsl() {
 
         /**
          * Registers a {@link Extension} to extend driver functionality.
+         * Calling this method adds a {@link Extension} and does not replace existing {@link Extension}s.
          *
          * @param extension extension to extend driver functionality
          * @return this {@link Builder}

src/main/java/io/r2dbc/postgresql/codec/EnumCodec.java

@@ -0,0 +1,184 @@
+/*
+ * 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.codec;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+import io.r2dbc.postgresql.client.Parameter;
+import io.r2dbc.postgresql.extension.CodecRegistrar;
+import io.r2dbc.postgresql.message.Format;
+import io.r2dbc.postgresql.util.Assert;
+import io.r2dbc.postgresql.util.ByteBufUtils;
+import reactor.core.publisher.Mono;
+import reactor.util.Logger;
+import reactor.util.Loggers;
+import reactor.util.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import static io.r2dbc.postgresql.client.Parameter.NULL_VALUE;
+import static io.r2dbc.postgresql.message.Format.FORMAT_TEXT;
+
+/**
+ * Codec to map Postgres {@code enumerated} types to Java {@link Enum} values.
+ * This codec uses {@link Enum#name()} to map Postgres enum values as these are represented as string values.
+ * <p>Note that enum values are case-sensitive.
+ *
+ * @param <T> enum type
+ * @since 0.8.4
+ */
+public final class EnumCodec<T extends Enum<T>> implements Codec<T> {
+
+    private static final Logger logger = Loggers.getLogger(EnumCodec.class);
+
+    private final ByteBufAllocator byteBufAllocator;
+
+    private final Class<T> type;
+
+    private final int oid;
+
+    public EnumCodec(ByteBufAllocator byteBufAllocator, Class<T> type, int oid) {
+        this.byteBufAllocator = Assert.requireNonNull(byteBufAllocator, "byteBufAllocator must not be null");
+        this.type = Assert.requireNonNull(type, "type must not be null");
+        this.oid = oid;
+    }
+
+    @Override
+    public boolean canDecode(int dataType, Format format, Class<?> type) {
+        Assert.requireNonNull(type, "type must not be null");
+        return this.type.equals(type) && dataType == this.oid;
+    }
+
+    @Override
+    public boolean canEncode(Object value) {
+        Assert.requireNonNull(value, "value must not be null");
+        return this.type.isInstance(value);
+    }
+
+    @Override
+    public boolean canEncodeNull(Class<?> type) {
+        Assert.requireNonNull(type, "type must not be null");
+        return this.type.equals(type);
+    }
+
+    @Override
+    public T decode(@Nullable ByteBuf buffer, int dataType, Format format, Class<? extends T> type) {
+        if (buffer == null) {
+            return null;
+        }
+
+        return Enum.valueOf(this.type, ByteBufUtils.decode(buffer));
+    }
+
+    @Override
+    public Parameter encode(Object value) {
+
+        Assert.requireNonNull(value, "value must not be null");
+
+        return new Parameter(FORMAT_TEXT, this.oid, Mono.fromSupplier(() -> ByteBufUtils.encode(this.byteBufAllocator, this.type.cast(value).name())));
+    }
+
+    @Override
+    public Parameter encodeNull() {
+        return new Parameter(Format.FORMAT_BINARY, this.oid, NULL_VALUE);
+    }
+
+    @Override
+    public Class<?> type() {
+        return this.type;
+    }
+
+    /**
+     * Create a new {@link Builder} to build a {@link CodecRegistrar} to dynamically register Postgres {@code enum} types to {@link Enum} values.
+     *
+     * @return a new builder.
+     */
+    public static EnumCodec.Builder builder() {
+        return new Builder();
+    }
+
+    /**
+     * Builder for {@link CodecRegistrar} to register {@link EnumCodec} for one or more enum type mappings.
+     */
+    public static final class Builder {
+
+        private final Map<String, Class<? extends Enum<?>>> mapping = new LinkedHashMap<>();
+
+        /**
+         * Add a Postgres enum type to {@link Enum} mapping.
+         *
+         * @param name      name of the Postgres enum type
+         * @param enumClass the corresponding Java type
+         * @return this {@link Builder}
+         */
+        public Builder withEnum(String name, Class<? extends Enum<?>> enumClass) {
+            Assert.requireNonNull(enumClass, "Enum class must not be null");
+            Assert.isTrue(enumClass.isEnum(), String.format("Enum class %s must be an enum type", enumClass.getName()));
+
+            if (this.mapping.containsKey(name)) {
+                throw new IllegalArgumentException(String.format("Builder contains already a mapping for Postgres type %s", name));
+            }
+
+            if (this.mapping.containsValue(enumClass)) {
+                throw new IllegalArgumentException(String.format("Builder contains already a mapping for Java type %s", enumClass.getName()));
+            }
+
+            this.mapping.put(Assert.requireNotEmpty(name, "Postgres type name must not be null"), enumClass);
+            return this;
+        }
+
+        /**
+         * Build a {@link CodecRegistrar} to be used with {@code PostgresqlConnectionConfiguration.Builder#codecRegistrar(CodecRegistrar)}.
+         * The codec registrar registers the codes to be used as part of the connection setup.
+         *
+         * @return a new {@link CodecRegistrar}.
+         */
+        @SuppressWarnings({"unchecked", "rawtypes"})
+        public CodecRegistrar build() {
+
+            Map<String, Class<? extends Enum<?>>> mapping = new LinkedHashMap<>(this.mapping);
+
+            return (connection, allocator, registry) -> {
+
+                List<String> missing = new ArrayList<>(mapping.keySet());
+                return PostgresTypes.from(connection).lookupTypes(mapping.keySet())
+                    .filter(PostgresTypes.PostgresType::isEnum)
+                    .doOnNext(it -> {
+
+                        Class<? extends Enum<?>> enumClass = mapping.get(it.getName());
+                        if (enumClass == null) {
+                            logger.warn(String.format("Cannot find Java type for enum type '%s' with oid %d. Known types are: %s", it.getName(), it.getOid(), mapping));
+                            return;
+                        }
+
+                        missing.remove(it.getName());
+                        logger.debug(String.format("Registering codec for type '%s' with oid %d using Java enum type '%s'", it.getName(), it.getOid(), enumClass.getName()));
+                        registry.addLast(new EnumCodec(allocator, enumClass, it.getOid()));
+                    }).doOnComplete(() -> {
+
+                        if (!missing.isEmpty()) {
+                            logger.warn(String.format("Could not lookup enum types for: %s", missing));
+                        }
+
+                    }).then();
+            };
+        }
+    }
+}