Introduce connection context to allow correlation of log statements to the actual connection

[resolves #339]

Author: mp911de

README.md

@@ -456,6 +456,52 @@ Logging facilities:
 * Transport Logging (`io.r2dbc.postgresql.client`)
     * `DEBUG` enables `Message` exchange logging
     * `TRACE` enables traffic logging
+    
+Logging that is associated with a connection reports the logical connection id (`cid`) which is a driver-local connection counter and the Postgres Process Id (`pid`) once the connection handshake finishes.
+    
+## Getting Help
+
+Having trouble with R2DBC? We'd love to help!
+
+* Check the [spec documentation](https://r2dbc.io/spec/0.8.0.RELEASE/spec/html/), and [Javadoc](https://r2dbc.io/spec/0.8.0.RELEASE/api/).
+* If you are upgrading, check out the [changelog](https://r2dbc.io/spec/0.8.0.RELEASE/CHANGELOG.txt) for "new and noteworthy" features.
+* Ask a question - we monitor [stackoverflow.com](https://stackoverflow.com) for questions
+  tagged with [`r2dbc`](https://stackoverflow.com/tags/r2dbc). 
+  You can also chat with the community on [Gitter](https://gitter.im/r2dbc/r2dbc).
+* Report bugs with R2DBC PostgreSQL at [github.com/pgjdbc/r2dbc-postgresql/issues](https://github.com/pgjdbc/r2dbc-postgresql/issues).
+
+## Reporting Issues
+
+R2DBC uses GitHub as issue tracking system to record bugs and feature requests. 
+If you want to raise an issue, please follow the recommendations below:
+
+* Before you log a bug, please search the [issue tracker](https://github.com/pgjdbc/r2dbc-postgresql/issues) to see if someone has already reported the problem.
+* If the issue doesn't already exist, [create a new issue](https://github.com/pgjdbc/r2dbc-postgresql/issues/new).
+* Please provide as much information as possible with the issue report, we like to know the version of R2DBC PostgreSQL that you are using and JVM version.
+* If you need to paste code, or include a stack trace use Markdown ``` escapes before and after your text.
+* If possible try to create a test-case or project that replicates the issue. 
+Attach a link to your code or a compressed file containing your code.
+
+## Building from Source
+
+You don't need to build from source to use R2DBC PostgreSQL (binaries in Maven Central), but if you want to try out the latest and greatest, R2DBC PostgreSQL can be easily built with the
+[maven wrapper](https://github.com/takari/maven-wrapper). You also need JDK 1.8 and Docker to run integration tests.
+
+```bash
+ $ ./mvnw clean install
+```
+
+If you want to build with the regular `mvn` command, you will need [Maven v3.5.0 or above](https://maven.apache.org/run-maven/index.html).
+
+_Also see [CONTRIBUTING.adoc](.github/CONTRIBUTING.adoc) if you wish to submit pull requests._
+
+### Running JMH Benchmarks
+
+Running the JMH benchmarks builds and runs the benchmarks without running tests.
+
+```bash
+ $ ./mvnw clean install -Pjmh
+```
 
 ## License
 This project is released under version 2.0 of the [Apache License][l].

src/main/java/io/r2dbc/postgresql/ConnectionContext.java renamed to src/main/java/io/r2dbc/postgresql/ConnectionResources.java

@@ -25,7 +25,7 @@
 /**
  * @author Mark Paluch
  */
-final class ConnectionContext {
+final class ConnectionResources {
 
     private final Client client;
 
@@ -39,8 +39,8 @@ final class ConnectionContext {
 
     private final PortalNameSupplier portalNameSupplier;
 
-    ConnectionContext(Client client, Codecs codecs, PostgresqlConnection connection, PostgresqlConnectionConfiguration configuration, PortalNameSupplier portalNameSupplier,
-                      StatementCache statementCache) {
+    ConnectionResources(Client client, Codecs codecs, PostgresqlConnection connection, PostgresqlConnectionConfiguration configuration, PortalNameSupplier portalNameSupplier,
+                        StatementCache statementCache) {
         this.client = client;
         this.codecs = codecs;
         this.connection = connection;

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

@@ -48,15 +48,15 @@ final class ExtendedQueryPostgresqlStatement implements PostgresqlStatement {
 
     private final Bindings bindings;
 
-    private final ConnectionContext context;
+    private final ConnectionResources context;
 
     private final String sql;
 
     private int fetchSize;
 
     private String[] generatedColumns;
 
-    ExtendedQueryPostgresqlStatement(ConnectionContext context, String sql) {
+    ExtendedQueryPostgresqlStatement(ConnectionResources context, String sql) {
         this.context = Assert.requireNonNull(context, "context must not be null");
         this.sql = Assert.requireNonNull(sql, "sql must not be null");
         this.bindings = new Bindings(expectedSize(sql));
@@ -181,7 +181,7 @@ private Flux<io.r2dbc.postgresql.api.PostgresqlResult> execute(String sql) {
             .cast(io.r2dbc.postgresql.api.PostgresqlResult.class);
     }
 
-    static PostgresqlResult createPostgresqlResult(String sql, ExceptionFactory factory, String statementName, Binding binding, ConnectionContext context, int fetchSize) {
+    static PostgresqlResult createPostgresqlResult(String sql, ExceptionFactory factory, String statementName, Binding binding, ConnectionResources context, int fetchSize) {
         Flux<BackendMessage> messages = ExtendedQueryMessageFlow
             .execute(binding, context.getClient(), context.getPortalNameSupplier(), statementName, sql, context.getConfiguration().isForceBinary(), fetchSize)
             .filter(RESULT_FRAME_FILTER);

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

@@ -28,11 +28,11 @@
  */
 final class PostgresqlBatch implements io.r2dbc.postgresql.api.PostgresqlBatch {
 
-    private final ConnectionContext context;
+    private final ConnectionResources context;
 
     private final List<String> statements = new ArrayList<>();
 
-    PostgresqlBatch(ConnectionContext context) {
+    PostgresqlBatch(ConnectionResources context) {
         this.context = Assert.requireNonNull(context, "context must not be null");
     }
 

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

@@ -20,6 +20,7 @@
 import io.r2dbc.postgresql.api.PostgresqlResult;
 import io.r2dbc.postgresql.api.PostgresqlStatement;
 import io.r2dbc.postgresql.client.Client;
+import io.r2dbc.postgresql.client.ConnectionContext;
 import io.r2dbc.postgresql.client.PortalNameSupplier;
 import io.r2dbc.postgresql.client.SimpleQueryMessageFlow;
 import io.r2dbc.postgresql.client.TransactionStatus;
@@ -56,10 +57,12 @@ final class PostgresqlConnection implements io.r2dbc.postgresql.api.PostgresqlCo
 
     private final Logger logger = Loggers.getLogger(this.getClass());
 
-    private final ConnectionContext context;
-
     private final Client client;
 
+    private final ConnectionResources resources;
+
+    private final ConnectionContext connectionContext;
+
     private final Codecs codecs;
 
     private final Flux<Integer> validationQuery;
@@ -70,11 +73,12 @@ final class PostgresqlConnection implements io.r2dbc.postgresql.api.PostgresqlCo
 
     PostgresqlConnection(Client client, Codecs codecs, PortalNameSupplier portalNameSupplier, StatementCache statementCache, IsolationLevel isolationLevel,
                          PostgresqlConnectionConfiguration configuration) {
-        this.context = new ConnectionContext(client, codecs, this, configuration, portalNameSupplier, statementCache);
         this.client = Assert.requireNonNull(client, "client must not be null");
+        this.resources = new ConnectionResources(client, codecs, this, configuration, portalNameSupplier, statementCache);
+        this.connectionContext = client.getContext();
         this.codecs = Assert.requireNonNull(codecs, "codecs must not be null");
         this.isolationLevel = Assert.requireNonNull(isolationLevel, "isolationLevel must not be null");
-        this.validationQuery = new SimpleQueryPostgresqlStatement(this.context, "SELECT 1").fetchSize(0).execute().flatMap(PostgresqlResult::getRowsUpdated);
+        this.validationQuery = new SimpleQueryPostgresqlStatement(this.resources, "SELECT 1").fetchSize(0).execute().flatMap(PostgresqlResult::getRowsUpdated);
     }
 
     Client getClient() {
@@ -87,7 +91,7 @@ public Mono<Void> beginTransaction() {
             if (IDLE == transactionStatus) {
                 return exchange("BEGIN");
             } else {
-                this.logger.debug("Skipping begin transaction because status is {}", transactionStatus);
+                this.logger.debug(this.connectionContext.getMessage("Skipping begin transaction because status is {}"), transactionStatus);
                 return Mono.empty();
             }
         });
@@ -111,15 +115,15 @@ public Mono<Void> commitTransaction() {
             if (IDLE != transactionStatus) {
                 return exchange("COMMIT");
             } else {
-                this.logger.debug("Skipping commit transaction because status is {}", transactionStatus);
+                this.logger.debug(this.connectionContext.getMessage("Skipping commit transaction because status is {}"), transactionStatus);
                 return Mono.empty();
             }
         });
     }
 
     @Override
     public PostgresqlBatch createBatch() {
-        return new PostgresqlBatch(this.context);
+        return new PostgresqlBatch(this.resources);
     }
 
     @Override
@@ -131,7 +135,7 @@ public Mono<Void> createSavepoint(String name) {
                 if (OPEN == transactionStatus) {
                     return exchange(String.format("SAVEPOINT %s", name));
                 } else {
-                    this.logger.debug("Skipping create savepoint because status is {}", transactionStatus);
+                    this.logger.debug(this.connectionContext.getMessage("Skipping create savepoint because status is {}"), transactionStatus);
                     return Mono.empty();
                 }
             }));
@@ -142,9 +146,9 @@ public PostgresqlStatement createStatement(String sql) {
         Assert.requireNonNull(sql, "sql must not be null");
 
         if (SimpleQueryPostgresqlStatement.supports(sql)) {
-            return new SimpleQueryPostgresqlStatement(this.context, sql);
+            return new SimpleQueryPostgresqlStatement(this.resources, sql);
         } else if (ExtendedQueryPostgresqlStatement.supports(sql)) {
-            return new ExtendedQueryPostgresqlStatement(this.context, sql);
+            return new ExtendedQueryPostgresqlStatement(this.resources, sql);
         } else {
             throw new IllegalArgumentException(String.format("Statement '%s' cannot be created. This is often due to the presence of both multiple statements and parameters at the same time.", sql));
         }
@@ -203,7 +207,7 @@ public Mono<Void> releaseSavepoint(String name) {
             if (OPEN == transactionStatus) {
                 return exchange(String.format("RELEASE SAVEPOINT %s", name));
             } else {
-                this.logger.debug("Skipping release savepoint because status is {}", transactionStatus);
+                this.logger.debug(this.connectionContext.getMessage("Skipping release savepoint because status is {}"), transactionStatus);
                 return Mono.empty();
             }
         });
@@ -215,7 +219,7 @@ public Mono<Void> rollbackTransaction() {
             if (IDLE != transactionStatus) {
                 return exchange("ROLLBACK");
             } else {
-                this.logger.debug("Skipping rollback transaction because status is {}", transactionStatus);
+                this.logger.debug(this.connectionContext.getMessage("Skipping rollback transaction because status is {}"), transactionStatus);
                 return Mono.empty();
             }
         });
@@ -229,7 +233,7 @@ public Mono<Void> rollbackTransactionToSavepoint(String name) {
             if (IDLE != transactionStatus) {
                 return exchange(String.format("ROLLBACK TO SAVEPOINT %s", name));
             } else {
-                this.logger.debug("Skipping rollback transaction to savepoint because status is {}", transactionStatus);
+                this.logger.debug(this.connectionContext.getMessage("Skipping rollback transaction to savepoint because status is {}"), transactionStatus);
                 return Mono.empty();
             }
         });
@@ -240,17 +244,17 @@ public Mono<Void> setAutoCommit(boolean autoCommit) {
 
         return useTransactionStatus(transactionStatus -> {
 
-            this.logger.debug(String.format("Setting auto-commit mode to [%s]", autoCommit));
+            this.logger.debug(this.connectionContext.getMessage(String.format("Setting auto-commit mode to [%s]", autoCommit)));
 
             if (isAutoCommit()) {
                 if (!autoCommit) {
-                    this.logger.debug("Beginning transaction");
+                    this.logger.debug(this.connectionContext.getMessage("Beginning transaction"));
                     return beginTransaction();
                 }
             } else {
 
                 if (autoCommit) {
-                    this.logger.debug("Committing pending transactions");
+                    this.logger.debug(this.connectionContext.getMessage("Committing pending transactions"));
                     return commitTransaction();
                 }
             }
@@ -305,7 +309,7 @@ public void onNext(Integer integer) {
 
                 @Override
                 public void onError(Throwable t) {
-                    PostgresqlConnection.this.logger.debug("Validation failed", t);
+                    PostgresqlConnection.this.logger.debug(PostgresqlConnection.this.connectionContext.getMessage("Validation failed"), t);
                     sink.success(false);
                 }
 

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

@@ -45,7 +45,7 @@ final class PostgresqlResult extends AbstractReferenceCounted implements io.r2db
 
     private static final Predicate<BackendMessage> TAKE_UNTIL = or(CommandComplete.class::isInstance, EmptyQueryResponse.class::isInstance);
 
-    private final ConnectionContext context;
+    private final ConnectionResources context;
 
     private final Flux<BackendMessage> messages;
 
@@ -55,7 +55,7 @@ final class PostgresqlResult extends AbstractReferenceCounted implements io.r2db
 
     private volatile RowDescription rowDescription;
 
-    PostgresqlResult(ConnectionContext context, Flux<BackendMessage> messages, ExceptionFactory factory) {
+    PostgresqlResult(ConnectionResources context, Flux<BackendMessage> messages, ExceptionFactory factory) {
         this.context = Assert.requireNonNull(context, "context must not be null");
         this.messages = Assert.requireNonNull(messages, "messages must not be null");
         this.factory = Assert.requireNonNull(factory, "factory must not be null");
@@ -139,7 +139,7 @@ public String toString() {
             '}';
     }
 
-    static PostgresqlResult toResult(ConnectionContext context, Flux<BackendMessage> messages, ExceptionFactory factory) {
+    static PostgresqlResult toResult(ConnectionResources context, Flux<BackendMessage> messages, ExceptionFactory factory) {
         return new PostgresqlResult(context, messages, factory);
     }
 

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

@@ -32,15 +32,15 @@
  */
 final class PostgresqlRow implements io.r2dbc.postgresql.api.PostgresqlRow {
 
-    private final ConnectionContext context;
+    private final ConnectionResources context;
 
     private final List<RowDescription.Field> fields;
 
     private final ByteBuf[] data;
 
     private volatile boolean isReleased = false;
 
-    PostgresqlRow(ConnectionContext context, List<RowDescription.Field> fields, ByteBuf[] data) {
+    PostgresqlRow(ConnectionResources context, List<RowDescription.Field> fields, ByteBuf[] data) {
         this.context = Assert.requireNonNull(context, "context must not be null");
         this.fields = Assert.requireNonNull(fields, "fields must not be null");
         this.data = Assert.requireNonNull(data, "data must not be null");
@@ -107,7 +107,7 @@ public String toString() {
             '}';
     }
 
-    static PostgresqlRow toRow(ConnectionContext context, DataRow dataRow, RowDescription rowDescription) {
+    static PostgresqlRow toRow(ConnectionResources context, DataRow dataRow, RowDescription rowDescription) {
         Assert.requireNonNull(dataRow, "dataRow must not be null");
         Assert.requireNonNull(rowDescription, "rowDescription must not be null");
 

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

@@ -45,15 +45,15 @@ final class SimpleQueryPostgresqlStatement implements PostgresqlStatement {
 
     private static final Predicate<BackendMessage> WINDOW_UNTIL = or(CommandComplete.class::isInstance, EmptyQueryResponse.class::isInstance, ErrorResponse.class::isInstance);
 
-    private final ConnectionContext context;
+    private final ConnectionResources context;
 
     private final String sql;
 
     private String[] generatedColumns;
 
     private int fetchSize;
 
-    SimpleQueryPostgresqlStatement(ConnectionContext context, String sql) {
+    SimpleQueryPostgresqlStatement(ConnectionResources context, String sql) {
         this.context = Assert.requireNonNull(context, "context must not be null");
         this.sql = Assert.requireNonNull(sql, "sql must not be null");
         fetchSize(isBatch() ? NO_LIMIT : this.context.getConfiguration().getFetchSize(sql));

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

@@ -104,6 +104,14 @@ default Flux<BackendMessage> exchange(Publisher<FrontendMessage> requests) {
      */
     ByteBufAllocator getByteBufAllocator();
 
+    /**
+     * Returns the {@link ConnectionContext}.
+     *
+     * @return the {@link ConnectionContext}
+     * @since 0.8.6
+     */
+    ConnectionContext getContext();
+
     /**
      * Returns the connected process id if it has been communicated.
      *