diff --git a/pom.xml b/pom.xml index 3952055e5..b743e8b9b 100644 --- a/pom.xml +++ b/pom.xml @@ -102,7 +102,7 @@ ${java.version} ${mockito.version} 3.10.0 - 4.4.9 + 4.4.10 4.3.6 3.0.1 ${project.build.directory}/docs diff --git a/src/test/java/org/springframework/data/neo4j/integration/issues/gh2632/Movie.java b/src/test/java/org/springframework/data/neo4j/integration/issues/gh2632/Movie.java new file mode 100644 index 000000000..0de4b02b4 --- /dev/null +++ b/src/test/java/org/springframework/data/neo4j/integration/issues/gh2632/Movie.java @@ -0,0 +1,46 @@ +/* + * Copyright 2011-2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.neo4j.integration.issues.gh2632; + +import java.util.UUID; + +import org.springframework.data.neo4j.core.schema.GeneratedValue; +import org.springframework.data.neo4j.core.schema.Id; +import org.springframework.data.neo4j.core.schema.Node; + +/** + * @author Michael J. Simons + */ +@Node +public class Movie { + + @Id @GeneratedValue + private UUID id; + + private String title; + + public UUID getId() { + return id; + } + + public String getTitle() { + return title; + } + + public void setTitle(String title) { + this.title = title; + } +} diff --git a/src/test/java/org/springframework/data/neo4j/integration/issues/gh2632/MovieRepository.java b/src/test/java/org/springframework/data/neo4j/integration/issues/gh2632/MovieRepository.java new file mode 100644 index 000000000..f58bd6d60 --- /dev/null +++ b/src/test/java/org/springframework/data/neo4j/integration/issues/gh2632/MovieRepository.java @@ -0,0 +1,26 @@ +/* + * Copyright 2011-2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.neo4j.integration.issues.gh2632; + +import java.util.UUID; + +import org.springframework.data.neo4j.repository.ReactiveNeo4jRepository; + +/** + * @author Michael J. Simons + */ +public interface MovieRepository extends ReactiveNeo4jRepository { +} diff --git a/src/test/java/org/springframework/data/neo4j/integration/issues/gh2632/ReactiveConnectionAcquisitionIT.java b/src/test/java/org/springframework/data/neo4j/integration/issues/gh2632/ReactiveConnectionAcquisitionIT.java new file mode 100644 index 000000000..19621a54a --- /dev/null +++ b/src/test/java/org/springframework/data/neo4j/integration/issues/gh2632/ReactiveConnectionAcquisitionIT.java @@ -0,0 +1,152 @@ +/* + * Copyright 2011-2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.neo4j.integration.issues.gh2632; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.util.Collections; +import java.util.UUID; +import java.util.concurrent.TimeUnit; + +import org.junit.jupiter.api.Test; +import org.neo4j.driver.Driver; +import org.neo4j.driver.GraphDatabase; +import org.neo4j.driver.Query; +import org.neo4j.driver.Session; +import org.neo4j.driver.reactive.RxResult; +import org.neo4j.driver.reactive.RxSession; +import org.neo4j.driver.reactive.RxTransaction; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.data.neo4j.config.AbstractReactiveNeo4jConfig; +import org.springframework.data.neo4j.repository.config.EnableReactiveNeo4jRepositories; +import org.springframework.data.neo4j.test.Neo4jExtension; +import org.springframework.data.neo4j.test.Neo4jIntegrationTest; +import org.springframework.transaction.annotation.EnableTransactionManagement; + +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.test.StepVerifier; + +/** + * @author Michael J. Simons + */ +@Neo4jIntegrationTest +class ReactiveConnectionAcquisitionIT { + + protected static Neo4jExtension.Neo4jConnectionSupport neo4jConnectionSupport; + + @Test // GH-2632 + void connectionAcquisitionAfterErrorViaSDNTxManagerShouldWork(@Autowired MovieRepository movieRepository, @Autowired Driver driver) { + UUID id = UUID.randomUUID(); + Flux + .range(1, 5) + .flatMap( + i -> movieRepository + .findById(id) + .switchIfEmpty(Mono.error(new RuntimeException())) + ) + .then() + .as(StepVerifier::create) + .verifyError(); + + try (Session session = driver.session()) { + long aNumber = session.run("RETURN 1").single().get(0).asLong(); + assertThat(aNumber).isOne(); + } + } + + @Test // GH-2632 + void connectionAcquisitionAfterErrorViaImplicitTXShouldWork(@Autowired Driver driver) { + Flux + .range(1, 5) + .flatMap( + i -> { + Query query = new Query("MATCH (p:Product) WHERE p.id = $id RETURN p.title", Collections.singletonMap("id", 0)); + return Flux.usingWhen( + Mono.fromSupplier(driver::rxSession), + session -> Mono.fromSupplier(() -> session.run(query)) + .flatMapMany(result -> Flux.from(result.records())) + .map(record -> record.get(0).asString()), + session -> Mono.fromDirect(session.close())) + .switchIfEmpty(Mono.error(new RuntimeException())); + } + ) + .then() + .as(StepVerifier::create) + .verifyError(); + + try (Session session = driver.session()) { + long aNumber = session.run("RETURN 1").single().get(0).asLong(); + assertThat(aNumber).isOne(); + } + } + + private static class SessionAndTx { + RxSession session; + RxTransaction tx; + + SessionAndTx(RxSession session, RxTransaction tx) { + this.session = session; + this.tx = tx; + } + } + + @Test // GH-2632 + void connectionAcquisitionAfterErrorViaExplicitTXShouldWork(@Autowired Driver driver) { + Flux + .range(1, 5) + .flatMap( + i -> { + Mono f = Mono + .just(driver.rxSession()) + .flatMap(s -> Mono.fromDirect(s.beginTransaction()).map(tx -> new SessionAndTx(s, tx))); + return Flux.usingWhen(f, + h -> Mono.fromSupplier(() -> h.tx.run("MATCH (n) WHERE false = true RETURN n")).flatMapMany(RxResult::records), + h -> Mono.from(h.tx.commit()).then(Mono.from(h.session.close())), + (h, e) -> Mono.from(h.tx.rollback()).then(Mono.from(h.session.close())), + h -> Mono.from(h.tx.rollback()).then(Mono.from(h.session.close())) + ).switchIfEmpty(Mono.error(new RuntimeException())); + } + ) + .then() + .as(StepVerifier::create) + .verifyError(); + + try (Session session = driver.session()) { + long aNumber = session.run("RETURN 1").single().get(0).asLong(); + assertThat(aNumber).isOne(); + } + } + + @Configuration + @EnableTransactionManagement + @EnableReactiveNeo4jRepositories(considerNestedRepositories = true) + static class Config extends AbstractReactiveNeo4jConfig { + + @Bean + public Driver driver() { + org.neo4j.driver.Config config = org.neo4j.driver.Config.builder() + .withMaxConnectionPoolSize(2) + .withConnectionAcquisitionTimeout(20, TimeUnit.SECONDS) + .withLeakedSessionsLogging() + .build(); + return + GraphDatabase.driver(neo4jConnectionSupport.url, neo4jConnectionSupport.authToken, config); + } + } +}