Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@

- Upgraded SDK constraints and lints.
- Supporting more URL-based connection-string parameters (mostly for pool).
- Optimized `StackTrace` capture [#432](https://github.com/isoos/postgresql-dart/pull/432) by [gmpassos](https://github.com/gmpassos).

## 3.5.7.

Expand Down
47 changes: 33 additions & 14 deletions lib/src/v3/connection.dart
Original file line number Diff line number Diff line change
Expand Up @@ -86,8 +86,11 @@ abstract class _PgSessionBase implements Session {

/// Sends a message to the server and waits for a response [T], gracefully
/// handling error messages that might come in instead.
Future<T> _sendAndWaitForQuery<T extends ServerMessage>(ClientMessage send) {
final trace = StackTrace.current;
Future<T> _sendAndWaitForQuery<T extends ServerMessage>(
ClientMessage send, {
StackTrace? stackTrace,
}) {
final trace = stackTrace ?? StackTrace.current;

return _withResource(() {
_connection._channel.sink.add(
Expand Down Expand Up @@ -193,7 +196,8 @@ abstract class _PgSessionBase implements Session {
}

Future<_PreparedStatement> _prepare(Object query) async {
final trace = Trace.current();
final stackTrace = StackTrace.current;
final trace = Trace.from(stackTrace);
final conn = _connection;
final name = 's/${conn._statementCounter++}';
final description = InternalQueryDescription.wrap(
Expand All @@ -207,6 +211,7 @@ abstract class _PgSessionBase implements Session {
statementName: name,
typeOids: description.parameterTypes?.map((e) => e?.oid).toList(),
),
stackTrace: stackTrace,
);

return _PreparedStatement(description, name, this, trace);
Expand Down Expand Up @@ -710,18 +715,23 @@ class _PreparedStatement extends Statement {

@override
Future<Result> run(Object? parameters, {Duration? timeout}) async {
final stackTrace = StackTrace.current;
final trace = Trace.from(stackTrace);
_session._connection._queryCount++;
timeout ??= _session._settings.queryTimeout;
final items = <ResultRow>[];
final subscription = bind(parameters).listen(items.add);
final subscription = (bind(parameters) as _BoundStatement).listen(
items.add,
callerTrace: trace,
);
try {
return await (subscription as _PgResultStreamSubscription)._waitForResult(
items: items,
timeout: timeout,
);
} finally {
await subscription.cancel();
await _closePendingPortals();
await _closePendingPortals(stackTrace: stackTrace);
}
}

Expand All @@ -741,12 +751,13 @@ class _PreparedStatement extends Statement {
_portalsToClose!.add(portalName);
}

Future<void> _closePendingPortals() async {
Future<void> _closePendingPortals({StackTrace? stackTrace}) async {
final list = _portalsToClose;
while (list != null && list.isNotEmpty) {
final portalName = list.removeFirst();
await _session._sendAndWaitForQuery<CloseCompleteMessage>(
CloseMessage.portal(portalName),
stackTrace: stackTrace,
);
}
}
Expand All @@ -764,6 +775,7 @@ class _BoundStatement extends Stream<ResultRow> implements ResultStream {
Function? onError,
void Function()? onDone,
bool? cancelOnError,
Trace? callerTrace,
}) {
final controller = StreamController<ResultRow>();

Expand All @@ -774,7 +786,12 @@ class _BoundStatement extends Stream<ResultRow> implements ResultStream {
onDone: onDone,
cancelOnError: cancelOnError,
);
return _PgResultStreamSubscription(this, controller, subscription);
return _PgResultStreamSubscription(
this,
controller,
subscription,
callerTrace: callerTrace,
);
}
}

Expand Down Expand Up @@ -803,12 +820,13 @@ class _PgResultStreamSubscription
_PgResultStreamSubscription(
_BoundStatement statement,
this._controller,
this._source,
) : session = statement.statement._session,
ignoreRows = false,
_boundStatement = statement,
_parentTrace = statement.statement._trace,
_callerTrace = Trace.current() {
this._source, {
Trace? callerTrace,
}) : session = statement.statement._session,
ignoreRows = false,
_boundStatement = statement,
_parentTrace = statement.statement._trace,
_callerTrace = callerTrace ?? Trace.current() {
_scheduleStatement(() async {
connection._pending = this;

Expand Down Expand Up @@ -847,9 +865,10 @@ class _PgResultStreamSubscription
this._controller,
this._source,
this.ignoreRows, {
Trace? callerTrace,
void Function()? cleanup,
}) : _parentTrace = null,
_callerTrace = Trace.current() {
_callerTrace = callerTrace ?? Trace.current() {
_scheduleStatement(() async {
connection._pending = this;

Expand Down