Align LocalDate/LocalDateTime conversion to configured time zone.

Local temporals are converted into their local representation using the session timezone.

[closes #521]

Signed-off-by: Mark Paluch <[email protected]>

Author: mp911de

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

@@ -34,11 +34,13 @@
 import reactor.core.publisher.Mono;
 import reactor.util.annotation.Nullable;
 
+import java.time.ZoneId;
 import java.util.ArrayList;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
+import java.util.TimeZone;
 
 /**
  * An implementation of {@link ConnectionFactory} for creating connections to a PostgreSQL database.
@@ -115,17 +117,20 @@ public Mono<io.r2dbc.postgresql.api.PostgresqlReplicationConnection> replication
 
         ConnectionSettings connectionSettings = this.configuration.getConnectionSettings().mutate(builder -> builder.startupOptions(options));
 
-        ConnectionStrategy connectionStrategy = ConnectionStrategyFactory.getConnectionStrategy(connectionFunction, this.configuration, connectionSettings);
+        ConnectionStrategy connectionStrategy = ConnectionStrategyFactory.getConnectionStrategy(this.connectionFunction, this.configuration, connectionSettings);
 
         return doCreateConnection(true, connectionStrategy).map(DefaultPostgresqlReplicationConnection::new);
     }
 
     private Mono<PostgresqlConnection> doCreateConnection(boolean forReplication, ConnectionStrategy connectionStrategy) {
 
+        ZoneId defaultZone = TimeZone.getDefault().toZoneId();
+
         return connectionStrategy.connect()
             .flatMap(client -> {
 
-                DefaultCodecs codecs = new DefaultCodecs(client.getByteBufAllocator(), this.configuration.isPreferAttachedBuffers());
+                DefaultCodecs codecs = new DefaultCodecs(client.getByteBufAllocator(), this.configuration.isPreferAttachedBuffers(),
+                    () -> client.getTimeZone().map(TimeZone::toZoneId).orElse(defaultZone));
                 StatementCache statementCache = StatementCache.fromPreparedStatementCacheQueries(client, this.configuration.getPreparedStatementCacheQueries());
 
                 // early connection object to retrieve initialization details

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

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2022 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 java.time.ZoneId;
+
+/**
+ * Configuration object for codecs providing various details required for encoding and decoding data.
+ *
+ * @since 1.0
+ */
+public interface CodecConfiguration {
+
+    /**
+     * Return the zone identifier used for temporal conversions from local to zoned temporals.
+     *
+     * @return the zone identifier
+     */
+    ZoneId getZoneId();
+
+}

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

@@ -26,16 +26,19 @@
 import java.time.LocalDateTime;
 import java.time.ZoneId;
 import java.util.Date;
+import java.util.function.Supplier;
 
 final class DateCodec extends AbstractCodec<Date> implements ArrayCodecDelegate<Date> {
 
     private final LocalDateTimeCodec delegate;
 
-    DateCodec(ByteBufAllocator byteBufAllocator) {
+    private final Supplier<ZoneId> zoneIdSupplier;
+
+    DateCodec(ByteBufAllocator byteBufAllocator, Supplier<ZoneId> zoneIdSupplier) {
         super(Date.class);
 
-        Assert.requireNonNull(byteBufAllocator, "byteBufAllocator must not be null");
-        this.delegate = new LocalDateTimeCodec(byteBufAllocator);
+        this.delegate = new LocalDateTimeCodec(byteBufAllocator, zoneIdSupplier);
+        this.zoneIdSupplier = Assert.requireNonNull(zoneIdSupplier, "zoneIdSupplier must not be null");
     }
 
     @Override
@@ -61,7 +64,7 @@ Date doDecode(ByteBuf buffer, PostgresTypeIdentifier dataType, @Nullable Format
         Assert.requireNonNull(buffer, "byteBuf must not be null");
 
         LocalDateTime intermediary = this.delegate.doDecode(buffer, dataType, format, LocalDateTime.class);
-        return Date.from(intermediary.atZone(ZoneId.systemDefault()).toInstant());
+        return Date.from(intermediary.atZone(this.zoneIdSupplier.get()).toInstant());
     }
 
     @Override
@@ -90,8 +93,8 @@ public PostgresTypeIdentifier getArrayDataType() {
         return this.delegate.getArrayDataType();
     }
 
-    private static LocalDateTime normalize(Date value) {
-        return value.toInstant().atZone(ZoneId.systemDefault()).toLocalDateTime();
+    private LocalDateTime normalize(Date value) {
+        return value.toInstant().atZone(this.zoneIdSupplier.get()).toLocalDateTime();
     }
 
 }

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

@@ -31,6 +31,7 @@
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
+import java.util.TimeZone;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.function.Function;
 
@@ -63,33 +64,46 @@ public DefaultCodecs(ByteBufAllocator byteBufAllocator) {
      * @param preferAttachedBuffers whether to prefer attached (pooled) {@link ByteBuf buffers}. Use {@code false} (default) to use detached buffers which minimize the risk of memory leaks.
      */
     public DefaultCodecs(ByteBufAllocator byteBufAllocator, boolean preferAttachedBuffers) {
-        this(byteBufAllocator, preferAttachedBuffers, CachedCodecLookup::new);
+        this(byteBufAllocator, preferAttachedBuffers, () -> TimeZone.getDefault().toZoneId());
     }
 
     /**
      * Create a new instance of {@link DefaultCodecs}.
      *
      * @param byteBufAllocator      the {@link ByteBufAllocator} to use for encoding
      * @param preferAttachedBuffers whether to prefer attached (pooled) {@link ByteBuf buffers}. Use {@code false} (default) to use detached buffers which minimize the risk of memory leaks.
+     * @param configuration         the {@link CodecConfiguration} to use for encoding/decoding
+     */
+    public DefaultCodecs(ByteBufAllocator byteBufAllocator, boolean preferAttachedBuffers, CodecConfiguration configuration) {
+        this(byteBufAllocator, preferAttachedBuffers, configuration, CachedCodecLookup::new);
+    }
+
+    /**
+     * Create a new instance of {@link DefaultCodecs}.
+     *
+     * @param byteBufAllocator      the {@link ByteBufAllocator} to use for encoding
+     * @param configuration         the {@link CodecConfiguration} to use for encoding/decoding
+     * @param preferAttachedBuffers whether to prefer attached (pooled) {@link ByteBuf buffers}. Use {@code false} (default) to use detached buffers which minimize the risk of memory leaks.
      * @param codecLookupFunction   provides the {@link CodecLookup} to use for finding relevant codecs
      */
-    DefaultCodecs(ByteBufAllocator byteBufAllocator, boolean preferAttachedBuffers, Function<CodecRegistry, CodecLookup> codecLookupFunction) {
+    DefaultCodecs(ByteBufAllocator byteBufAllocator, boolean preferAttachedBuffers, CodecConfiguration configuration, Function<CodecRegistry, CodecLookup> codecLookupFunction) {
         Assert.requireNonNull(byteBufAllocator, "byteBufAllocator must not be null");
+        Assert.requireNonNull(configuration, "configuration must not be null");
         Assert.requireNonNull(codecLookupFunction, "codecLookupFunction must not be null");
 
         this.codecLookup = codecLookupFunction.apply(this);
-        this.codecs = getDefaultCodecs(byteBufAllocator, preferAttachedBuffers);
+        this.codecs = getDefaultCodecs(byteBufAllocator, preferAttachedBuffers, configuration);
         this.codecLookup.afterCodecAdded();
     }
 
     @SuppressWarnings({"unchecked", "rawtypes"})
-    private static List<Codec<?>> getDefaultCodecs(ByteBufAllocator byteBufAllocator, boolean preferAttachedBuffers) {
+    private static List<Codec<?>> getDefaultCodecs(ByteBufAllocator byteBufAllocator, boolean preferAttachedBuffers, CodecConfiguration configuration) {
 
         List<Codec<?>> codecs = new CopyOnWriteArrayList<>(Arrays.asList(
 
             // Prioritized Codecs
             new StringCodec(byteBufAllocator),
-            new InstantCodec(byteBufAllocator),
+            new InstantCodec(byteBufAllocator, configuration::getZoneId),
             new ZonedDateTimeCodec(byteBufAllocator),
             new BinaryByteBufferCodec(byteBufAllocator),
             new BinaryByteArrayCodec(byteBufAllocator),
@@ -104,7 +118,7 @@ private static List<Codec<?>> getDefaultCodecs(ByteBufAllocator byteBufAllocator
             new IntegerCodec(byteBufAllocator),
             new IntervalCodec(byteBufAllocator),
             new LocalDateCodec(byteBufAllocator),
-            new LocalDateTimeCodec(byteBufAllocator),
+            new LocalDateTimeCodec(byteBufAllocator, configuration::getZoneId),
             new LocalTimeCodec(byteBufAllocator),
             new LongCodec(byteBufAllocator),
             new OffsetDateTimeCodec(byteBufAllocator),
@@ -125,7 +139,7 @@ private static List<Codec<?>> getDefaultCodecs(ByteBufAllocator byteBufAllocator
 
             // Fallback for Object.class
             new ByteCodec(byteBufAllocator),
-            new DateCodec(byteBufAllocator),
+            new DateCodec(byteBufAllocator, configuration::getZoneId),
 
             new BlobCodec(byteBufAllocator),
             new ClobCodec(byteBufAllocator),

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

@@ -27,14 +27,18 @@
 import java.time.LocalDateTime;
 import java.time.ZoneId;
 import java.time.ZoneOffset;
+import java.util.function.Supplier;
 
 import static io.r2dbc.postgresql.codec.PostgresqlObjectId.TIMESTAMPTZ;
 import static io.r2dbc.postgresql.codec.PostgresqlObjectId.TIMESTAMPTZ_ARRAY;
 
 final class InstantCodec extends AbstractTemporalCodec<Instant> {
 
-    InstantCodec(ByteBufAllocator byteBufAllocator) {
+    private final Supplier<ZoneId> zoneIdSupplier;
+
+    InstantCodec(ByteBufAllocator byteBufAllocator, Supplier<ZoneId> zoneIdSupplier) {
         super(Instant.class, byteBufAllocator, TIMESTAMPTZ, TIMESTAMPTZ_ARRAY, Instant::toString);
+        this.zoneIdSupplier = zoneIdSupplier;
     }
 
     @Override
@@ -44,11 +48,15 @@ Instant doDecode(ByteBuf buffer, PostgresTypeIdentifier dataType, @Nullable Form
         return decodeTemporal(buffer, dataType, format, Instant.class, temporal -> {
 
             if (temporal instanceof LocalDateTime) {
-                return ((LocalDateTime) temporal).toInstant(ZoneOffset.UTC);
+                ZoneId zoneId = this.zoneIdSupplier.get();
+                ZoneOffset offset = zoneId.getRules().getOffset((LocalDateTime) temporal);
+                return ((LocalDateTime) temporal).toInstant(offset);
             }
 
             if (temporal instanceof LocalDate) {
-                return ((LocalDate) temporal).atStartOfDay(ZoneId.systemDefault()).toInstant();
+                ZoneId zoneId = this.zoneIdSupplier.get();
+
+                return ((LocalDate) temporal).atStartOfDay(zoneId).toInstant();
             }
 
             return Instant.from(temporal);

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

@@ -27,26 +27,34 @@
 import java.time.LocalDateTime;
 import java.time.ZoneId;
 import java.time.ZoneOffset;
+import java.util.function.Supplier;
 
 import static io.r2dbc.postgresql.codec.PostgresqlObjectId.TIMESTAMP;
 import static io.r2dbc.postgresql.codec.PostgresqlObjectId.TIMESTAMP_ARRAY;
 
 final class LocalDateTimeCodec extends AbstractTemporalCodec<LocalDateTime> {
 
-    LocalDateTimeCodec(ByteBufAllocator byteBufAllocator) {
+    private final Supplier<ZoneId> zoneIdSupplier;
+
+    LocalDateTimeCodec(ByteBufAllocator byteBufAllocator, Supplier<ZoneId> zoneIdSupplier) {
         super(LocalDateTime.class, byteBufAllocator, TIMESTAMP, TIMESTAMP_ARRAY, LocalDateTime::toString);
+        this.zoneIdSupplier = Assert.requireNonNull(zoneIdSupplier, "zoneIdSupplier must not be null");
     }
 
     @Override
     LocalDateTime doDecode(ByteBuf buffer, PostgresTypeIdentifier dataType, @Nullable Format format, @Nullable Class<? extends LocalDateTime> type) {
         Assert.requireNonNull(buffer, "byteBuf must not be null");
 
         return decodeTemporal(buffer, dataType, format, LocalDateTime.class, temporal -> {
+            ZoneId zone = this.zoneIdSupplier.get();
+
             if (temporal instanceof LocalDate) {
-                return ((LocalDate) temporal).atStartOfDay(ZoneId.systemDefault()).toLocalDateTime();
+                return ((LocalDate) temporal).atStartOfDay(zone).toLocalDateTime();
             }
 
-            return Instant.from(temporal).atOffset(ZoneOffset.UTC).toLocalDateTime();
+            Instant instant = Instant.from(temporal);
+            ZoneOffset offset = zone.getRules().getOffset(instant);
+            return instant.atOffset(offset).toLocalDateTime();
         });
     }
 

src/test/java/io/r2dbc/postgresql/codec/DateCodecUnitTests.java

@@ -44,9 +44,11 @@ final class DateCodecUnitTests {
 
     private static final int dataType = TIMESTAMP.getObjectId();
 
+    private final DateCodec codec = new DateCodec(TEST, ZoneId::systemDefault);
+
     @Test
     void constructorNoByteBufAllocator() {
-        assertThatIllegalArgumentException().isThrownBy(() -> new DateCodec(null))
+        assertThatIllegalArgumentException().isThrownBy(() -> new DateCodec(null, null))
             .withMessage("byteBufAllocator must not be null");
     }
 
@@ -55,26 +57,26 @@ void decode() {
         Instant testInstant = LocalDateTime.parse("2010-02-01T10:08:04.412").atZone(ZoneId.systemDefault()).toInstant();
         Date date = Date.from(testInstant);
 
-        assertThat(new DateCodec(TEST).decode(encode(TEST, "2010-02-01 10:08:04.412"), dataType, FORMAT_TEXT, Date.class))
+        assertThat(this.codec.decode(encode(TEST, "2010-02-01 10:08:04.412"), dataType, FORMAT_TEXT, Date.class))
             .isEqualTo(date);
     }
 
     @Test
     void decodeFromDate() {
         Date date = Date.from(LocalDateTime.parse("2018-11-04T00:00:00.000").atZone(ZoneId.systemDefault()).toInstant());
 
-        assertThat(new DateCodec(TEST).decode(encode(TEST, "2018-11-04"), DATE.getObjectId(), FORMAT_TEXT, Date.class))
+        assertThat(this.codec.decode(encode(TEST, "2018-11-04"), DATE.getObjectId(), FORMAT_TEXT, Date.class))
             .isEqualTo(date);
     }
 
     @Test
     void decodeNoByteBuf() {
-        assertThat(new DateCodec(TEST).decode(null, dataType, FORMAT_TEXT, Date.class)).isNull();
+        assertThat(this.codec.decode(null, dataType, FORMAT_TEXT, Date.class)).isNull();
     }
 
     @Test
     void doCanDecode() {
-        DateCodec codec = new DateCodec(TEST);
+        DateCodec codec = this.codec;
 
         assertThat(codec.doCanDecode(TIMESTAMP, FORMAT_BINARY)).isTrue();
         assertThat(codec.doCanDecode(MONEY, FORMAT_TEXT)).isFalse();
@@ -83,13 +85,13 @@ void doCanDecode() {
 
     @Test
     void doCanDecodeNoFormat() {
-        assertThatIllegalArgumentException().isThrownBy(() -> new DateCodec(TEST).doCanDecode(VARCHAR, null))
+        assertThatIllegalArgumentException().isThrownBy(() -> this.codec.doCanDecode(VARCHAR, null))
             .withMessage("format must not be null");
     }
 
     @Test
     void doCanDecodeNoType() {
-        assertThatIllegalArgumentException().isThrownBy(() -> new DateCodec(TEST).doCanDecode(null, FORMAT_TEXT))
+        assertThatIllegalArgumentException().isThrownBy(() -> this.codec.doCanDecode(null, FORMAT_TEXT))
             .withMessage("type must not be null");
     }
 
@@ -98,21 +100,21 @@ void doEncode() {
         Instant testInstant = LocalDateTime.parse("2010-02-01T10:08:04.412").atZone(ZoneId.systemDefault()).toInstant();
         Date date = Date.from(testInstant);
 
-        assertThat(new DateCodec(TEST).doEncode(date))
+        assertThat(this.codec.doEncode(date))
             .hasFormat(FORMAT_TEXT)
             .hasType(TIMESTAMP.getObjectId())
             .hasValue(encode(TEST, "2010-02-01T10:08:04.412"));
     }
 
     @Test
     void doEncodeNoValue() {
-        assertThatIllegalArgumentException().isThrownBy(() -> new DateCodec(TEST).doEncode(null))
+        assertThatIllegalArgumentException().isThrownBy(() -> this.codec.doEncode(null))
             .withMessage("value must not be null");
     }
 
     @Test
     void encodeNull() {
-        assertThat(new DateCodec(TEST).encodeNull())
+        assertThat(this.codec.encodeNull())
             .isEqualTo(new EncodedParameter(FORMAT_TEXT, TIMESTAMP.getObjectId(), NULL_VALUE));
     }