Skip to content

Commit 8f34be6

Browse files
mp911demp911de
committed
Introduce connection context to allow correlation of log statements to the actual connection
[resolves #339]
1 parent 5f228c6 commit 8f34be6

18 files changed

+214
-62
lines changed

‎README.md

+2
Original file line numberDiff line numberDiff line change
@@ -484,6 +484,8 @@ Logging facilities:
484484
* `DEBUG` enables `Message` exchange logging
485485
* `TRACE` enables traffic logging
486486

487+
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.
488+
487489
## Getting Help
488490

489491
Having trouble with R2DBC? We'd love to help!

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

+3-3
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@
2525
/**
2626
* Value object capturing contextual connection resources such as {@link Client}, {@link Codecs} and the {@link PostgresqlConnection connection facade}.
2727
*/
28-
final class ConnectionContext {
28+
final class ConnectionResources {
2929

3030
private final Client client;
3131

@@ -39,8 +39,8 @@ final class ConnectionContext {
3939

4040
private final PortalNameSupplier portalNameSupplier;
4141

42-
ConnectionContext(Client client, Codecs codecs, PostgresqlConnection connection, PostgresqlConnectionConfiguration configuration, PortalNameSupplier portalNameSupplier,
43-
StatementCache statementCache) {
42+
ConnectionResources(Client client, Codecs codecs, PostgresqlConnection connection, PostgresqlConnectionConfiguration configuration, PortalNameSupplier portalNameSupplier,
43+
StatementCache statementCache) {
4444
this.client = client;
4545
this.codecs = codecs;
4646
this.connection = connection;

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

+3-3
Original file line numberDiff line numberDiff line change
@@ -48,15 +48,15 @@ final class ExtendedQueryPostgresqlStatement implements PostgresqlStatement {
4848

4949
private final Bindings bindings;
5050

51-
private final ConnectionContext context;
51+
private final ConnectionResources context;
5252

5353
private final String sql;
5454

5555
private int fetchSize;
5656

5757
private String[] generatedColumns;
5858

59-
ExtendedQueryPostgresqlStatement(ConnectionContext context, String sql) {
59+
ExtendedQueryPostgresqlStatement(ConnectionResources context, String sql) {
6060
this.context = Assert.requireNonNull(context, "context must not be null");
6161
this.sql = Assert.requireNonNull(sql, "sql must not be null");
6262
this.bindings = new Bindings(expectedSize(sql));
@@ -181,7 +181,7 @@ private Flux<io.r2dbc.postgresql.api.PostgresqlResult> execute(String sql) {
181181
.cast(io.r2dbc.postgresql.api.PostgresqlResult.class);
182182
}
183183

184-
static PostgresqlResult createPostgresqlResult(String sql, ExceptionFactory factory, String statementName, Binding binding, ConnectionContext context, int fetchSize) {
184+
static PostgresqlResult createPostgresqlResult(String sql, ExceptionFactory factory, String statementName, Binding binding, ConnectionResources context, int fetchSize) {
185185
Flux<BackendMessage> messages = ExtendedQueryMessageFlow
186186
.execute(binding, context.getClient(), context.getPortalNameSupplier(), statementName, sql, context.getConfiguration().isForceBinary(), fetchSize)
187187
.filter(RESULT_FRAME_FILTER);

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

+2-2
Original file line numberDiff line numberDiff line change
@@ -28,11 +28,11 @@
2828
*/
2929
final class PostgresqlBatch implements io.r2dbc.postgresql.api.PostgresqlBatch {
3030

31-
private final ConnectionContext context;
31+
private final ConnectionResources context;
3232

3333
private final List<String> statements = new ArrayList<>();
3434

35-
PostgresqlBatch(ConnectionContext context) {
35+
PostgresqlBatch(ConnectionResources context) {
3636
this.context = Assert.requireNonNull(context, "context must not be null");
3737
}
3838

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

+21-17
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121
import io.r2dbc.postgresql.api.PostgresqlResult;
2222
import io.r2dbc.postgresql.api.PostgresqlStatement;
2323
import io.r2dbc.postgresql.client.Client;
24+
import io.r2dbc.postgresql.client.ConnectionContext;
2425
import io.r2dbc.postgresql.client.PortalNameSupplier;
2526
import io.r2dbc.postgresql.client.SimpleQueryMessageFlow;
2627
import io.r2dbc.postgresql.client.TransactionStatus;
@@ -59,10 +60,12 @@ final class PostgresqlConnection implements io.r2dbc.postgresql.api.PostgresqlCo
5960

6061
private final Logger logger = Loggers.getLogger(this.getClass());
6162

62-
private final ConnectionContext context;
63-
6463
private final Client client;
6564

65+
private final ConnectionResources resources;
66+
67+
private final ConnectionContext connectionContext;
68+
6669
private final Codecs codecs;
6770

6871
private final Flux<Integer> validationQuery;
@@ -73,11 +76,12 @@ final class PostgresqlConnection implements io.r2dbc.postgresql.api.PostgresqlCo
7376

7477
PostgresqlConnection(Client client, Codecs codecs, PortalNameSupplier portalNameSupplier, StatementCache statementCache, IsolationLevel isolationLevel,
7578
PostgresqlConnectionConfiguration configuration) {
76-
this.context = new ConnectionContext(client, codecs, this, configuration, portalNameSupplier, statementCache);
7779
this.client = Assert.requireNonNull(client, "client must not be null");
80+
this.resources = new ConnectionResources(client, codecs, this, configuration, portalNameSupplier, statementCache);
81+
this.connectionContext = client.getContext();
7882
this.codecs = Assert.requireNonNull(codecs, "codecs must not be null");
7983
this.isolationLevel = Assert.requireNonNull(isolationLevel, "isolationLevel must not be null");
80-
this.validationQuery = new SimpleQueryPostgresqlStatement(this.context, "SELECT 1").fetchSize(0).execute().flatMap(PostgresqlResult::getRowsUpdated);
84+
this.validationQuery = new SimpleQueryPostgresqlStatement(this.resources, "SELECT 1").fetchSize(0).execute().flatMap(PostgresqlResult::getRowsUpdated);
8185
}
8286

8387
Client getClient() {
@@ -90,7 +94,7 @@ public Mono<Void> beginTransaction() {
9094
if (IDLE == transactionStatus) {
9195
return exchange("BEGIN");
9296
} else {
93-
this.logger.debug("Skipping begin transaction because status is {}", transactionStatus);
97+
this.logger.debug(this.connectionContext.getMessage("Skipping begin transaction because status is {}"), transactionStatus);
9498
return Mono.empty();
9599
}
96100
});
@@ -137,15 +141,15 @@ public Mono<Void> commitTransaction() {
137141
sink.next(message);
138142
});
139143
} else {
140-
this.logger.debug("Skipping commit transaction because status is {}", transactionStatus);
144+
this.logger.debug(this.connectionContext.getMessage("Skipping commit transaction because status is {}"), transactionStatus);
141145
return Mono.empty();
142146
}
143147
});
144148
}
145149

146150
@Override
147151
public PostgresqlBatch createBatch() {
148-
return new PostgresqlBatch(this.context);
152+
return new PostgresqlBatch(this.resources);
149153
}
150154

151155
@Override
@@ -157,7 +161,7 @@ public Mono<Void> createSavepoint(String name) {
157161
if (OPEN == transactionStatus) {
158162
return exchange(String.format("SAVEPOINT %s", name));
159163
} else {
160-
this.logger.debug("Skipping create savepoint because status is {}", transactionStatus);
164+
this.logger.debug(this.connectionContext.getMessage("Skipping create savepoint because status is {}"), transactionStatus);
161165
return Mono.empty();
162166
}
163167
}));
@@ -168,9 +172,9 @@ public PostgresqlStatement createStatement(String sql) {
168172
Assert.requireNonNull(sql, "sql must not be null");
169173

170174
if (SimpleQueryPostgresqlStatement.supports(sql)) {
171-
return new SimpleQueryPostgresqlStatement(this.context, sql);
175+
return new SimpleQueryPostgresqlStatement(this.resources, sql);
172176
} else if (ExtendedQueryPostgresqlStatement.supports(sql)) {
173-
return new ExtendedQueryPostgresqlStatement(this.context, sql);
177+
return new ExtendedQueryPostgresqlStatement(this.resources, sql);
174178
} else {
175179
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));
176180
}
@@ -229,7 +233,7 @@ public Mono<Void> releaseSavepoint(String name) {
229233
if (OPEN == transactionStatus) {
230234
return exchange(String.format("RELEASE SAVEPOINT %s", name));
231235
} else {
232-
this.logger.debug("Skipping release savepoint because status is {}", transactionStatus);
236+
this.logger.debug(this.connectionContext.getMessage("Skipping release savepoint because status is {}"), transactionStatus);
233237
return Mono.empty();
234238
}
235239
});
@@ -241,7 +245,7 @@ public Mono<Void> rollbackTransaction() {
241245
if (IDLE != transactionStatus) {
242246
return exchange("ROLLBACK");
243247
} else {
244-
this.logger.debug("Skipping rollback transaction because status is {}", transactionStatus);
248+
this.logger.debug(this.connectionContext.getMessage("Skipping rollback transaction because status is {}"), transactionStatus);
245249
return Mono.empty();
246250
}
247251
});
@@ -255,7 +259,7 @@ public Mono<Void> rollbackTransactionToSavepoint(String name) {
255259
if (IDLE != transactionStatus) {
256260
return exchange(String.format("ROLLBACK TO SAVEPOINT %s", name));
257261
} else {
258-
this.logger.debug("Skipping rollback transaction to savepoint because status is {}", transactionStatus);
262+
this.logger.debug(this.connectionContext.getMessage("Skipping rollback transaction to savepoint because status is {}"), transactionStatus);
259263
return Mono.empty();
260264
}
261265
});
@@ -266,17 +270,17 @@ public Mono<Void> setAutoCommit(boolean autoCommit) {
266270

267271
return useTransactionStatus(transactionStatus -> {
268272

269-
this.logger.debug(String.format("Setting auto-commit mode to [%s]", autoCommit));
273+
this.logger.debug(this.connectionContext.getMessage(String.format("Setting auto-commit mode to [%s]", autoCommit)));
270274

271275
if (isAutoCommit()) {
272276
if (!autoCommit) {
273-
this.logger.debug("Beginning transaction");
277+
this.logger.debug(this.connectionContext.getMessage("Beginning transaction"));
274278
return beginTransaction();
275279
}
276280
} else {
277281

278282
if (autoCommit) {
279-
this.logger.debug("Committing pending transactions");
283+
this.logger.debug(this.connectionContext.getMessage("Committing pending transactions"));
280284
return commitTransaction();
281285
}
282286
}
@@ -331,7 +335,7 @@ public void onNext(Integer integer) {
331335

332336
@Override
333337
public void onError(Throwable t) {
334-
PostgresqlConnection.this.logger.debug("Validation failed", t);
338+
PostgresqlConnection.this.logger.debug(PostgresqlConnection.this.connectionContext.getMessage("Validation failed"), t);
335339
sink.success(false);
336340
}
337341

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

+3-3
Original file line numberDiff line numberDiff line change
@@ -45,7 +45,7 @@ final class PostgresqlResult extends AbstractReferenceCounted implements io.r2db
4545

4646
private static final Predicate<BackendMessage> TAKE_UNTIL = or(CommandComplete.class::isInstance, EmptyQueryResponse.class::isInstance);
4747

48-
private final ConnectionContext context;
48+
private final ConnectionResources context;
4949

5050
private final Flux<BackendMessage> messages;
5151

@@ -55,7 +55,7 @@ final class PostgresqlResult extends AbstractReferenceCounted implements io.r2db
5555

5656
private volatile RowDescription rowDescription;
5757

58-
private PostgresqlResult(ConnectionContext context, Flux<BackendMessage> messages, ExceptionFactory factory) {
58+
private PostgresqlResult(ConnectionResources context, Flux<BackendMessage> messages, ExceptionFactory factory) {
5959
this.context = context;
6060
this.messages = messages;
6161
this.factory = factory;
@@ -138,7 +138,7 @@ public String toString() {
138138
'}';
139139
}
140140

141-
static PostgresqlResult toResult(ConnectionContext context, Flux<BackendMessage> messages, ExceptionFactory factory) {
141+
static PostgresqlResult toResult(ConnectionResources context, Flux<BackendMessage> messages, ExceptionFactory factory) {
142142
return new PostgresqlResult(context, messages, factory);
143143
}
144144

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

+5-5
Original file line numberDiff line numberDiff line change
@@ -35,15 +35,15 @@
3535
*/
3636
final class PostgresqlRow implements io.r2dbc.postgresql.api.PostgresqlRow {
3737

38-
private final ConnectionContext context;
38+
private final ConnectionResources context;
3939

4040
private final List<RowDescription.Field> fields;
4141

4242
private final ByteBuf[] data;
4343

4444
private volatile boolean isReleased = false;
4545

46-
PostgresqlRow(ConnectionContext context, List<RowDescription.Field> fields, ByteBuf[] data) {
46+
PostgresqlRow(ConnectionResources context, List<RowDescription.Field> fields, ByteBuf[] data) {
4747
this.context = Assert.requireNonNull(context, "context must not be null");
4848
this.fields = Assert.requireNonNull(fields, "fields must not be null");
4949
this.data = Assert.requireNonNull(data, "data must not be null");
@@ -128,7 +128,7 @@ public String toString() {
128128
'}';
129129
}
130130

131-
static PostgresqlRow toRow(ConnectionContext context, DataRow dataRow, RowDescription rowDescription) {
131+
static PostgresqlRow toRow(ConnectionResources context, DataRow dataRow, RowDescription rowDescription) {
132132
Assert.requireNonNull(dataRow, "dataRow must not be null");
133133
Assert.requireNonNull(rowDescription, "rowDescription must not be null");
134134

@@ -186,11 +186,11 @@ private void requireNotReleased() {
186186
*/
187187
static class AttachedRefCursor implements RefCursor {
188188

189-
private final ConnectionContext context;
189+
private final ConnectionResources context;
190190

191191
private final String portal;
192192

193-
AttachedRefCursor(ConnectionContext context, String portal) {
193+
AttachedRefCursor(ConnectionResources context, String portal) {
194194
this.context = Assert.requireNonNull(context, "connection must not be null");
195195
this.portal = Assert.requireNotEmpty(portal, "portal must not be empty");
196196
}

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

+2-2
Original file line numberDiff line numberDiff line change
@@ -45,15 +45,15 @@ final class SimpleQueryPostgresqlStatement implements PostgresqlStatement {
4545

4646
private static final Predicate<BackendMessage> WINDOW_UNTIL = or(CommandComplete.class::isInstance, EmptyQueryResponse.class::isInstance, ErrorResponse.class::isInstance);
4747

48-
private final ConnectionContext context;
48+
private final ConnectionResources context;
4949

5050
private final String sql;
5151

5252
private String[] generatedColumns;
5353

5454
private int fetchSize;
5555

56-
SimpleQueryPostgresqlStatement(ConnectionContext context, String sql) {
56+
SimpleQueryPostgresqlStatement(ConnectionResources context, String sql) {
5757
this.context = Assert.requireNonNull(context, "context must not be null");
5858
this.sql = Assert.requireNonNull(sql, "sql must not be null");
5959
fetchSize(isBatch() ? NO_LIMIT : this.context.getConfiguration().getFetchSize(sql));

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

+8
Original file line numberDiff line numberDiff line change
@@ -105,6 +105,14 @@ default Flux<BackendMessage> exchange(Publisher<FrontendMessage> requests) {
105105
*/
106106
ByteBufAllocator getByteBufAllocator();
107107

108+
/**
109+
* Returns the {@link ConnectionContext}.
110+
*
111+
* @return the {@link ConnectionContext}
112+
* @since 0.8.6
113+
*/
114+
ConnectionContext getContext();
115+
108116
/**
109117
* Returns the connected process id if it has been communicated.
110118
*

0 commit comments

Comments
 (0)