GH-2755 - Use bookmark manager in Client.

Also make the `AbstractNeo4jConfig` and its reactive counter-part
use the same defined bookmark manager in transaction manager and
client.

Closes #2755

(cherry picked from commit 21c8a42)

Author: meistermeier

pom.xml

@@ -72,6 +72,7 @@
 		<archunit.version>0.23.1</archunit.version>
 		<asciidoctor-maven-plugin.version>2.1.0</asciidoctor-maven-plugin.version>
 		<asciidoctorj-diagram.version>2.1.0</asciidoctorj-diagram.version>
+		<blockhound.version>1.0.8.RELEASE</blockhound.version>
 		<byte-buddy.version>1.14.3</byte-buddy.version>
 		<cdi>3.0.1</cdi>
 		<checkstyle.skip>${skipTests}</checkstyle.skip>
@@ -237,6 +238,11 @@
 				<type>pom</type>
 				<scope>import</scope>
 			</dependency>
+			<dependency>
+				<groupId>io.projectreactor.tools</groupId>
+				<artifactId>blockhound</artifactId>
+				<version>${blockhound.version}</version>
+			</dependency>
 		</dependencies>
 	</dependencyManagement>
 
@@ -449,6 +455,11 @@
 				</exclusion>
 			</exclusions>
 		</dependency>
+		<dependency>
+			<groupId>io.projectreactor.tools</groupId>
+			<artifactId>blockhound</artifactId>
+			<scope>test</scope>
+		</dependency>
 	</dependencies>
 
 	<repositories>

src/main/java/org/springframework/data/neo4j/config/AbstractNeo4jConfig.java

@@ -27,6 +27,7 @@
 import org.springframework.data.neo4j.core.Neo4jTemplate;
 import org.springframework.data.neo4j.core.UserSelectionProvider;
 import org.springframework.data.neo4j.core.mapping.Neo4jMappingContext;
+import org.springframework.data.neo4j.core.transaction.Neo4jBookmarkManager;
 import org.springframework.data.neo4j.core.transaction.Neo4jTransactionManager;
 import org.springframework.data.neo4j.repository.config.Neo4jRepositoryConfigurationExtension;
 import org.springframework.lang.Nullable;
@@ -47,6 +48,9 @@ public abstract class AbstractNeo4jConfig extends Neo4jConfigurationSupport {
 	@Autowired
 	private ObjectProvider<UserSelectionProvider> userSelectionProviders;
 
+	@Autowired
+	private Neo4jBookmarkManager bookmarkManager;
+
 	/**
 	 * The driver to be used for interacting with Neo4j.
 	 *
@@ -66,6 +70,7 @@ public Neo4jClient neo4jClient(Driver driver, DatabaseSelectionProvider database
 		return Neo4jClient.with(driver)
 				.withDatabaseSelectionProvider(databaseSelectionProvider)
 				.withUserSelectionProvider(getUserSelectionProvider())
+				.withNeo4jBookmarkManager(bookmarkManager)
 				.build();
 	}
 
@@ -94,9 +99,15 @@ public PlatformTransactionManager transactionManager(Driver driver, DatabaseSele
 				.with(driver)
 				.withDatabaseSelectionProvider(databaseSelectionProvider)
 				.withUserSelectionProvider(getUserSelectionProvider())
+				.withBookmarkManager(bookmarkManager)
 				.build();
 	}
 
+	@Bean
+	public Neo4jBookmarkManager bookmarkManager() {
+		return Neo4jBookmarkManager.create();
+	}
+
 	/**
 	 * Configures the database selection provider.
 	 *

src/main/java/org/springframework/data/neo4j/config/AbstractReactiveNeo4jConfig.java

@@ -26,6 +26,7 @@
 import org.springframework.data.neo4j.core.ReactiveNeo4jTemplate;
 import org.springframework.data.neo4j.core.ReactiveUserSelectionProvider;
 import org.springframework.data.neo4j.core.mapping.Neo4jMappingContext;
+import org.springframework.data.neo4j.core.transaction.Neo4jBookmarkManager;
 import org.springframework.data.neo4j.core.transaction.ReactiveNeo4jTransactionManager;
 import org.springframework.data.neo4j.repository.config.ReactiveNeo4jRepositoryConfigurationExtension;
 import org.springframework.lang.Nullable;
@@ -47,6 +48,9 @@ public abstract class AbstractReactiveNeo4jConfig extends Neo4jConfigurationSupp
 	@Autowired
 	private ObjectProvider<ReactiveUserSelectionProvider> userSelectionProviders;
 
+	@Autowired
+	private Neo4jBookmarkManager bookmarkManager;
+
 	/**
 	 * The driver to be used for interacting with Neo4j.
 	 *
@@ -66,6 +70,7 @@ public ReactiveNeo4jClient neo4jClient(Driver driver, ReactiveDatabaseSelectionP
 		return ReactiveNeo4jClient.with(driver)
 				.withDatabaseSelectionProvider(databaseSelectionProvider)
 				.withUserSelectionProvider(getUserSelectionProvider())
+				.withNeo4jBookmarkManager(bookmarkManager)
 				.build();
 	}
 
@@ -94,9 +99,15 @@ public ReactiveTransactionManager reactiveTransactionManager(Driver driver,
 		return ReactiveNeo4jTransactionManager.with(driver)
 				.withDatabaseSelectionProvider(databaseSelectionProvider)
 				.withUserSelectionProvider(getUserSelectionProvider())
+				.withBookmarkManager(bookmarkManager)
 				.build();
 	}
 
+	@Bean
+	public Neo4jBookmarkManager bookmarkManager() {
+		return Neo4jBookmarkManager.createReactive();
+	}
+
 	/**
 	 * Configures the database name provider.
 	 *

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

@@ -16,13 +16,9 @@
 package org.springframework.data.neo4j.core;
 
 import java.util.Collection;
-import java.util.Collections;
-import java.util.HashSet;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
-import java.util.Set;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
 import java.util.function.BiConsumer;
 import java.util.function.BiFunction;
 import java.util.function.Function;
@@ -45,6 +41,7 @@
 import org.springframework.dao.DataAccessException;
 import org.springframework.dao.support.PersistenceExceptionTranslator;
 import org.springframework.data.neo4j.core.convert.Neo4jConversions;
+import org.springframework.data.neo4j.core.transaction.Neo4jBookmarkManager;
 import org.springframework.data.neo4j.core.transaction.Neo4jTransactionManager;
 import org.springframework.data.neo4j.core.transaction.Neo4jTransactionUtils;
 import org.springframework.lang.Nullable;
@@ -67,15 +64,15 @@ final class DefaultNeo4jClient implements Neo4jClient {
 	private final ConversionService conversionService;
 	private final Neo4jPersistenceExceptionTranslator persistenceExceptionTranslator = new Neo4jPersistenceExceptionTranslator();
 
-	// Basically a local bookmark manager
-	private final Set<Bookmark> bookmarks = new HashSet<>();
-	private final ReentrantReadWriteLock bookmarksLock = new ReentrantReadWriteLock();
+	// Local bookmark manager when using outside managed transactions
+	private final Neo4jBookmarkManager bookmarkManager;
 
 	DefaultNeo4jClient(Builder builder) {
 
 		this.driver = builder.driver;
 		this.databaseSelectionProvider = builder.databaseSelectionProvider;
 		this.userSelectionProvider = builder.userSelectionProvider;
+		this.bookmarkManager = builder.bookmarkManager != null ? builder.bookmarkManager : Neo4jBookmarkManager.create();
 
 		this.conversionService = new DefaultConversionService();
 		Optional.ofNullable(builder.neo4jConversions).orElseGet(Neo4jConversions::new).registerConvertersIn((ConverterRegistry) conversionService);
@@ -85,29 +82,13 @@ final class DefaultNeo4jClient implements Neo4jClient {
 	public QueryRunner getQueryRunner(DatabaseSelection databaseSelection, UserSelection impersonatedUser) {
 
 		QueryRunner queryRunner = Neo4jTransactionManager.retrieveTransaction(driver, databaseSelection, impersonatedUser);
-		Collection<Bookmark> lastBookmarks = Collections.emptySet();
+		Collection<Bookmark> lastBookmarks = bookmarkManager.getBookmarks();
+
 		if (queryRunner == null) {
-			ReentrantReadWriteLock.ReadLock lock = bookmarksLock.readLock();
-			try {
-				lock.lock();
-				lastBookmarks = new HashSet<>(bookmarks);
-				queryRunner = driver.session(Neo4jTransactionUtils.sessionConfig(false, lastBookmarks, databaseSelection, impersonatedUser));
-			} finally {
-				lock.unlock();
-			}
+			queryRunner = driver.session(Neo4jTransactionUtils.sessionConfig(false, lastBookmarks, databaseSelection, impersonatedUser));
 		}
 
-		return new DelegatingQueryRunner(queryRunner, lastBookmarks, (usedBookmarks, newBookmarks) -> {
-
-			ReentrantReadWriteLock.WriteLock lock = bookmarksLock.writeLock();
-			try {
-				lock.lock();
-				bookmarks.removeAll(usedBookmarks);
-				bookmarks.addAll(newBookmarks);
-			} finally {
-				lock.unlock();
-			}
-		});
+		return new DelegatingQueryRunner(queryRunner, lastBookmarks, bookmarkManager::updateBookmarks);
 	}
 
 	private static class DelegatingQueryRunner implements QueryRunner {

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

@@ -31,6 +31,7 @@
 import org.springframework.core.convert.support.DefaultConversionService;
 import org.springframework.dao.DataAccessException;
 import org.springframework.data.neo4j.core.convert.Neo4jConversions;
+import org.springframework.data.neo4j.core.transaction.Neo4jBookmarkManager;
 import org.springframework.data.neo4j.core.transaction.Neo4jTransactionUtils;
 import org.springframework.data.neo4j.core.transaction.ReactiveNeo4jTransactionManager;
 import org.springframework.lang.Nullable;
@@ -43,12 +44,8 @@
 import reactor.util.function.Tuples;
 
 import java.util.Collection;
-import java.util.Collections;
-import java.util.HashSet;
 import java.util.Map;
 import java.util.Optional;
-import java.util.Set;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
 import java.util.function.BiConsumer;
 import java.util.function.BiFunction;
 import java.util.function.Function;
@@ -70,9 +67,8 @@ final class DefaultReactiveNeo4jClient implements ReactiveNeo4jClient {
 	private final ConversionService conversionService;
 	private final Neo4jPersistenceExceptionTranslator persistenceExceptionTranslator = new Neo4jPersistenceExceptionTranslator();
 
-	// Basically a local bookmark manager
-	private final Set<Bookmark> bookmarks = new HashSet<>();
-	private final ReentrantReadWriteLock bookmarksLock = new ReentrantReadWriteLock();
+	// Local bookmark manager when using outside managed transactions
+	private final Neo4jBookmarkManager bookmarkManager;
 
 	DefaultReactiveNeo4jClient(Builder builder) {
 
@@ -82,6 +78,7 @@ final class DefaultReactiveNeo4jClient implements ReactiveNeo4jClient {
 
 		this.conversionService = new DefaultConversionService();
 		Optional.ofNullable(builder.neo4jConversions).orElseGet(Neo4jConversions::new).registerConvertersIn((ConverterRegistry) conversionService);
+		this.bookmarkManager = builder.bookmarkManager != null ? builder.bookmarkManager : Neo4jBookmarkManager.createReactive();
 	}
 
 	@Override
@@ -91,27 +88,12 @@ public Mono<ReactiveQueryRunner> getQueryRunner(Mono<DatabaseSelection> database
 				.flatMap(targetDatabaseAndUser ->
 						ReactiveNeo4jTransactionManager.retrieveReactiveTransaction(driver, targetDatabaseAndUser.getT1(), targetDatabaseAndUser.getT2())
 								.map(ReactiveQueryRunner.class::cast)
-								.zipWith(Mono.just(Collections.<Bookmark>emptySet()))
+								.zipWith(Mono.just(bookmarkManager.getBookmarks()))
 								.switchIfEmpty(Mono.fromSupplier(() -> {
-									ReentrantReadWriteLock.ReadLock lock = bookmarksLock.readLock();
-									try {
-										lock.lock();
-										Set<Bookmark> lastBookmarks = new HashSet<>(bookmarks);
-										return Tuples.of(driver.session(ReactiveSession.class, Neo4jTransactionUtils.sessionConfig(false, lastBookmarks, targetDatabaseAndUser.getT1(), targetDatabaseAndUser.getT2())), lastBookmarks);
-									} finally {
-										lock.unlock();
-									}
+									Collection<Bookmark> lastBookmarks = bookmarkManager.getBookmarks();
+									return Tuples.of(driver.session(ReactiveSession.class, Neo4jTransactionUtils.sessionConfig(false, lastBookmarks, targetDatabaseAndUser.getT1(), targetDatabaseAndUser.getT2())), lastBookmarks);
 								})))
-				.map(t -> new DelegatingQueryRunner(t.getT1(), t.getT2(), (usedBookmarks, newBookmarks) -> {
-					ReentrantReadWriteLock.WriteLock lock = bookmarksLock.writeLock();
-					try {
-						lock.lock();
-						bookmarks.removeAll(usedBookmarks);
-						bookmarks.addAll(newBookmarks);
-					} finally {
-						lock.unlock();
-					}
-				}));
+				.map(t -> new DelegatingQueryRunner(t.getT1(), t.getT2(), bookmarkManager::updateBookmarks));
 	}
 
 	private static class DelegatingQueryRunner implements ReactiveQueryRunner {

src/main/java/org/springframework/data/neo4j/core/Neo4jClient.java

@@ -31,6 +31,7 @@
 import org.neo4j.driver.types.TypeSystem;
 import org.springframework.core.log.LogAccessor;
 import org.springframework.data.neo4j.core.convert.Neo4jConversions;
+import org.springframework.data.neo4j.core.transaction.Neo4jBookmarkManager;
 import org.springframework.lang.Nullable;
 
 /**
@@ -79,6 +80,9 @@ class Builder {
 		@Nullable
 		Neo4jConversions neo4jConversions;
 
+		@Nullable
+		Neo4jBookmarkManager bookmarkManager;
+
 		private Builder(Driver driver) {
 			this.driver = driver;
 		}
@@ -121,6 +125,20 @@ public Builder withNeo4jConversions(Neo4jConversions neo4jConversions) {
 			return this;
 		}
 
+		/**
+		 * Configures the {@link Neo4jBookmarkManager} to use.
+		 * This should be the same instance as provided for the {@link org.springframework.data.neo4j.core.transaction.Neo4jTransactionManager}
+		 * respectively the {@link org.springframework.data.neo4j.core.transaction.ReactiveNeo4jTransactionManager}.
+		 *
+		 * @param bookmarkManager Neo4jBookmarkManager instance that is shared with the transaction manager.
+		 * @return The builder
+		 * @since 7.1.2
+		 */
+		public Builder withNeo4jBookmarkManager(Neo4jBookmarkManager bookmarkManager) {
+			this.bookmarkManager = bookmarkManager;
+			return this;
+		}
+
 		public Neo4jClient build() {
 			return new DefaultNeo4jClient(this);
 		}

src/main/java/org/springframework/data/neo4j/core/ReactiveNeo4jClient.java

@@ -15,6 +15,7 @@
  */
 package org.springframework.data.neo4j.core;
 
+import org.springframework.data.neo4j.core.transaction.Neo4jBookmarkManager;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
@@ -82,6 +83,9 @@ class Builder {
 		@Nullable
 		Neo4jConversions neo4jConversions;
 
+		@Nullable
+		Neo4jBookmarkManager bookmarkManager;
+
 		private Builder(Driver driver) {
 			this.driver = driver;
 		}
@@ -124,6 +128,20 @@ public Builder withNeo4jConversions(Neo4jConversions neo4jConversions) {
 			return this;
 		}
 
+		/**
+		 * Configures the {@link Neo4jBookmarkManager} to use.
+		 * This should be the same instance as provided for the {@link org.springframework.data.neo4j.core.transaction.Neo4jTransactionManager}
+		 * respectively the {@link org.springframework.data.neo4j.core.transaction.ReactiveNeo4jTransactionManager}.
+		 *
+		 * @param bookmarkManager Neo4jBookmarkManager instance that is shared with the transaction manager.
+		 * @return The builder
+		 * @since 7.1.2
+		 */
+		public Builder withNeo4jBookmarkManager(Neo4jBookmarkManager bookmarkManager) {
+			this.bookmarkManager = bookmarkManager;
+			return this;
+		}
+
 		public ReactiveNeo4jClient build() {
 			return new DefaultReactiveNeo4jClient(this);
 		}

src/main/java/org/springframework/data/neo4j/core/transaction/Neo4jBookmarkManager.java

@@ -41,6 +41,13 @@ static Neo4jBookmarkManager create() {
 		return new DefaultBookmarkManager(null);
 	}
 
+	/**
+	 * @return default reactive version of bookmark manager
+	 */
+	static Neo4jBookmarkManager createReactive() {
+		return new ReactiveDefaultBookmarkManager(null);
+	}
+
 	/**
 	 * Use this factory method to add supplier of initial "seeding" bookmarks to the transaction managers
 	 * <p>
@@ -55,6 +62,20 @@ static Neo4jBookmarkManager create(@Nullable Supplier<Set<Bookmark>> bookmarksSu
 		return new DefaultBookmarkManager(bookmarksSupplier);
 	}
 
+	/**
+	 * Use this factory method to add supplier of initial "seeding" bookmarks to the transaction managers
+	 * <p>
+	 * While this class will make sure that the supplier will be accessed in a thread-safe manner,
+	 * it is the caller's duty to provide a thread safe supplier (not changing the seed during a call, etc.).
+	 *
+	 * @param bookmarksSupplier A supplier for seeding bookmarks, can be null. The supplier is free to provide different
+	 *                          bookmarks on each call.
+	 * @return A reactive bookmark manager
+	 */
+	static Neo4jBookmarkManager createReactive(@Nullable Supplier<Set<Bookmark>> bookmarksSupplier) {
+		return new ReactiveDefaultBookmarkManager(bookmarksSupplier);
+	}
+
 	/**
 	 * Use this bookmark manager at your own risk, it will effectively disable any bookmark management by dropping all
 	 * bookmarks and never  supplying any. In a cluster you will be at a high risk of experiencing stale reads. In a single