Skip to content

Ensure BoltConnection does not send signals after summaries are finished #1592

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Nov 25, 2024
Merged

Ensure BoltConnection does not send signals after summaries are finished #1592

Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
92 changes: 60 additions & 32 deletions driver/src/main/java/org/neo4j/driver/internal/bolt/basicimpl/BoltConnectionImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -563,94 +563,122 @@ private ResponseHandleImpl(ResponseHandler delegate, int expectedSummaries) {
@Override
public void onError(Throwable throwable) {
if (!(throwable instanceof MessageIgnoredException)) {
runIgnoringError(() -> delegate.onError(throwable));
if (!(throwable instanceof Neo4jException)
|| throwable instanceof ServiceUnavailableException
|| throwable instanceof ProtocolException) {
// assume unrecoverable error, ensure onComplete
expectedSummaries = 1;
if (!summariesFuture.isDone()) {
runIgnoringError(() -> delegate.onError(throwable));
if (!(throwable instanceof Neo4jException)
|| throwable instanceof ServiceUnavailableException
|| throwable instanceof ProtocolException) {
// assume unrecoverable error, ensure onComplete
expectedSummaries = 1;
}
handleSummary();
}
handleSummary();
} else {
onIgnored();
}
}

@Override
public void onBeginSummary(BeginSummary summary) {
runIgnoringError(() -> delegate.onBeginSummary(summary));
handleSummary();
if (!summariesFuture.isDone()) {
runIgnoringError(() -> delegate.onBeginSummary(summary));
handleSummary();
}
}

@Override
public void onRunSummary(RunSummary summary) {
runIgnoringError(() -> delegate.onRunSummary(summary));
handleSummary();
if (!summariesFuture.isDone()) {
runIgnoringError(() -> delegate.onRunSummary(summary));
handleSummary();
}
}

@Override
public void onRecord(Value[] fields) {
runIgnoringError(() -> delegate.onRecord(fields));
if (!summariesFuture.isDone()) {
runIgnoringError(() -> delegate.onRecord(fields));
}
}

@Override
public void onPullSummary(PullSummary summary) {
runIgnoringError(() -> delegate.onPullSummary(summary));
handleSummary();
if (!summariesFuture.isDone()) {
runIgnoringError(() -> delegate.onPullSummary(summary));
handleSummary();
}
}

@Override
public void onDiscardSummary(DiscardSummary summary) {
runIgnoringError(() -> delegate.onDiscardSummary(summary));
handleSummary();
if (!summariesFuture.isDone()) {
runIgnoringError(() -> delegate.onDiscardSummary(summary));
handleSummary();
}
}

@Override
public void onCommitSummary(CommitSummary summary) {
runIgnoringError(() -> delegate.onCommitSummary(summary));
handleSummary();
if (!summariesFuture.isDone()) {
runIgnoringError(() -> delegate.onCommitSummary(summary));
handleSummary();
}
}

@Override
public void onRollbackSummary(RollbackSummary summary) {
runIgnoringError(() -> delegate.onRollbackSummary(summary));
handleSummary();
if (!summariesFuture.isDone()) {
runIgnoringError(() -> delegate.onRollbackSummary(summary));
handleSummary();
}
}

@Override
public void onResetSummary(ResetSummary summary) {
runIgnoringError(() -> delegate.onResetSummary(summary));
handleSummary();
if (!summariesFuture.isDone()) {
runIgnoringError(() -> delegate.onResetSummary(summary));
handleSummary();
}
}

@Override
public void onRouteSummary(RouteSummary summary) {
runIgnoringError(() -> delegate.onRouteSummary(summary));
handleSummary();
if (!summariesFuture.isDone()) {
runIgnoringError(() -> delegate.onRouteSummary(summary));
handleSummary();
}
}

@Override
public void onLogoffSummary(LogoffSummary summary) {
runIgnoringError(() -> delegate.onLogoffSummary(summary));
handleSummary();
if (!summariesFuture.isDone()) {
runIgnoringError(() -> delegate.onLogoffSummary(summary));
handleSummary();
}
}

@Override
public void onLogonSummary(LogonSummary summary) {
runIgnoringError(() -> delegate.onLogonSummary(summary));
handleSummary();
if (!summariesFuture.isDone()) {
runIgnoringError(() -> delegate.onLogonSummary(summary));
handleSummary();
}
}

@Override
public void onTelemetrySummary(TelemetrySummary summary) {
runIgnoringError(() -> delegate.onTelemetrySummary(summary));
handleSummary();
if (!summariesFuture.isDone()) {
runIgnoringError(() -> delegate.onTelemetrySummary(summary));
handleSummary();
}
}

@Override
public void onIgnored() {
runIgnoringError(delegate::onIgnored);
handleSummary();
if (!summariesFuture.isDone()) {
runIgnoringError(delegate::onIgnored);
handleSummary();
}
}

@Override
Expand Down