[jOOQ/jOOQ#14499] Add an R2DBC LoggingConnection

jOOQ/src/main/java/org/jooq/tools/jdbc/LoggingConnection.java

@@ -46,6 +46,9 @@ import java.sql.Statement;
 import org.jooq.tools.JooqLogger;
 
 /**
+ * A JDBC {@link Connection} proxy that logs all statements that are prepared or
+ * executed using it.
+ *
  * @author Lukas Eder
  */
 public class LoggingConnection extends DefaultConnection {

jOOQ/src/main/java/org/jooq/tools/r2dbc/DefaultConnection.java

@@ -0,0 +1,159 @@
+/*
+ * 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.
+ *
+ * Other licenses:
+ * -----------------------------------------------------------------------------
+ * Commercial licenses for this work are available. These replace the above
+ * ASL 2.0 and offer limited warranties, support, maintenance, and commercial
+ * database integrations.
+ *
+ * For more information, please visit: https://www.jooq.org/legal/licensing
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ */
+package org.jooq.tools.r2dbc;
+
+import java.time.Duration;
+
+import org.reactivestreams.Publisher;
+
+import io.r2dbc.spi.Batch;
+import io.r2dbc.spi.Connection;
+import io.r2dbc.spi.ConnectionMetadata;
+import io.r2dbc.spi.IsolationLevel;
+import io.r2dbc.spi.Statement;
+import io.r2dbc.spi.TransactionDefinition;
+import io.r2dbc.spi.ValidationDepth;
+
+/**
+ * A default R2DBC {@link Connection} implementation delegating all R2DBC
+ * calls to an internal getDelegate().
+ *
+ * @author Lukas Eder
+ */
+public class DefaultConnection implements Connection {
+
+    private final Connection delegate;
+
+    public DefaultConnection(Connection delegate) {
+        this.delegate = delegate;
+    }
+
+    public Connection getDelegate() {
+        return delegate;
+    }
+
+    @Override
+    public Publisher<Void> beginTransaction() {
+        return getDelegate().beginTransaction();
+    }
+
+    @Override
+    public Publisher<Void> beginTransaction(TransactionDefinition definition) {
+        return getDelegate().beginTransaction(definition);
+    }
+
+    @Override
+    public Publisher<Void> close() {
+        return getDelegate().close();
+    }
+
+    @Override
+    public Publisher<Void> commitTransaction() {
+        return getDelegate().commitTransaction();
+    }
+
+    @Override
+    public Batch createBatch() {
+        return getDelegate().createBatch();
+    }
+
+    @Override
+    public Publisher<Void> createSavepoint(String name) {
+        return getDelegate().createSavepoint(name);
+    }
+
+    @Override
+    public Statement createStatement(String sql) {
+        return getDelegate().createStatement(sql);
+    }
+
+    @Override
+    public boolean isAutoCommit() {
+        return getDelegate().isAutoCommit();
+    }
+
+    @Override
+    public ConnectionMetadata getMetadata() {
+        return getDelegate().getMetadata();
+    }
+
+    @Override
+    public IsolationLevel getTransactionIsolationLevel() {
+        return getDelegate().getTransactionIsolationLevel();
+    }
+
+    @Override
+    public Publisher<Void> releaseSavepoint(String name) {
+        return getDelegate().releaseSavepoint(name);
+    }
+
+    @Override
+    public Publisher<Void> rollbackTransaction() {
+        return getDelegate().rollbackTransaction();
+    }
+
+    @Override
+    public Publisher<Void> rollbackTransactionToSavepoint(String name) {
+        return getDelegate().rollbackTransactionToSavepoint(name);
+    }
+
+    @Override
+    public Publisher<Void> setAutoCommit(boolean autoCommit) {
+        return getDelegate().setAutoCommit(autoCommit);
+    }
+
+    @Override
+    public Publisher<Void> setLockWaitTimeout(Duration timeout) {
+        return getDelegate().setLockWaitTimeout(timeout);
+    }
+
+    @Override
+    public Publisher<Void> setStatementTimeout(Duration timeout) {
+        return getDelegate().setStatementTimeout(timeout);
+    }
+
+    @Override
+    public Publisher<Void> setTransactionIsolationLevel(IsolationLevel isolationLevel) {
+        return getDelegate().setTransactionIsolationLevel(isolationLevel);
+    }
+
+    @Override
+    public Publisher<Boolean> validate(ValidationDepth depth) {
+        return getDelegate().validate(depth);
+    }
+}

jOOQ/src/main/java/org/jooq/tools/r2dbc/DefaultStatement.java

@@ -0,0 +1,102 @@
+/*
+ * 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.
+ *
+ * Other licenses:
+ * -----------------------------------------------------------------------------
+ * Commercial licenses for this work are available. These replace the above
+ * ASL 2.0 and offer limited warranties, support, maintenance, and commercial
+ * database integrations.
+ *
+ * For more information, please visit: https://www.jooq.org/legal/licensing
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ */
+package org.jooq.tools.r2dbc;
+
+import org.reactivestreams.Publisher;
+
+import io.r2dbc.spi.Result;
+import io.r2dbc.spi.Statement;
+
+/**
+ * A default R2DBC {@link Statement} implementation delegating all R2DBC calls
+ * to an internal getDelegate().
+ *
+ * @author Lukas Eder
+ */
+public class DefaultStatement implements Statement {
+
+    private final Statement delegate;
+
+    public DefaultStatement(Statement delegate) {
+        this.delegate = delegate;
+    }
+
+    public Statement getDelegate() {
+        return delegate;
+    }
+
+    @Override
+    public Statement add() {
+        return getDelegate().add();
+    }
+
+    @Override
+    public Statement bind(int index, Object value) {
+        return getDelegate().bind(index, value);
+    }
+
+    @Override
+    public Statement bind(String name, Object value) {
+        return getDelegate().bind(name, value);
+    }
+
+    @Override
+    public Statement bindNull(int index, Class<?> type) {
+        return getDelegate().bindNull(index, type);
+    }
+
+    @Override
+    public Statement bindNull(String name, Class<?> type) {
+        return getDelegate().bindNull(name, type);
+    }
+
+    @Override
+    public Publisher<? extends Result> execute() {
+        return getDelegate().execute();
+    }
+
+    @Override
+    public Statement returnGeneratedValues(String... columns) {
+        return getDelegate().returnGeneratedValues(columns);
+    }
+
+    @Override
+    public Statement fetchSize(int rows) {
+        return getDelegate().fetchSize(rows);
+    }
+}

jOOQ/src/main/java/org/jooq/tools/r2dbc/LoggingBatch.java

@@ -0,0 +1,86 @@
+/*
+ * 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.
+ *
+ * Other licenses:
+ * -----------------------------------------------------------------------------
+ * Commercial licenses for this work are available. These replace the above
+ * ASL 2.0 and offer limited warranties, support, maintenance, and commercial
+ * database integrations.
+ *
+ * For more information, please visit: https://www.jooq.org/legal/licensing
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ */
+package org.jooq.tools.r2dbc;
+
+import org.jooq.tools.JooqLogger;
+
+import org.reactivestreams.Publisher;
+
+import io.r2dbc.spi.Batch;
+import io.r2dbc.spi.Connection;
+import io.r2dbc.spi.Result;
+
+/**
+ * An R2DBC {@link Batch} proxy that logs all statements that are prepared
+ * or executed using it.
+ *
+ * @author Lukas Eder
+ */
+public class LoggingBatch implements Batch {
+
+    private static final JooqLogger log = JooqLogger.getLogger(LoggingBatch.class);
+
+    private final Batch delegate;
+
+    public LoggingBatch(Batch delegate) {
+        this.delegate = delegate;
+    }
+
+    public Batch getDelegate() {
+        return delegate;
+    }
+
+    @Override
+    public Batch add(String sql) {
+        if (log.isDebugEnabled())
+            log.debug("Batch::add", sql);
+
+        getDelegate().add(sql);
+        return this;
+    }
+
+    @Override
+    public Publisher<? extends Result> execute() {
+        return s -> {
+            if (log.isDebugEnabled())
+                log.debug("Batch::execute");
+
+            getDelegate().execute().subscribe(s);
+        };
+    }
+}

jOOQ/src/main/java/org/jooq/tools/r2dbc/LoggingConnection.java

@@ -0,0 +1,209 @@
+/*
+ * 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.
+ *
+ * Other licenses:
+ * -----------------------------------------------------------------------------
+ * Commercial licenses for this work are available. These replace the above
+ * ASL 2.0 and offer limited warranties, support, maintenance, and commercial
+ * database integrations.
+ *
+ * For more information, please visit: https://www.jooq.org/legal/licensing
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ */
+package org.jooq.tools.r2dbc;
+
+import java.time.Duration;
+
+import org.jooq.tools.JooqLogger;
+
+import org.reactivestreams.Publisher;
+
+import io.r2dbc.spi.Batch;
+import io.r2dbc.spi.Connection;
+import io.r2dbc.spi.IsolationLevel;
+import io.r2dbc.spi.Statement;
+import io.r2dbc.spi.TransactionDefinition;
+import io.r2dbc.spi.ValidationDepth;
+
+/**
+ * An R2DBC {@link Connection} proxy that logs all statements that are prepared
+ * or executed using it.
+ *
+ * @author Lukas Eder
+ */
+public class LoggingConnection extends DefaultConnection {
+
+    private static final JooqLogger log = JooqLogger.getLogger(LoggingConnection.class);
+
+    public LoggingConnection(Connection delegate) {
+        super(delegate);
+    }
+
+    @Override
+    public Publisher<Void> beginTransaction() {
+        return s -> {
+            if (log.isDebugEnabled())
+                log.debug("Connection::beginTransaction");
+
+            getDelegate().beginTransaction().subscribe(s);
+        };
+    }
+
+    @Override
+    public Publisher<Void> beginTransaction(TransactionDefinition definition) {
+        return s -> {
+            if (log.isDebugEnabled())
+                log.debug("Connection::beginTransaction", definition);
+
+            getDelegate().beginTransaction(definition).subscribe(s);
+        };
+    }
+
+    @Override
+    public Publisher<Void> close() {
+        return s -> {
+            if (log.isDebugEnabled())
+                log.debug("Connection::close");
+
+            getDelegate().close().subscribe(s);
+        };
+    }
+
+    @Override
+    public Publisher<Void> commitTransaction() {
+        return s -> {
+            if (log.isDebugEnabled())
+                log.debug("Connection::commitTransaction");
+
+            getDelegate().commitTransaction().subscribe(s);
+        };
+    }
+
+    @Override
+    public Batch createBatch() {
+        return new LoggingBatch(getDelegate().createBatch());
+    }
+
+    @Override
+    public Publisher<Void> createSavepoint(String name) {
+        return s -> {
+            if (log.isDebugEnabled())
+                log.debug("Connection::createSavepoint", name);
+
+            getDelegate().createSavepoint(name).subscribe(s);
+        };
+    }
+
+    @Override
+    public Statement createStatement(String sql) {
+        if (log.isDebugEnabled())
+            log.debug("Connection::createStatement", sql);
+
+        return new LoggingStatement(getDelegate().createStatement(sql));
+    }
+
+    @Override
+    public Publisher<Void> releaseSavepoint(String name) {
+        return s -> {
+            if (log.isDebugEnabled())
+                log.debug("Connection::releaseSavepoint", name);
+
+            getDelegate().releaseSavepoint(name).subscribe(s);
+        };
+    }
+
+    @Override
+    public Publisher<Void> rollbackTransaction() {
+        return s -> {
+            if (log.isDebugEnabled())
+                log.debug("Connection::rollbackTransaction");
+
+            getDelegate().rollbackTransaction().subscribe(s);
+        };
+    }
+
+    @Override
+    public Publisher<Void> rollbackTransactionToSavepoint(String name) {
+        return s -> {
+            if (log.isDebugEnabled())
+                log.debug("Connection::rollbackTransactionToSavepoint", name);
+
+            getDelegate().rollbackTransactionToSavepoint(name).subscribe(s);
+        };
+    }
+
+    @Override
+    public Publisher<Void> setAutoCommit(boolean autoCommit) {
+        return s -> {
+            if (log.isDebugEnabled())
+                log.debug("Connection::setAutoCommit", autoCommit);
+
+            getDelegate().setAutoCommit(autoCommit).subscribe(s);
+        };
+    }
+
+    @Override
+    public Publisher<Void> setLockWaitTimeout(Duration timeout) {
+        return s -> {
+            if (log.isDebugEnabled())
+                log.debug("Connection::setLockWaitTimeout", timeout);
+
+            getDelegate().setLockWaitTimeout(timeout).subscribe(s);
+        };
+    }
+
+    @Override
+    public Publisher<Void> setStatementTimeout(Duration timeout) {
+        return s -> {
+            if (log.isDebugEnabled())
+                log.debug("Connection::setStatementTimeout", timeout);
+
+            getDelegate().setStatementTimeout(timeout).subscribe(s);
+        };
+    }
+
+    @Override
+    public Publisher<Void> setTransactionIsolationLevel(IsolationLevel isolationLevel) {
+        return s -> {
+            if (log.isDebugEnabled())
+                log.debug("Connection::setTransactionIsolationLevel", isolationLevel);
+
+            getDelegate().setTransactionIsolationLevel(isolationLevel).subscribe(s);
+        };
+    }
+
+    @Override
+    public Publisher<Boolean> validate(ValidationDepth depth) {
+        return s -> {
+            if (log.isDebugEnabled())
+                log.debug("Connection::validate", depth);
+
+            getDelegate().validate(depth).subscribe(s);
+        };
+    }
+}

jOOQ/src/main/java/org/jooq/tools/r2dbc/LoggingStatement.java

@@ -0,0 +1,115 @@
+/*
+ * 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.
+ *
+ * Other licenses:
+ * -----------------------------------------------------------------------------
+ * Commercial licenses for this work are available. These replace the above
+ * ASL 2.0 and offer limited warranties, support, maintenance, and commercial
+ * database integrations.
+ *
+ * For more information, please visit: https://www.jooq.org/legal/licensing
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ */
+package org.jooq.tools.r2dbc;
+
+import org.jooq.tools.JooqLogger;
+
+import org.reactivestreams.Publisher;
+
+import io.r2dbc.spi.Result;
+import io.r2dbc.spi.Statement;
+
+/**
+ * An R2DBC {@link Statement} proxy that logs all statements that are prepared
+ * or executed using it.
+ *
+ * @author Lukas Eder
+ */
+public class LoggingStatement extends DefaultStatement {
+
+    private static final JooqLogger log = JooqLogger.getLogger(LoggingStatement.class);
+
+    public LoggingStatement(Statement delegate) {
+        super(delegate);
+    }
+
+    @Override
+    public Statement add() {
+        if (log.isDebugEnabled())
+            log.debug("Statement::add");
+
+        getDelegate().add();
+        return this;
+    }
+
+    @Override
+    public Statement bind(int index, Object value) {
+        if (log.isTraceEnabled())
+            log.trace("Statement::bind", "index = " + index + ", value = " + value);
+
+        getDelegate().bind(index, value);
+        return this;
+    }
+
+    @Override
+    public Statement bind(String name, Object value) {
+        if (log.isTraceEnabled())
+            log.trace("Statement::bind", "name = " + name + ", value = " + value);
+
+        getDelegate().bind(name, value);
+        return this;
+    }
+
+    @Override
+    public Statement bindNull(int index, Class<?> type) {
+        if (log.isTraceEnabled())
+            log.trace("Statement::bindNull", "index = " + index + ", type = " + type);
+
+        getDelegate().bindNull(index, type);
+        return this;
+    }
+
+    @Override
+    public Statement bindNull(String name, Class<?> type) {
+        if (log.isTraceEnabled())
+            log.trace("Statement::bindNull", "name = " + name + ", type = " + type);
+
+        getDelegate().bindNull(name, type);
+        return this;
+    }
+
+    @Override
+    public Publisher<? extends Result> execute() {
+        return s -> {
+            if (log.isDebugEnabled())
+                log.debug("Statement::execute");
+
+            getDelegate().execute().subscribe(s);
+        };
+    }
+}