Skip to content

Commit

Permalink
Refactoring
Browse files Browse the repository at this point in the history
  • Loading branch information
injectives committed Dec 13, 2022
1 parent d56407b commit 0347029
Show file tree
Hide file tree
Showing 6 changed files with 107 additions and 151 deletions.
Original file line number Diff line number Diff line change
@@ -0,0 +1,73 @@
/*
* 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 neo4j.org.testkit.backend.messages.requests;

import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
import java.time.Duration;
import java.util.Map;
import java.util.Optional;
import lombok.Setter;
import neo4j.org.testkit.backend.CustomDriverError;
import neo4j.org.testkit.backend.messages.requests.deserializer.TestkitCypherParamDeserializer;
import org.neo4j.driver.TransactionConfig;

@Setter
abstract class AbstractTestkitRequestWithTransactionConfig<
T extends AbstractTestkitRequestWithTransactionConfig.TransactionConfigBody>
implements TestkitRequest {
protected T data;

protected TransactionConfig buildTxConfig() {
return configureTx(TransactionConfig.builder()).build();
}

private TransactionConfig.Builder configureTx(TransactionConfig.Builder builder) {
return configureTxMetadata(configureTxTimeout(builder));
}

private TransactionConfig.Builder configureTxMetadata(TransactionConfig.Builder builder) {
data.getTxMeta().ifPresent(builder::withMetadata);
return builder;
}

private TransactionConfig.Builder configureTxTimeout(TransactionConfig.Builder builder) {
try {
data.getTimeout().ifPresent((timeout) -> builder.withTimeout(Duration.ofMillis(timeout)));
} catch (IllegalArgumentException e) {
throw new CustomDriverError(e);
}
return builder;
}

@Setter
abstract static class TransactionConfigBody {
protected Integer timeout;

@JsonDeserialize(using = TestkitCypherParamDeserializer.class)
protected Map<String, Object> txMeta;

Optional<Integer> getTimeout() {
return Optional.ofNullable(timeout);
}

Optional<Map<String, Object>> getTxMeta() {
return Optional.ofNullable(txMeta);
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -20,8 +20,6 @@

import static reactor.adapter.JdkFlowAdapter.flowPublisherToFlux;

import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
import java.util.Map;
import java.util.concurrent.CompletionStage;
import lombok.Getter;
import lombok.Setter;
Expand All @@ -32,7 +30,6 @@
import neo4j.org.testkit.backend.holder.RxTransactionHolder;
import neo4j.org.testkit.backend.holder.SessionHolder;
import neo4j.org.testkit.backend.holder.TransactionHolder;
import neo4j.org.testkit.backend.messages.requests.deserializer.TestkitCypherParamDeserializer;
import neo4j.org.testkit.backend.messages.responses.TestkitResponse;
import neo4j.org.testkit.backend.messages.responses.Transaction;
import org.neo4j.driver.Session;
Expand All @@ -44,15 +41,14 @@

@Setter
@Getter
public class SessionBeginTransaction extends WithTxConfig {
private SessionBeginTransactionBody data;

public class SessionBeginTransaction
extends AbstractTestkitRequestWithTransactionConfig<SessionBeginTransaction.SessionBeginTransactionBody> {
@Override
public TestkitResponse process(TestkitState testkitState) {
SessionHolder sessionHolder = testkitState.getSessionHolder(data.getSessionId());
Session session = sessionHolder.getSession();

org.neo4j.driver.Transaction transaction = session.beginTransaction(getTxConfig());
org.neo4j.driver.Transaction transaction = session.beginTransaction(buildTxConfig());
return transaction(testkitState.addTransactionHolder(new TransactionHolder(sessionHolder, transaction)));
}

Expand All @@ -62,7 +58,7 @@ public CompletionStage<TestkitResponse> processAsync(TestkitState testkitState)
AsyncSession session = sessionHolder.getSession();
TransactionConfig.Builder builder = TransactionConfig.builder();

return session.beginTransactionAsync(getTxConfig())
return session.beginTransactionAsync(buildTxConfig())
.thenApply(tx -> transaction(
testkitState.addAsyncTransactionHolder(new AsyncTransactionHolder(sessionHolder, tx))));
});
Expand All @@ -75,7 +71,7 @@ public Mono<TestkitResponse> processRx(TestkitState testkitState) {
RxSession session = sessionHolder.getSession();
TransactionConfig.Builder builder = TransactionConfig.builder();

return Mono.fromDirect(session.beginTransaction(getTxConfig()))
return Mono.fromDirect(session.beginTransaction(buildTxConfig()))
.map(tx -> transaction(
testkitState.addRxTransactionHolder(new RxTransactionHolder(sessionHolder, tx))));
});
Expand All @@ -87,7 +83,7 @@ public Mono<TestkitResponse> processReactive(TestkitState testkitState) {
ReactiveSession session = sessionHolder.getSession();
TransactionConfig.Builder builder = TransactionConfig.builder();

return Mono.fromDirect(flowPublisherToFlux(session.beginTransaction(getTxConfig())))
return Mono.fromDirect(flowPublisherToFlux(session.beginTransaction(buildTxConfig())))
.map(tx -> transaction(testkitState.addReactiveTransactionHolder(
new ReactiveTransactionHolder(sessionHolder, tx))));
});
Expand All @@ -99,7 +95,7 @@ public Mono<TestkitResponse> processReactiveStreams(TestkitState testkitState) {
var session = sessionHolder.getSession();
TransactionConfig.Builder builder = TransactionConfig.builder();

return Mono.fromDirect(session.beginTransaction(getTxConfig()))
return Mono.fromDirect(session.beginTransaction(buildTxConfig()))
.map(tx -> transaction(testkitState.addReactiveTransactionStreamsHolder(
new ReactiveTransactionStreamsHolder(sessionHolder, tx))));
});
Expand All @@ -113,18 +109,8 @@ private Transaction transaction(String txId) {

@Getter
@Setter
public static class SessionBeginTransactionBody implements WithTxConfig.ITxConfigBody {
public static class SessionBeginTransactionBody
extends AbstractTestkitRequestWithTransactionConfig.TransactionConfigBody {
private String sessionId;

@JsonDeserialize(using = TestkitCypherParamDeserializer.class)
private Map<String, Object> txMeta;

private Integer timeout;
private Boolean timeoutPresent = false;

public void setTimeout(Integer timeout) {
this.timeout = timeout;
timeoutPresent = true;
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -21,8 +21,6 @@
import static reactor.adapter.JdkFlowAdapter.flowPublisherToFlux;
import static reactor.adapter.JdkFlowAdapter.publisherToFlowPublisher;

import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ExecutionException;
Expand All @@ -37,7 +35,6 @@
import neo4j.org.testkit.backend.holder.RxTransactionHolder;
import neo4j.org.testkit.backend.holder.SessionHolder;
import neo4j.org.testkit.backend.holder.TransactionHolder;
import neo4j.org.testkit.backend.messages.requests.deserializer.TestkitCypherParamDeserializer;
import neo4j.org.testkit.backend.messages.responses.RetryableDone;
import neo4j.org.testkit.backend.messages.responses.RetryableTry;
import neo4j.org.testkit.backend.messages.responses.TestkitResponse;
Expand All @@ -52,15 +49,14 @@

@Setter
@Getter
public class SessionReadTransaction extends WithTxConfig {
private SessionReadTransactionBody data;

public class SessionReadTransaction
extends AbstractTestkitRequestWithTransactionConfig<SessionReadTransaction.SessionReadTransactionBody> {
@Override
@SuppressWarnings("deprecation")
public TestkitResponse process(TestkitState testkitState) {
SessionHolder sessionHolder = testkitState.getSessionHolder(data.getSessionId());
Session session = sessionHolder.getSession();
session.readTransaction(handle(testkitState, sessionHolder), getTxConfig());
session.readTransaction(handle(testkitState, sessionHolder), buildTxConfig());
return retryableDone();
}

Expand All @@ -81,7 +77,7 @@ public CompletionStage<TestkitResponse> processAsync(TestkitState testkitState)
return txWorkFuture;
};

return session.readTransactionAsync(workWrapper, getTxConfig());
return session.readTransactionAsync(workWrapper, buildTxConfig());
})
.thenApply(nothing -> retryableDone());
}
Expand All @@ -100,7 +96,7 @@ public Mono<TestkitResponse> processRx(TestkitState testkitState) {
return Mono.fromCompletionStage(tryResult);
};

return Mono.fromDirect(sessionHolder.getSession().readTransaction(workWrapper, getTxConfig()));
return Mono.fromDirect(sessionHolder.getSession().readTransaction(workWrapper, buildTxConfig()));
})
.then(Mono.just(retryableDone()));
}
Expand All @@ -120,7 +116,7 @@ public Mono<TestkitResponse> processReactive(TestkitState testkitState) {
};

return Mono.fromDirect(
flowPublisherToFlux(sessionHolder.getSession().executeRead(workWrapper, getTxConfig())));
flowPublisherToFlux(sessionHolder.getSession().executeRead(workWrapper, buildTxConfig())));
})
.then(Mono.just(retryableDone()));
}
Expand All @@ -140,7 +136,7 @@ public Mono<TestkitResponse> processReactiveStreams(TestkitState testkitState) {
return Mono.fromCompletionStage(tryResult);
};

return Mono.fromDirect(sessionHolder.getSession().executeRead(workWrapper, getTxConfig()));
return Mono.fromDirect(sessionHolder.getSession().executeRead(workWrapper, buildTxConfig()));
})
.then(Mono.just(retryableDone()));
}
Expand Down Expand Up @@ -180,18 +176,8 @@ private RetryableDone retryableDone() {

@Setter
@Getter
public static class SessionReadTransactionBody implements WithTxConfig.ITxConfigBody {
public static class SessionReadTransactionBody
extends AbstractTestkitRequestWithTransactionConfig.TransactionConfigBody {
private String sessionId;

@JsonDeserialize(using = TestkitCypherParamDeserializer.class)
private Map<String, Object> txMeta;

private Integer timeout;
private Boolean timeoutPresent = false;

public void setTimeout(Integer timeout) {
this.timeout = timeout;
timeoutPresent = true;
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -47,17 +47,15 @@

@Setter
@Getter
public class SessionRun extends WithTxConfig {
private SessionRunBody data;

public class SessionRun extends AbstractTestkitRequestWithTransactionConfig<SessionRun.SessionRunBody> {
@Override
public TestkitResponse process(TestkitState testkitState) {
SessionHolder sessionHolder = testkitState.getSessionHolder(data.getSessionId());
Session session = sessionHolder.getSession();
Query query = Optional.ofNullable(data.params)
.map(params -> new Query(data.cypher, data.params))
.orElseGet(() -> new Query(data.cypher));
org.neo4j.driver.Result result = session.run(query, getTxConfig());
org.neo4j.driver.Result result = session.run(query, buildTxConfig());
String id = testkitState.addResultHolder(new ResultHolder(sessionHolder, result));

return createResponse(id, result.keys());
Expand All @@ -71,7 +69,7 @@ public CompletionStage<TestkitResponse> processAsync(TestkitState testkitState)
.map(params -> new Query(data.cypher, data.params))
.orElseGet(() -> new Query(data.cypher));

return session.runAsync(query, getTxConfig()).thenApply(resultCursor -> {
return session.runAsync(query, buildTxConfig()).thenApply(resultCursor -> {
String id = testkitState.addAsyncResultHolder(new ResultCursorHolder(sessionHolder, resultCursor));
return createResponse(id, resultCursor.keys());
});
Expand All @@ -87,7 +85,7 @@ public Mono<TestkitResponse> processRx(TestkitState testkitState) {
.map(params -> new Query(data.cypher, data.params))
.orElseGet(() -> new Query(data.cypher));

RxResult result = session.run(query, getTxConfig());
RxResult result = session.run(query, buildTxConfig());
String id = testkitState.addRxResultHolder(new RxResultHolder(sessionHolder, result));

// The keys() method causes RUN message exchange.
Expand All @@ -104,7 +102,7 @@ public Mono<TestkitResponse> processReactive(TestkitState testkitState) {
.map(params -> new Query(data.cypher, data.params))
.orElseGet(() -> new Query(data.cypher));

return Mono.fromDirect(flowPublisherToFlux(session.run(query, getTxConfig())))
return Mono.fromDirect(flowPublisherToFlux(session.run(query, buildTxConfig())))
.map(result -> {
String id =
testkitState.addReactiveResultHolder(new ReactiveResultHolder(sessionHolder, result));
Expand All @@ -121,7 +119,7 @@ public Mono<TestkitResponse> processReactiveStreams(TestkitState testkitState) {
.map(params -> new Query(data.cypher, data.params))
.orElseGet(() -> new Query(data.cypher));

return Mono.fromDirect(session.run(query, getTxConfig())).map(result -> {
return Mono.fromDirect(session.run(query, buildTxConfig())).map(result -> {
String id = testkitState.addReactiveResultStreamsHolder(
new ReactiveResultStreamsHolder(sessionHolder, result));
return createResponse(id, result.keys());
Expand All @@ -137,22 +135,11 @@ private Result createResponse(String resultId, List<String> keys) {

@Setter
@Getter
public static class SessionRunBody implements ITxConfigBody {
public static class SessionRunBody extends AbstractTestkitRequestWithTransactionConfig.TransactionConfigBody {
@JsonDeserialize(using = TestkitCypherParamDeserializer.class)
private Map<String, Object> params;

private String sessionId;
private String cypher;

@JsonDeserialize(using = TestkitCypherParamDeserializer.class)
private Map<String, Object> txMeta;

private Integer timeout;
private Boolean timeoutPresent = false;

public void setTimeout(Integer timeout) {
this.timeout = timeout;
timeoutPresent = true;
}
}
}
Loading

0 comments on commit 0347029

Please sign in to comment.