Use PostgresTypeIdentifier instead of PostgresqlObjectId to accept dynamic types in codecs

[resolves pgjdbc#378]

Author: mp911de

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

@@ -144,7 +144,7 @@ final EncodedParameter doEncode(Object[] value) {
     }
 
     @Override
-    EncodedParameter doEncode(Object[] value, PostgresqlObjectId dataType) {
+    EncodedParameter doEncode(Object[] value, PostgresTypeIdentifier dataType) {
         Assert.requireNonNull(value, "value must not be null");
 
         return encodeArray(() -> {
@@ -161,7 +161,7 @@ EncodedParameter doEncode(Object[] value, PostgresqlObjectId dataType) {
      * @param dataType        the Postgres data type
      * @return encoded {@link EncodedParameter} item
      */
-    abstract EncodedParameter encodeArray(Supplier<ByteBuf> encodedSupplier, PostgresqlObjectId dataType);
+    abstract EncodedParameter encodeArray(Supplier<ByteBuf> encodedSupplier, PostgresTypeIdentifier dataType);
 
     /**
      * Encode a single array item using text format.

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

@@ -94,7 +94,7 @@ public final EncodedParameter encode(Object value) {
     public EncodedParameter encode(Object value, int dataType) {
         Assert.requireNonNull(value, "value must not be null");
 
-        return doEncode((T) value, PostgresqlObjectId.valueOf(dataType));
+        return doEncode((T) value, PostgresqlObjectId.isValid(dataType) ? PostgresqlObjectId.valueOf(dataType) : new SimplePostgresTypeIdentifier(dataType));
     }
 
     public EncodedParameter encodeNull(int dataType) {
@@ -116,7 +116,7 @@ public Class<?> type() {
      * @implNote use deferred buffer creation instead of {@link Mono#just(Object)} and {@link Flux#just(Object)} to avoid memory
      * leaks
      */
-    static EncodedParameter create(Format format, PostgresqlObjectId type, Publisher<? extends ByteBuf> value) {
+    static EncodedParameter create(Format format, PostgresTypeIdentifier type, Publisher<? extends ByteBuf> value) {
         Assert.requireNonNull(type, "type must not be null");
         return new EncodedParameter(format, type.getObjectId(), value);
     }
@@ -125,40 +125,52 @@ static EncodedParameter create(Format format, PostgresqlObjectId type, Publisher
      * Create a {@link EncodedParameter}.
      *
      * @param format         the format to use
-     * @param type           the well-known {@link PostgresqlObjectId type OID}
+     * @param type           the well-known {@link PostgresTypeIdentifier type OID}
      * @param bufferSupplier {@link Supplier} supplying the encoded {@link ByteBuf buffer}
      * @return the encoded  {@link EncodedParameter}
      */
-    static EncodedParameter create(Format format, PostgresqlObjectId type, Supplier<? extends ByteBuf> bufferSupplier) {
+    static EncodedParameter create(Format format, PostgresTypeIdentifier type, Supplier<? extends ByteBuf> bufferSupplier) {
         Assert.requireNonNull(type, "type must not be null");
-        return new EncodedParameter(format, type.getObjectId(), Mono.fromSupplier(bufferSupplier));
+        return create(format, type.getObjectId(), bufferSupplier);
+    }
+
+    /**
+     * Create a {@link EncodedParameter}.
+     *
+     * @param format         the format to use
+     * @param type           the well-known type OID
+     * @param bufferSupplier {@link Supplier} supplying the encoded {@link ByteBuf buffer}
+     * @return the encoded  {@link EncodedParameter}
+     */
+    static EncodedParameter create(Format format, int type, Supplier<? extends ByteBuf> bufferSupplier) {
+        return new EncodedParameter(format, type, Mono.fromSupplier(bufferSupplier));
     }
 
     /**
      * Encode a {@code null} value.
      *
      * @param format the data type {@link Format}, text or binary
-     * @param type   the well-known {@link PostgresqlObjectId type OID}
+     * @param type   the well-known {@link PostgresTypeIdentifier type OID}
      * @return the encoded {@code null} value
      */
-    static EncodedParameter createNull(Format format, PostgresqlObjectId type) {
+    static EncodedParameter createNull(Format format, PostgresTypeIdentifier type) {
         return create(format, type, NULL_VALUE);
     }
 
     /**
-     * Determine whether this {@link Codec} is capable of decoding column values based on the given {@link Format} and {@link PostgresqlObjectId}.
+     * Determine whether this {@link Codec} is capable of decoding column values based on the given {@link Format} and {@link PostgresTypeIdentifier}.
      *
-     * @param type   the well-known {@link PostgresqlObjectId type OID}
+     * @param type   the well-known {@link PostgresTypeIdentifier type OID}
      * @param format the data type {@link Format}, text or binary
-     * @return {@code true} if this codec is able to decode values of {@link Format} and {@link PostgresqlObjectId}
+     * @return {@code true} if this codec is able to decode values of {@link Format} and {@link PostgresTypeIdentifier}
      */
     abstract boolean doCanDecode(PostgresqlObjectId type, Format format);
 
     /**
      * Decode the {@link ByteBuf data} into the {@link Class value type}.
      *
      * @param buffer   the data buffer
-     * @param dataType the well-known {@link PostgresqlObjectId type OID}
+     * @param dataType the well-known {@link PostgresTypeIdentifier type OID}
      * @param format   data type format
      * @param type     the desired value type
      * @return the decoded value, can be {@code null} if the column value is {@code null}
@@ -177,12 +189,27 @@ static EncodedParameter createNull(Format format, PostgresqlObjectId type) {
      * @return the encoded value
      * @since 0.9
      */
-    abstract EncodedParameter doEncode(T value, PostgresqlObjectId dataType);
+    abstract EncodedParameter doEncode(T value, PostgresTypeIdentifier dataType);
 
     boolean isTypeAssignable(Class<?> type) {
         Assert.requireNonNull(type, "type must not be null");
 
         return type.isAssignableFrom(this.type);
     }
 
+    static class SimplePostgresTypeIdentifier implements PostgresTypeIdentifier {
+
+        private final int oid;
+
+        public SimplePostgresTypeIdentifier(int oid) {
+            this.oid = oid;
+        }
+
+        @Override
+        public int getObjectId() {
+            return this.oid;
+        }
+
+    }
+
 }

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

@@ -88,7 +88,7 @@ EncodedParameter doEncode(T value) {
     }
 
     @Override
-    EncodedParameter doEncode(T value, PostgresqlObjectId dataType) {
+    EncodedParameter doEncode(T value, PostgresTypeIdentifier dataType) {
         Assert.requireNonNull(value, "value must not be null");
 
         return create(Format.FORMAT_BINARY, dataType, () -> doEncodeBinary(value));

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

@@ -81,7 +81,7 @@ EncodedParameter doEncode(T value) {
         return doEncode(value, getDefaultType());
     }
 
-    EncodedParameter doEncode(T value, PostgresqlObjectId dataType) {
+    EncodedParameter doEncode(T value, PostgresTypeIdentifier dataType) {
         Assert.requireNonNull(value, "value must not be null");
 
         if (dataType == NUMERIC) {
@@ -91,9 +91,11 @@ EncodedParameter doEncode(T value, PostgresqlObjectId dataType) {
         return create(FORMAT_BINARY, dataType, () -> doEncodeNumber(value, dataType));
     }
 
-    private ByteBuf doEncodeNumber(Number value, PostgresqlObjectId dataType) {
+    private ByteBuf doEncodeNumber(Number value, PostgresTypeIdentifier identifier) {
 
-        switch (dataType) {
+        PostgresqlObjectId oid = PostgresqlObjectId.valueOf(identifier.getObjectId());
+
+        switch (oid) {
 
             case FLOAT8:
                 return this.byteBufAllocator.buffer(8).writeDouble(value.doubleValue());
@@ -111,7 +113,7 @@ private ByteBuf doEncodeNumber(Number value, PostgresqlObjectId dataType) {
                 return this.byteBufAllocator.buffer(4).writeShort(value.shortValue());
         }
 
-        throw new IllegalArgumentException(String.format("Cannot encode %s to %s", value, dataType));
+        throw new IllegalArgumentException(String.format("Cannot encode %s to %s", value, oid));
 
     }
 

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

@@ -51,7 +51,7 @@ String doEncodeText(BigDecimal value) {
     }
 
     @Override
-    EncodedParameter encodeArray(Supplier<ByteBuf> encodedSupplier, PostgresqlObjectId dataType) {
+    EncodedParameter encodeArray(Supplier<ByteBuf> encodedSupplier, PostgresTypeIdentifier dataType) {
         return create(Format.FORMAT_TEXT, dataType, encodedSupplier);
     }
 

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

@@ -46,7 +46,7 @@ BigDecimal doDecode(ByteBuf buffer, PostgresqlObjectId dataType, @Nullable Forma
     }
 
     @Override
-    EncodedParameter doEncode(BigDecimal value, PostgresqlObjectId dataType) {
+    EncodedParameter doEncode(BigDecimal value, PostgresTypeIdentifier dataType) {
         Assert.requireNonNull(value, "value must not be null");
 
         return create(FORMAT_TEXT, dataType, () -> ByteBufUtils.encode(this.byteBufAllocator, value.toString()));

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

@@ -37,7 +37,7 @@ byte[] doDecode(ByteBuf buffer, PostgresqlObjectId dataType, Format format, Clas
     }
 
     @Override
-    EncodedParameter doEncode(byte[] value, PostgresqlObjectId dataType) {
+    EncodedParameter doEncode(byte[] value, PostgresTypeIdentifier dataType) {
         Assert.requireNonNull(value, "value must not be null");
 
         return create(FORMAT_TEXT, dataType, () -> encodeToHex(Unpooled.wrappedBuffer(value)));

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

@@ -38,7 +38,7 @@ ByteBuffer doDecode(ByteBuf buffer, PostgresqlObjectId dataType, Format format,
     }
 
     @Override
-    EncodedParameter doEncode(ByteBuffer value, PostgresqlObjectId dataType) {
+    EncodedParameter doEncode(ByteBuffer value, PostgresTypeIdentifier dataType) {
         return create(FORMAT_TEXT, dataType, () -> encodeToHex(Unpooled.wrappedBuffer(value)));
     }
 

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

@@ -67,7 +67,7 @@ EncodedParameter doEncode(Blob value) {
     }
 
     @Override
-    EncodedParameter doEncode(Blob value, PostgresqlObjectId dataType) {
+    EncodedParameter doEncode(Blob value, PostgresTypeIdentifier dataType) {
         Assert.requireNonNull(value, "value must not be null");
 
         return create(FORMAT_TEXT, dataType,

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

@@ -72,7 +72,7 @@ EncodedParameter doEncode(Boolean value) {
     }
 
     @Override
-    EncodedParameter doEncode(Boolean value, PostgresqlObjectId dataType) {
+    EncodedParameter doEncode(Boolean value, PostgresTypeIdentifier dataType) {
         Assert.requireNonNull(value, "value must not be null");
 
         return create(FORMAT_TEXT, dataType, () -> ByteBufUtils.encode(this.byteBufAllocator, value ? "TRUE" : "FALSE"));

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

@@ -62,7 +62,7 @@ EncodedParameter doEncode(Byte value) {
     }
 
     @Override
-    EncodedParameter doEncode(Byte value, PostgresqlObjectId dataType) {
+    EncodedParameter doEncode(Byte value, PostgresTypeIdentifier dataType) {
         return this.delegate.doEncode((short) value, dataType);
     }
 

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

@@ -62,7 +62,7 @@ EncodedParameter doEncode(Character value) {
     }
 
     @Override
-    EncodedParameter doEncode(Character value, PostgresqlObjectId dataType) {
+    EncodedParameter doEncode(Character value, PostgresTypeIdentifier dataType) {
         Assert.requireNonNull(value, "value must not be null");
 
         return this.delegate.doEncode(value.toString(), dataType);

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

@@ -66,7 +66,7 @@ EncodedParameter doEncode(Clob value) {
     }
 
     @Override
-    EncodedParameter doEncode(Clob value, PostgresqlObjectId dataType) {
+    EncodedParameter doEncode(Clob value, PostgresTypeIdentifier dataType) {
         Assert.requireNonNull(value, "value must not be null");
 
         return create(FORMAT_TEXT, dataType,

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

@@ -67,7 +67,7 @@ EncodedParameter doEncode(Date value) {
     }
 
     @Override
-    EncodedParameter doEncode(Date value, PostgresqlObjectId dataType) {
+    EncodedParameter doEncode(Date value, PostgresTypeIdentifier dataType) {
         Assert.requireNonNull(value, "value must not be null");
 
         return this.delegate.doEncode(normalize(value), dataType);

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

@@ -166,7 +166,7 @@ public <T> T decode(@Nullable ByteBuf buffer, int dataType, Format format, Class
     public EncodedParameter encode(Object value) {
         Assert.requireNonNull(value, "value must not be null");
 
-        int dataType = -1;
+        PostgresTypeIdentifier dataType = null;
         Object parameterValue = value;
 
         if (value instanceof Parameter) {
@@ -181,15 +181,15 @@ public EncodedParameter encode(Object value) {
             if (parameter.getType() instanceof R2dbcTypes) {
 
                 PostgresqlObjectId targetType = PostgresqlObjectId.valueOf((R2dbcTypes) parameter.getType());
-                dataType = targetType.getObjectId();
+                dataType = targetType;
             }
 
             if (parameter.getType() instanceof PostgresTypeIdentifier) {
-                dataType = ((PostgresTypeIdentifier) parameter.getType()).getObjectId();
+                dataType = (PostgresTypeIdentifier) parameter.getType();
             }
         }
 
-        if (dataType == -1) {
+        if (dataType == null) {
 
             if (parameterValue == null) {
                 throw new IllegalArgumentException(String.format("Cannot encode null value %s using type inference", value));
@@ -203,14 +203,13 @@ public EncodedParameter encode(Object value) {
         } else {
 
             if (parameterValue == null) {
-                return new EncodedParameter(Format.FORMAT_BINARY, dataType, NULL_VALUE);
-            } else {
+                return new EncodedParameter(Format.FORMAT_BINARY, dataType.getObjectId(), NULL_VALUE);
+            }
 
-                for (Codec<?> codec : this.codecs) {
+            for (Codec<?> codec : this.codecs) {
 
-                    if (codec.canEncode(parameterValue)) {
-                        return codec.encode(parameterValue, dataType);
-                    }
+                if (codec.canEncode(parameterValue)) {
+                    return codec.encode(parameterValue, dataType.getObjectId());
                 }
             }
         }

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

@@ -83,7 +83,7 @@ EncodedParameter doEncode(InetAddress value) {
     }
 
     @Override
-    EncodedParameter doEncode(InetAddress value, PostgresqlObjectId dataType) {
+    EncodedParameter doEncode(InetAddress value, PostgresTypeIdentifier dataType) {
         Assert.requireNonNull(value, "value must not be null");
 
         return create(FORMAT_TEXT, dataType, () -> ByteBufUtils.encode(this.byteBufAllocator, value.getHostAddress()));

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

@@ -71,7 +71,7 @@ EncodedParameter doEncode(Instant value) {
     }
 
     @Override
-    EncodedParameter doEncode(Instant value, PostgresqlObjectId dataType) {
+    EncodedParameter doEncode(Instant value, PostgresTypeIdentifier dataType) {
         Assert.requireNonNull(value, "value must not be null");
 
         return create(FORMAT_TEXT, dataType, () -> ByteBufUtils.encode(this.byteBufAllocator, value.toString()));

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

@@ -43,7 +43,7 @@ Integer doDecodeText(String text) {
     }
 
     @Override
-    EncodedParameter encodeArray(Supplier<ByteBuf> encodedSupplier, PostgresqlObjectId dataType) {
+    EncodedParameter encodeArray(Supplier<ByteBuf> encodedSupplier, PostgresTypeIdentifier dataType) {
         return create(FORMAT_TEXT, dataType, encodedSupplier);
     }
 

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

@@ -54,7 +54,7 @@ EncodedParameter doEncode(Interval value) {
     }
 
     @Override
-    EncodedParameter doEncode(Interval value, PostgresqlObjectId dataType) {
+    EncodedParameter doEncode(Interval value, PostgresTypeIdentifier dataType) {
         Assert.requireNonNull(value, "value must not be null");
 
         return create(FORMAT_TEXT, dataType,

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

@@ -43,7 +43,7 @@ byte[] doDecode(ByteBuf buffer, PostgresqlObjectId dataType, Format format, Clas
     }
 
     @Override
-    EncodedParameter doEncode(byte[] value, PostgresqlObjectId dataType) {
+    EncodedParameter doEncode(byte[] value, PostgresTypeIdentifier dataType) {
         Assert.requireNonNull(value, "value must not be null");
 
         return create(FORMAT_TEXT, JSON, () -> this.byteBufAllocator.buffer(value.length).writeBytes(value));

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

@@ -46,7 +46,7 @@ EncodedParameter doEncode(ByteBuf value) {
     }
 
     @Override
-    EncodedParameter doEncode(ByteBuf value, PostgresqlObjectId dataType) {
+    EncodedParameter doEncode(ByteBuf value, PostgresTypeIdentifier dataType) {
         Assert.requireNonNull(value, "value must not be null");
 
         return create(FORMAT_BINARY, dataType, () -> {

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

@@ -46,7 +46,7 @@ ByteBuffer doDecode(ByteBuf buffer, PostgresqlObjectId dataType, Format format,
     }
 
     @Override
-    EncodedParameter doEncode(ByteBuffer value, PostgresqlObjectId dataType) {
+    EncodedParameter doEncode(ByteBuffer value, PostgresTypeIdentifier dataType) {
         Assert.requireNonNull(value, "value must not be null");
 
         return create(FORMAT_TEXT, dataType, () -> this.byteBufAllocator.buffer(value.remaining()).writeBytes(value));

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

@@ -67,7 +67,7 @@ EncodedParameter doEncode(Json value) {
     }
 
     @Override
-    EncodedParameter doEncode(Json value, PostgresqlObjectId dataType) {
+    EncodedParameter doEncode(Json value, PostgresTypeIdentifier dataType) {
         Assert.requireNonNull(value, "value must not be null");
         if (!(value instanceof Json.JsonInput || value instanceof Json.JsonOutput)) {
             throw new IllegalArgumentException("value must be JsonInput or JsonOutput");

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

@@ -51,7 +51,7 @@ EncodedParameter doEncode(InputStream value) {
     }
 
     @Override
-    EncodedParameter doEncode(InputStream value, PostgresqlObjectId dataType) {
+    EncodedParameter doEncode(InputStream value, PostgresTypeIdentifier dataType) {
         Assert.requireNonNull(value, "value must not be null");
 
         return create(FORMAT_BINARY, dataType, () -> doEncode(value, this.byteBufAllocator));

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

@@ -46,7 +46,7 @@ EncodedParameter doEncode(String value) {
     }
 
     @Override
-    EncodedParameter doEncode(String value, PostgresqlObjectId dataType) {
+    EncodedParameter doEncode(String value, PostgresTypeIdentifier dataType) {
         Assert.requireNonNull(value, "value must not be null");
 
         return create(FORMAT_TEXT, dataType, () -> ByteBufUtils.encode(this.byteBufAllocator, value));