Associate client exceptions with SQL state and ErrorDetails

We now provide error details for client-side exceptions for easier handling of failures.

[resolves #538]

Signed-off-by: Mark Paluch <mpaluch@vmware.com>

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

@@ -17,6 +17,8 @@
 package io.r2dbc.postgresql;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.r2dbc.postgresql.api.ErrorDetails;
+import io.r2dbc.postgresql.api.PostgresqlException;
 import io.r2dbc.postgresql.client.Client;
 import io.r2dbc.postgresql.client.ConnectionSettings;
 import io.r2dbc.postgresql.client.ReactorNettyClient;
@@ -218,10 +220,20 @@ private Mono<IsolationLevel> getIsolationLevel(io.r2dbc.postgresql.api.Postgresq
             })).defaultIfEmpty(IsolationLevel.READ_COMMITTED).last();
     }
 
-    static class PostgresConnectionException extends R2dbcNonTransientResourceException {
+    static class PostgresConnectionException extends R2dbcNonTransientResourceException implements PostgresqlException {
 
-        public PostgresConnectionException(String msg, @Nullable Throwable cause) {
-            super(msg, cause);
+        private static final String CONNECTION_DOES_NOT_EXIST = "08003";
+
+        private final ErrorDetails errorDetails;
+
+        public PostgresConnectionException(String reason, @Nullable Throwable cause) {
+            super(reason, CONNECTION_DOES_NOT_EXIST, 0, null, cause);
+            this.errorDetails = ErrorDetails.fromCodeAndMessage(CONNECTION_DOES_NOT_EXIST, reason);
+        }
+
+        @Override
+        public ErrorDetails getErrorDetails() {
+            return this.errorDetails;
         }
 
     }

src/main/java/io/r2dbc/postgresql/api/ErrorDetails.java

@@ -23,6 +23,7 @@
 
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -130,6 +131,22 @@ public static ErrorDetails fromMessage(String message) {
         return new ErrorDetails(Collections.singletonMap(MESSAGE, message));
     }
 
+    /**
+     * Create a new {@link ErrorDetails}
+     *
+     * @param code    the error code
+     * @param message the error message
+     * @return the {@link ErrorDetails} object
+     */
+    public static ErrorDetails fromCodeAndMessage(String code, String message) {
+
+        Map<FieldType, String> details = new LinkedHashMap<>(2);
+        details.put(CODE, code);
+        details.put(MESSAGE, message);
+
+        return new ErrorDetails(details);
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) {

src/main/java/io/r2dbc/postgresql/client/AbstractPostgresSSLHandlerAdapter.java

@@ -22,6 +22,8 @@
 import io.netty.handler.ssl.SslHandler;
 import io.netty.util.concurrent.Future;
 import io.netty.util.concurrent.GenericFutureListener;
+import io.r2dbc.postgresql.api.ErrorDetails;
+import io.r2dbc.postgresql.api.PostgresqlException;
 import io.r2dbc.spi.R2dbcPermissionDeniedException;
 import reactor.core.publisher.Mono;
 
@@ -86,10 +88,18 @@ SslHandler getSslHandler() {
     /**
      * Postgres-specific {@link R2dbcPermissionDeniedException}.
      */
-    static final class PostgresqlSslException extends R2dbcPermissionDeniedException {
+    static final class PostgresqlSslException extends R2dbcPermissionDeniedException implements PostgresqlException {
 
-        PostgresqlSslException(String msg) {
-            super(msg);
+        private final ErrorDetails errorDetails;
+
+        PostgresqlSslException(String reason) {
+            super(reason, ReactorNettyClient.CONNECTION_FAILURE, 0, (String) null);
+            this.errorDetails = ErrorDetails.fromCodeAndMessage(ReactorNettyClient.CONNECTION_FAILURE, reason);
+        }
+
+        @Override
+        public ErrorDetails getErrorDetails() {
+            return this.errorDetails;
         }
 
     }

src/main/java/io/r2dbc/postgresql/client/ReactorNettyClient.java

@@ -28,6 +28,8 @@
 import io.netty.util.ReferenceCountUtil;
 import io.netty.util.internal.logging.InternalLogger;
 import io.netty.util.internal.logging.InternalLoggerFactory;
+import io.r2dbc.postgresql.api.ErrorDetails;
+import io.r2dbc.postgresql.api.PostgresqlException;
 import io.r2dbc.postgresql.message.backend.BackendKeyData;
 import io.r2dbc.postgresql.message.backend.BackendMessage;
 import io.r2dbc.postgresql.message.backend.BackendMessageDecoder;
@@ -89,6 +91,8 @@
  */
 public final class ReactorNettyClient implements Client {
 
+    static final String CONNECTION_FAILURE = "08006";
+
     private static final Logger logger = Loggers.getLogger(ReactorNettyClient.class);
 
     private static final boolean DEBUG_ENABLED = logger.isDebugEnabled();
@@ -551,38 +555,70 @@ public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
 
     }
 
-    static class PostgresConnectionClosedException extends R2dbcNonTransientResourceException {
+    static class PostgresConnectionClosedException extends R2dbcNonTransientResourceException implements PostgresqlException {
+
+        private final ErrorDetails errorDetails;
 
         public PostgresConnectionClosedException(String reason) {
-            super(reason);
+            super(reason, CONNECTION_FAILURE, 0, (String) null);
+            this.errorDetails = ErrorDetails.fromCodeAndMessage(CONNECTION_FAILURE, reason);
         }
 
         public PostgresConnectionClosedException(String reason, @Nullable Throwable cause) {
-            super(reason, cause);
+            super(reason, CONNECTION_FAILURE, 0, null, cause);
+            this.errorDetails = ErrorDetails.fromCodeAndMessage(CONNECTION_FAILURE, reason);
+        }
+
+        @Override
+        public ErrorDetails getErrorDetails() {
+            return this.errorDetails;
         }
 
     }
 
-    static class PostgresConnectionException extends R2dbcNonTransientResourceException {
+    static class PostgresConnectionException extends R2dbcNonTransientResourceException implements PostgresqlException {
+
+        private final static ErrorDetails ERROR_DETAILS = ErrorDetails.fromCodeAndMessage(CONNECTION_FAILURE, "An I/O error occurred while sending to the backend or receiving from the backend");
 
         public PostgresConnectionException(Throwable cause) {
-            super(cause);
+            super(ERROR_DETAILS.getMessage(), ERROR_DETAILS.getCode(), 0, null, cause);
+        }
+
+        @Override
+        public ErrorDetails getErrorDetails() {
+            return ERROR_DETAILS;
         }
 
     }
 
-    static class RequestQueueException extends R2dbcTransientResourceException {
+    static class RequestQueueException extends R2dbcTransientResourceException implements PostgresqlException {
+
+        private final ErrorDetails errorDetails;
 
         public RequestQueueException(String message) {
-            super(message);
+            super(message, CONNECTION_FAILURE, 0, (String) null);
+            this.errorDetails = ErrorDetails.fromCodeAndMessage(CONNECTION_FAILURE, message);
+        }
+
+        @Override
+        public ErrorDetails getErrorDetails() {
+            return this.errorDetails;
         }
 
     }
 
-    static class ResponseQueueException extends R2dbcNonTransientResourceException {
+    static class ResponseQueueException extends R2dbcNonTransientResourceException implements PostgresqlException {
+
+        private final ErrorDetails errorDetails;
 
         public ResponseQueueException(String message) {
-            super(message);
+            super(message, CONNECTION_FAILURE, 0, (String) null);
+            this.errorDetails = ErrorDetails.fromCodeAndMessage(CONNECTION_FAILURE, message);
+        }
+
+        @Override
+        public ErrorDetails getErrorDetails() {
+            return this.errorDetails;
         }
 
     }

src/test/java/io/r2dbc/postgresql/Repro.java

@@ -0,0 +1,55 @@
+/*
+ * 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;
+
+import org.junit.jupiter.api.Test;
+
+import java.time.Instant;
+import java.time.LocalTime;
+import java.time.OffsetDateTime;
+
+/**
+ * @author Mark Paluch
+ */
+public class Repro extends AbstractIntegrationTests{
+
+    @Test
+    void name() {
+
+
+        SERVER.getJdbcOperations().execute("DROP TABLE IF EXISTS repro;");
+        SERVER.getJdbcOperations().execute("CREATE TABLE repro (    created_at timestamp not null,\n" +
+            "    mytime time not null,\n" +
+            "    updated_at timestamp not null,\n" +
+            "    transaction_amount bigint not null,\n" +
+            "    transaction_currency varchar(3) not null,\n" +
+            "    transaction_at timestamptz not null)");
+
+        for (int i = 0; i < 10; i++) {
+
+            connection.createStatement("INSERT INTO repro VALUES($1, $2, $3, $4, $5, $6)")
+            .bind("$1", Instant.now())
+            .bind("$2", LocalTime.now())
+            .bind("$3", Instant.now())
+            .bind("$4", 1234)
+            .bind("$5", "FOO")
+            .bind("$6", OffsetDateTime.now()).execute().flatMap(it -> it.getRowsUpdated()).blockLast();
+        }
+
+    }
+
+}