Introduce new Reactive Streams session

This update introduces support for reactive session with Reactive Streams types. While it is similar to the deprecated `RxSession`, it includes the improvements introduced with the `ReactiveSession` that uses Flow API types.

Sample session creation:
```
var session = driver.reactiveSession(org.neo4j.driver.reactivestreams.ReactiveSession.class);
```

driver/clirr-ignored-differences.xml

@@ -397,4 +397,16 @@
  <method>org.neo4j.driver.types.TypeSystem getDefault()</method>
  </difference>
 
+ <difference>
+ <className>org/neo4j/driver/Driver</className>
+ <differenceType>7012</differenceType>
+ <method>org.neo4j.driver.BaseReactiveSession reactiveSession(java.lang.Class)</method>
+ </difference>
+
+ <difference>
+ <className>org/neo4j/driver/Driver</className>
+ <differenceType>7012</differenceType>
+ <method>org.neo4j.driver.BaseReactiveSession reactiveSession(java.lang.Class, org.neo4j.driver.SessionConfig)</method>
+ </difference>
+
 </differences>

driver/src/main/java/module-info.java

@@ -21,6 +21,7 @@
  exports org.neo4j.driver;
  exports org.neo4j.driver.async;
  exports org.neo4j.driver.reactive;
+ exports org.neo4j.driver.reactivestreams;
  exports org.neo4j.driver.types;
  exports org.neo4j.driver.summary;
  exports org.neo4j.driver.net;

driver/src/main/java/org/neo4j/driver/BaseReactiveSession.java

@@ -0,0 +1,24 @@
+/*
+ * Copyright (c) "Neo4j"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * 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
+ *
+ * http://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.neo4j.driver;
+
+/**
+ * A base interface for reactive sessions, used by {@link Driver#reactiveSession(Class)} and {@link Driver#reactiveSession(Class, SessionConfig)}.
+ */
+public interface BaseReactiveSession {}

driver/src/main/java/org/neo4j/driver/Driver.java

@@ -77,7 +77,9 @@ public interface Driver extends AutoCloseable {
  *
  * @return a new {@link Session} object.
  */
- Session session();
+ default Session session() {
+ return session(SessionConfig.defaultConfig());
+ }
 
  /**
  * Create a new {@link Session} with a specified {@link SessionConfig session configuration}.
@@ -133,7 +135,41 @@ default ReactiveSession reactiveSession() {
  * @param sessionConfig used to customize the session.
  * @return a new {@link ReactiveSession} object.
  */
- ReactiveSession reactiveSession(SessionConfig sessionConfig);
+ default ReactiveSession reactiveSession(SessionConfig sessionConfig) {
+ return reactiveSession(ReactiveSession.class, sessionConfig);
+ }
+
+ /**
+ * Create a new reactive session of supported type with default {@link SessionConfig session configuration}.
+ * <p>
+ * Supported types are:
+ * <ul>
+ * <li>{@link org.neo4j.driver.reactive.ReactiveSession} - reactive session using Flow API</li>
+ * <li>{@link org.neo4j.driver.reactivestreams.ReactiveSession} - reactive session using Reactive Streams API</li>
+ * </ul>
+ *
+ * @param sessionClass session type class
+ * @return session instance
+ * @param <T> session type
+ */
+ default <T extends BaseReactiveSession> T reactiveSession(Class<T> sessionClass) {
+ return reactiveSession(sessionClass, SessionConfig.defaultConfig());
+ }
+
+ /**
+ * Create a new reactive session of supported type with a specified {@link SessionConfig session configuration}.
+ * <p>
+ * Supported types are:
+ * <ul>
+ * <li>{@link org.neo4j.driver.reactive.ReactiveSession} - reactive session using Flow API</li>
+ * <li>{@link org.neo4j.driver.reactivestreams.ReactiveSession} - reactive session using Reactive Streams API</li>
+ * </ul>
+ *
+ * @param sessionClass session type class
+ * @return session instance
+ * @param <T> session type
+ */
+ <T extends BaseReactiveSession> T reactiveSession(Class<T> sessionClass, SessionConfig sessionConfig);
 
  /**
  * Create a new general purpose {@link AsyncSession} with default {@link SessionConfig session configuration}. The {@link AsyncSession} provides an
@@ -143,7 +179,9 @@ default ReactiveSession reactiveSession() {
  *
  * @return a new {@link AsyncSession} object.
  */
- AsyncSession asyncSession();
+ default AsyncSession asyncSession() {
+ return asyncSession(SessionConfig.defaultConfig());
+ }
 
  /**
  * Create a new {@link AsyncSession} with a specified {@link SessionConfig session configuration}.

driver/src/main/java/org/neo4j/driver/internal/InternalDriver.java

@@ -18,10 +18,12 @@
  */
 package org.neo4j.driver.internal;
 
+import static java.util.Objects.requireNonNull;
 import static org.neo4j.driver.internal.util.Futures.completedWithNull;
 
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.atomic.AtomicBoolean;
+import org.neo4j.driver.BaseReactiveSession;
 import org.neo4j.driver.Driver;
 import org.neo4j.driver.Logger;
 import org.neo4j.driver.Logging;
@@ -33,12 +35,10 @@
 import org.neo4j.driver.internal.async.NetworkSession;
 import org.neo4j.driver.internal.metrics.DevNullMetricsProvider;
 import org.neo4j.driver.internal.metrics.MetricsProvider;
-import org.neo4j.driver.internal.reactive.InternalReactiveSession;
 import org.neo4j.driver.internal.reactive.InternalRxSession;
 import org.neo4j.driver.internal.security.SecurityPlan;
 import org.neo4j.driver.internal.types.InternalTypeSystem;
 import org.neo4j.driver.internal.util.Futures;
-import org.neo4j.driver.reactive.ReactiveSession;
 import org.neo4j.driver.reactive.RxSession;
 import org.neo4j.driver.types.TypeSystem;
 
@@ -61,11 +61,6 @@ public class InternalDriver implements Driver {
  this.log = logging.getLog(getClass());
  }
 
- @Override
- public Session session() {
- return new InternalSession(newSession(SessionConfig.defaultConfig()));
- }
-
  @Override
  public Session session(SessionConfig sessionConfig) {
  return new InternalSession(newSession(sessionConfig));
@@ -77,14 +72,19 @@ public RxSession rxSession(SessionConfig sessionConfig) {
  return new InternalRxSession(newSession(sessionConfig));
  }
 
+ @SuppressWarnings({"deprecation", "unchecked"})
  @Override
- public ReactiveSession reactiveSession(SessionConfig sessionConfig) {
- return new InternalReactiveSession(newSession(sessionConfig));
- }
-
- @Override
- public AsyncSession asyncSession() {
- return new InternalAsyncSession(newSession(SessionConfig.defaultConfig()));
+ public <T extends BaseReactiveSession> T reactiveSession(Class<T> sessionClass, SessionConfig sessionConfig) {
+ requireNonNull(sessionClass, "sessionClass must not be null");
+ requireNonNull(sessionClass, "sessionConfig must not be null");
+ if (org.neo4j.driver.reactive.ReactiveSession.class.isAssignableFrom(sessionClass)) {
+ return (T) new org.neo4j.driver.internal.reactive.InternalReactiveSession(newSession(sessionConfig));
+ } else if (org.neo4j.driver.reactivestreams.ReactiveSession.class.isAssignableFrom(sessionClass)) {
+ return (T) new org.neo4j.driver.internal.reactivestreams.InternalReactiveSession(newSession(sessionConfig));
+ } else {
+ throw new IllegalArgumentException(
+ String.format("Unsupported session type '%s'", sessionClass.getCanonicalName()));
+ }
  }
 
  @Override

driver/src/main/java/org/neo4j/driver/internal/reactive/AbstractReactiveSession.java

@@ -34,7 +34,7 @@
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
 
-abstract class AbstractReactiveSession<S> {
+public abstract class AbstractReactiveSession<S> {
  protected final NetworkSession session;
 
  public AbstractReactiveSession(NetworkSession session) {
@@ -45,15 +45,15 @@ public AbstractReactiveSession(NetworkSession session) {
  this.session = session;
  }
 
- abstract S createTransaction(UnmanagedTransaction unmanagedTransaction);
+ protected abstract S createTransaction(UnmanagedTransaction unmanagedTransaction);
 
- abstract Publisher<Void> closeTransaction(S transaction, boolean commit);
+ protected abstract Publisher<Void> closeTransaction(S transaction, boolean commit);
 
  Publisher<S> doBeginTransaction(TransactionConfig config) {
  return doBeginTransaction(config, null);
  }
 
- Publisher<S> doBeginTransaction(TransactionConfig config, String txType) {
+ protected Publisher<S> doBeginTransaction(TransactionConfig config, String txType) {
  return createSingleItemPublisher(
  () -> {
  CompletableFuture<S> txFuture = new CompletableFuture<>();
@@ -87,7 +87,7 @@ Publisher<S> beginTransaction(AccessMode mode, TransactionConfig config) {
  "Unexpected condition, begin transaction call has completed successfully with transaction being null"));
  }
 
- <T> Publisher<T> runTransaction(
+ protected <T> Publisher<T> runTransaction(
  AccessMode mode, Function<S, ? extends Publisher<T>> work, TransactionConfig config) {
  Flux<T> repeatableWork = Flux.usingWhen(
  beginTransaction(mode, config),
@@ -119,7 +119,7 @@ public Set<Bookmark> lastBookmarks() {
  return session.lastBookmarks();
  }
 
- <T> Publisher<T> doClose() {
+ protected <T> Publisher<T> doClose() {
  return createEmptyPublisher(session::closeAsync);
  }
 }

driver/src/main/java/org/neo4j/driver/internal/reactive/AbstractReactiveTransaction.java

@@ -24,30 +24,30 @@
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Mono;
 
-abstract class AbstractReactiveTransaction {
+public abstract class AbstractReactiveTransaction {
  protected final UnmanagedTransaction tx;
 
  protected AbstractReactiveTransaction(UnmanagedTransaction tx) {
  this.tx = tx;
  }
 
- <T> Publisher<T> doCommit() {
+ protected <T> Publisher<T> doCommit() {
  return createEmptyPublisher(tx::commitAsync);
  }
 
- <T> Publisher<T> doRollback() {
+ protected <T> Publisher<T> doRollback() {
  return createEmptyPublisher(tx::rollbackAsync);
  }
 
- Publisher<Void> doClose() {
+ protected Publisher<Void> doClose() {
  return close(false);
  }
 
- Publisher<Boolean> doIsOpen() {
+ protected Publisher<Boolean> doIsOpen() {
  return Mono.just(tx.isOpen());
  }
 
- Publisher<Void> close(boolean commit) {
+ public Publisher<Void> close(boolean commit) {
  return createEmptyPublisher(() -> tx.closeAsync(commit));
  }
 }

driver/src/main/java/org/neo4j/driver/internal/reactive/InternalReactiveSession.java

@@ -46,12 +46,12 @@ public InternalReactiveSession(NetworkSession session) {
  }
 
  @Override
- ReactiveTransaction createTransaction(UnmanagedTransaction unmanagedTransaction) {
+ protected ReactiveTransaction createTransaction(UnmanagedTransaction unmanagedTransaction) {
  return new InternalReactiveTransaction(unmanagedTransaction);
  }
 
  @Override
- org.reactivestreams.Publisher<Void> closeTransaction(ReactiveTransaction transaction, boolean commit) {
+ protected org.reactivestreams.Publisher<Void> closeTransaction(ReactiveTransaction transaction, boolean commit) {
  return ((InternalReactiveTransaction) transaction).close(commit);
  }
 

driver/src/main/java/org/neo4j/driver/internal/reactive/InternalRxSession.java

@@ -43,12 +43,12 @@ public InternalRxSession(NetworkSession session) {
  }
 
  @Override
- RxTransaction createTransaction(UnmanagedTransaction unmanagedTransaction) {
+ protected RxTransaction createTransaction(UnmanagedTransaction unmanagedTransaction) {
  return new InternalRxTransaction(unmanagedTransaction);
  }
 
  @Override
- Publisher<Void> closeTransaction(RxTransaction transaction, boolean commit) {
+ protected Publisher<Void> closeTransaction(RxTransaction transaction, boolean commit) {
  return ((InternalRxTransaction) transaction).close(commit);
  }
 

driver/src/main/java/org/neo4j/driver/internal/reactivestreams/BaseReactiveQueryRunner.java

@@ -0,0 +1,74 @@
+/*
+ * Copyright (c) "Neo4j"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * 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
+ *
+ * http://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.neo4j.driver.internal.reactivestreams;
+
+import java.util.Map;
+import org.neo4j.driver.Query;
+import org.neo4j.driver.Record;
+import org.neo4j.driver.Value;
+import org.neo4j.driver.Values;
+import org.neo4j.driver.internal.util.Extract;
+import org.neo4j.driver.internal.value.MapValue;
+import org.neo4j.driver.reactivestreams.ReactiveQueryRunner;
+import org.neo4j.driver.reactivestreams.ReactiveResult;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Mono;
+
+interface BaseReactiveQueryRunner extends ReactiveQueryRunner {
+ @Override
+ default Publisher<ReactiveResult> run(String queryStr, Value parameters) {
+ try {
+ Query query = new Query(queryStr, parameters);
+ return run(query);
+ } catch (Throwable t) {
+ return Mono.error(t);
+ }
+ }
+
+ @Override
+ default Publisher<ReactiveResult> run(String query, Map<String, Object> parameters) {
+ return run(query, parameters(parameters));
+ }
+
+ @Override
+ default Publisher<ReactiveResult> run(String query, Record parameters) {
+ return run(query, parameters(parameters));
+ }
+
+ @Override
+ default Publisher<ReactiveResult> run(String queryStr) {
+ try {
+ Query query = new Query(queryStr);
+ return run(query);
+ } catch (Throwable t) {
+ return Mono.error(t);
+ }
+ }
+
+ static Value parameters(Record record) {
+ return record == null ? Values.EmptyMap : parameters(record.asMap());
+ }
+
+ static Value parameters(Map<String, Object> map) {
+ if (map == null || map.isEmpty()) {
+ return Values.EmptyMap;
+ }
+ return new MapValue(Extract.mapOfValues(map));
+ }
+}