Skip to content

Commit 14734c3

Browse files
michael-simonsmichael-simons
committed
GH-2370 - Consistently use auto commit query runners outside Spring transactions.
In addition, local bookmark managing has been introduced. Fixes #2370.
1 parent d742694 commit 14734c3

File tree

6 files changed

+157
-73
lines changed

6 files changed

+157
-73
lines changed

‎src/main/java/org/springframework/data/neo4j/core/DefaultNeo4jClient.java

Lines changed: 39 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -16,13 +16,19 @@
1616
package org.springframework.data.neo4j.core;
1717

1818
import java.util.Collection;
19+
import java.util.Collections;
20+
import java.util.HashSet;
1921
import java.util.Map;
2022
import java.util.Optional;
23+
import java.util.Set;
24+
import java.util.concurrent.locks.ReentrantReadWriteLock;
25+
import java.util.function.BiConsumer;
2126
import java.util.function.BiFunction;
2227
import java.util.function.Function;
2328
import java.util.function.Supplier;
2429
import java.util.stream.Collectors;
2530

31+
import org.neo4j.driver.Bookmark;
2632
import org.neo4j.driver.Driver;
2733
import org.neo4j.driver.Query;
2834
import org.neo4j.driver.QueryRunner;
@@ -58,6 +64,10 @@ class DefaultNeo4jClient implements Neo4jClient {
5864
private final ConversionService conversionService;
5965
private final Neo4jPersistenceExceptionTranslator persistenceExceptionTranslator = new Neo4jPersistenceExceptionTranslator();
6066

67+
// Basically a local bookmark manager
68+
private final Set<Bookmark> bookmarks = new HashSet<>();
69+
private final ReentrantReadWriteLock bookmarksLock = new ReentrantReadWriteLock();
70+
6171
DefaultNeo4jClient(Driver driver) {
6272

6373
this.driver = driver;
@@ -70,19 +80,41 @@ class DefaultNeo4jClient implements Neo4jClient {
7080
DelegatingQueryRunner getQueryRunner(@Nullable final String targetDatabase) {
7181

7282
QueryRunner queryRunner = Neo4jTransactionManager.retrieveTransaction(driver, targetDatabase);
83+
Collection<Bookmark> lastBookmarks = Collections.emptySet();
7384
if (queryRunner == null) {
74-
queryRunner = driver.session(Neo4jTransactionUtils.defaultSessionConfig(targetDatabase));
85+
ReentrantReadWriteLock.ReadLock lock = bookmarksLock.readLock();
86+
try {
87+
lock.lock();
88+
lastBookmarks = new HashSet<>(bookmarks);
89+
queryRunner = driver.session(Neo4jTransactionUtils.sessionConfig(false, lastBookmarks, targetDatabase));
90+
} finally {
91+
lock.unlock();
92+
}
7593
}
7694

77-
return new DelegatingQueryRunner(queryRunner);
95+
return new DelegatingQueryRunner(queryRunner, lastBookmarks, (usedBookmarks, newBookmark) -> {
96+
97+
ReentrantReadWriteLock.WriteLock lock = bookmarksLock.writeLock();
98+
try {
99+
lock.lock();
100+
bookmarks.removeAll(usedBookmarks);
101+
bookmarks.add(newBookmark);
102+
} finally {
103+
lock.unlock();
104+
}
105+
});
78106
}
79107

80108
static class DelegatingQueryRunner implements QueryRunner, AutoCloseable {
81109

82110
private final QueryRunner delegate;
111+
private final Collection<Bookmark> usedBookmarks;
112+
private final BiConsumer<Collection<Bookmark>, Bookmark> newBookmarkConsumer;
83113

84-
DelegatingQueryRunner(QueryRunner delegate) {
114+
private DelegatingQueryRunner(QueryRunner delegate, Collection<Bookmark> lastBookmarks, BiConsumer<Collection<Bookmark>, Bookmark> newBookmarkConsumer) {
85115
this.delegate = delegate;
116+
this.usedBookmarks = lastBookmarks;
117+
this.newBookmarkConsumer = newBookmarkConsumer;
86118
}
87119

88120
@Override
@@ -91,7 +123,10 @@ public void close() {
91123
// We're only going to close sessions we have acquired inside the client, not something that
92124
// has been retrieved from the tx manager.
93125
if (this.delegate instanceof Session) {
94-
((Session) this.delegate).close();
126+
127+
Session session = (Session) this.delegate;
128+
session.close();
129+
this.newBookmarkConsumer.accept(usedBookmarks, session.lastBookmark());
95130
}
96131
}
97132

‎src/main/java/org/springframework/data/neo4j/core/DefaultReactiveNeo4jClient.java

Lines changed: 92 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -18,14 +18,24 @@
1818
import reactor.core.publisher.Flux;
1919
import reactor.core.publisher.Mono;
2020
import reactor.util.function.Tuple2;
21+
import reactor.util.function.Tuples;
2122

23+
import java.util.Collection;
24+
import java.util.Collections;
25+
import java.util.HashSet;
2226
import java.util.Map;
27+
import java.util.Set;
28+
import java.util.concurrent.locks.ReentrantReadWriteLock;
29+
import java.util.function.BiConsumer;
2330
import java.util.function.BiFunction;
2431
import java.util.function.Function;
2532
import java.util.function.Supplier;
2633

34+
import org.neo4j.driver.Bookmark;
2735
import org.neo4j.driver.Driver;
36+
import org.neo4j.driver.Query;
2837
import org.neo4j.driver.Record;
38+
import org.neo4j.driver.Value;
2939
import org.neo4j.driver.reactive.RxQueryRunner;
3040
import org.neo4j.driver.reactive.RxResult;
3141
import org.neo4j.driver.reactive.RxSession;
@@ -57,6 +67,10 @@ class DefaultReactiveNeo4jClient implements ReactiveNeo4jClient {
5767
private final ConversionService conversionService;
5868
private final Neo4jPersistenceExceptionTranslator persistenceExceptionTranslator = new Neo4jPersistenceExceptionTranslator();
5969

70+
// Basically a local bookmark manager
71+
private final Set<Bookmark> bookmarks = new HashSet<>();
72+
private final ReentrantReadWriteLock bookmarksLock = new ReentrantReadWriteLock();
73+
6074
DefaultReactiveNeo4jClient(Driver driver) {
6175

6276
this.driver = driver;
@@ -65,28 +79,96 @@ class DefaultReactiveNeo4jClient implements ReactiveNeo4jClient {
6579
new Neo4jConversions().registerConvertersIn((ConverterRegistry) conversionService);
6680
}
6781

68-
Mono<RxStatementRunnerHolder> retrieveRxStatementRunnerHolder(String targetDatabase) {
82+
Mono<RxQueryRunner> retrieveRxStatementRunnerHolder(String targetDatabase) {
6983

7084
return ReactiveNeo4jTransactionManager.retrieveReactiveTransaction(driver, targetDatabase)
71-
.map(rxTransaction -> new RxStatementRunnerHolder(rxTransaction, Mono.empty(), Mono.empty())) //
72-
.switchIfEmpty(Mono.using(() -> driver.rxSession(Neo4jTransactionUtils.defaultSessionConfig(targetDatabase)),
73-
session -> Mono.from(session.beginTransaction())
74-
.map(tx -> new RxStatementRunnerHolder(tx, tx.commit(), tx.rollback())),
75-
RxSession::close));
85+
.map(RxQueryRunner.class::cast)
86+
.zipWith(Mono.just(Collections.<Bookmark>emptySet()))
87+
.switchIfEmpty(Mono.fromSupplier(() -> {
88+
ReentrantReadWriteLock.ReadLock lock = bookmarksLock.readLock();
89+
try {
90+
lock.lock();
91+
Set<Bookmark> lastBookmarks = new HashSet<>(bookmarks);
92+
return Tuples.of(driver.rxSession(Neo4jTransactionUtils.sessionConfig(false, lastBookmarks, targetDatabase)), lastBookmarks);
93+
} finally {
94+
lock.unlock();
95+
}
96+
}))
97+
.map(t -> new DelegatingQueryRunner(t.getT1(), t.getT2(), (usedBookmarks, newBookmark) -> {
98+
ReentrantReadWriteLock.WriteLock lock = bookmarksLock.writeLock();
99+
try {
100+
lock.lock();
101+
bookmarks.removeAll(usedBookmarks);
102+
bookmarks.add(newBookmark);
103+
} finally {
104+
lock.unlock();
105+
}
106+
}));
107+
}
108+
109+
private static class DelegatingQueryRunner implements RxQueryRunner {
110+
111+
private final RxQueryRunner delegate;
112+
private final Collection<Bookmark> usedBookmarks;
113+
private final BiConsumer<Collection<Bookmark>, Bookmark> newBookmarkConsumer;
114+
115+
private DelegatingQueryRunner(RxQueryRunner delegate, Collection<Bookmark> lastBookmarks, BiConsumer<Collection<Bookmark>, Bookmark> newBookmarkConsumer) {
116+
this.delegate = delegate;
117+
this.usedBookmarks = lastBookmarks;
118+
this.newBookmarkConsumer = newBookmarkConsumer;
119+
}
120+
121+
Mono<Void> close() {
122+
123+
// We're only going to close sessions we have acquired inside the client, not something that
124+
// has been retrieved from the tx manager.
125+
if (this.delegate instanceof RxSession) {
126+
RxSession session = (RxSession) this.delegate;
127+
return Mono.fromDirect(session.close()).then().doOnSuccess(signal ->
128+
this.newBookmarkConsumer.accept(usedBookmarks, session.lastBookmark()));
129+
}
130+
131+
return Mono.empty();
132+
}
133+
134+
@Override
135+
public RxResult run(String query, Value parameters) {
136+
return delegate.run(query, parameters);
137+
}
138+
139+
@Override
140+
public RxResult run(String query, Map<String, Object> parameters) {
141+
return delegate.run(query, parameters);
142+
}
143+
144+
@Override
145+
public RxResult run(String query, Record parameters) {
146+
return delegate.run(query, parameters);
147+
}
148+
149+
@Override
150+
public RxResult run(String query) {
151+
return delegate.run(query);
152+
}
153+
154+
@Override
155+
public RxResult run(Query query) {
156+
return delegate.run(query);
157+
}
76158
}
77159

78160
<T> Mono<T> doInQueryRunnerForMono(final String targetDatabase, Function<RxQueryRunner, Mono<T>> func) {
79161

80162
return Mono.usingWhen(retrieveRxStatementRunnerHolder(targetDatabase),
81-
holder -> func.apply(holder.getRxQueryRunner()), RxStatementRunnerHolder::getCommit,
82-
(holder, ex) -> holder.getRollback(), RxStatementRunnerHolder::getCommit);
163+
func::apply,
164+
runner -> ((DelegatingQueryRunner) runner).close());
83165
}
84166

85167
<T> Flux<T> doInStatementRunnerForFlux(final String targetDatabase, Function<RxQueryRunner, Flux<T>> func) {
86168

87169
return Flux.usingWhen(retrieveRxStatementRunnerHolder(targetDatabase),
88-
holder -> func.apply(holder.getRxQueryRunner()), RxStatementRunnerHolder::getCommit,
89-
(holder, ex) -> holder.getRollback(), RxStatementRunnerHolder::getCommit);
170+
func::apply,
171+
runner -> ((DelegatingQueryRunner) runner).close());
90172
}
91173

92174
@Override

‎src/test/java/org/springframework/data/neo4j/core/Neo4jClientTest.java

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -47,8 +47,10 @@
4747
import org.junit.jupiter.api.extension.ExtendWith;
4848
import org.mockito.ArgumentCaptor;
4949
import org.mockito.Mock;
50+
import org.mockito.Mockito;
5051
import org.mockito.hamcrest.MockitoHamcrest;
5152
import org.mockito.junit.jupiter.MockitoExtension;
53+
import org.neo4j.driver.Bookmark;
5254
import org.neo4j.driver.Driver;
5355
import org.neo4j.driver.Record;
5456
import org.neo4j.driver.Result;
@@ -84,6 +86,8 @@ void prepareMocks() {
8486

8587
when(driver.session(any(SessionConfig.class))).thenReturn(session);
8688
when(driver.defaultTypeSystem()).thenReturn(typeSystem);
89+
90+
when(session.lastBookmark()).thenReturn(Mockito.mock(Bookmark.class));
8791
}
8892

8993
@AfterEach

0 commit comments

Comments
 (0)