2020
2121import static reactor .adapter .JdkFlowAdapter .flowPublisherToFlux ;
2222
23- import java .time .Duration ;
24- import java .util .Map ;
25- import java .util .Optional ;
2623import java .util .concurrent .CompletionStage ;
2724import lombok .Getter ;
2825import lombok .Setter ;
29- import neo4j .org .testkit .backend .CustomDriverError ;
3026import neo4j .org .testkit .backend .TestkitState ;
3127import neo4j .org .testkit .backend .holder .AsyncTransactionHolder ;
3228import neo4j .org .testkit .backend .holder .ReactiveTransactionHolder ;
4339import org .neo4j .driver .reactive .RxSession ;
4440import reactor .core .publisher .Mono ;
4541
46- @ Setter
47- @ Getter
48- public class SessionBeginTransaction implements TestkitRequest {
49- private SessionBeginTransactionBody data ;
50-
51- private void configureTimeout (TransactionConfig .Builder builder ) {
52- if (data .getTimeoutPresent ()) {
53- try {
54- if (data .getTimeout () != null ) {
55- builder .withTimeout (Duration .ofMillis (data .getTimeout ()));
56- } else {
57- builder .withDefaultTimeout ();
58- }
59- } catch (IllegalArgumentException e ) {
60- throw new CustomDriverError (e );
61- }
62- }
63- }
64-
42+ public class SessionBeginTransaction
43+ extends AbstractTestkitRequestWithTransactionConfig <SessionBeginTransaction .SessionBeginTransactionBody > {
6544 @ Override
6645 public TestkitResponse process (TestkitState testkitState ) {
6746 SessionHolder sessionHolder = testkitState .getSessionHolder (data .getSessionId ());
6847 Session session = sessionHolder .getSession ();
69- TransactionConfig .Builder builder = TransactionConfig .builder ();
70- Optional .ofNullable (data .txMeta ).ifPresent (builder ::withMetadata );
7148
72- configureTimeout (builder );
73-
74- org .neo4j .driver .Transaction transaction = session .beginTransaction (builder .build ());
49+ org .neo4j .driver .Transaction transaction = session .beginTransaction (buildTxConfig ());
7550 return transaction (testkitState .addTransactionHolder (new TransactionHolder (sessionHolder , transaction )));
7651 }
7752
@@ -80,11 +55,8 @@ public CompletionStage<TestkitResponse> processAsync(TestkitState testkitState)
8055 return testkitState .getAsyncSessionHolder (data .getSessionId ()).thenCompose (sessionHolder -> {
8156 AsyncSession session = sessionHolder .getSession ();
8257 TransactionConfig .Builder builder = TransactionConfig .builder ();
83- Optional .ofNullable (data .txMeta ).ifPresent (builder ::withMetadata );
84-
85- configureTimeout (builder );
8658
87- return session .beginTransactionAsync (builder . build ())
59+ return session .beginTransactionAsync (buildTxConfig ())
8860 .thenApply (tx -> transaction (
8961 testkitState .addAsyncTransactionHolder (new AsyncTransactionHolder (sessionHolder , tx ))));
9062 });
@@ -96,11 +68,8 @@ public Mono<TestkitResponse> processRx(TestkitState testkitState) {
9668 return testkitState .getRxSessionHolder (data .getSessionId ()).flatMap (sessionHolder -> {
9769 RxSession session = sessionHolder .getSession ();
9870 TransactionConfig .Builder builder = TransactionConfig .builder ();
99- Optional .ofNullable (data .txMeta ).ifPresent (builder ::withMetadata );
10071
101- configureTimeout (builder );
102-
103- return Mono .fromDirect (session .beginTransaction (builder .build ()))
72+ return Mono .fromDirect (session .beginTransaction (buildTxConfig ()))
10473 .map (tx -> transaction (
10574 testkitState .addRxTransactionHolder (new RxTransactionHolder (sessionHolder , tx ))));
10675 });
@@ -111,11 +80,8 @@ public Mono<TestkitResponse> processReactive(TestkitState testkitState) {
11180 return testkitState .getReactiveSessionHolder (data .getSessionId ()).flatMap (sessionHolder -> {
11281 ReactiveSession session = sessionHolder .getSession ();
11382 TransactionConfig .Builder builder = TransactionConfig .builder ();
114- Optional .ofNullable (data .txMeta ).ifPresent (builder ::withMetadata );
115-
116- configureTimeout (builder );
11783
118- return Mono .fromDirect (flowPublisherToFlux (session .beginTransaction (builder . build ())))
84+ return Mono .fromDirect (flowPublisherToFlux (session .beginTransaction (buildTxConfig ())))
11985 .map (tx -> transaction (testkitState .addReactiveTransactionHolder (
12086 new ReactiveTransactionHolder (sessionHolder , tx ))));
12187 });
@@ -126,11 +92,8 @@ public Mono<TestkitResponse> processReactiveStreams(TestkitState testkitState) {
12692 return testkitState .getReactiveSessionStreamsHolder (data .getSessionId ()).flatMap (sessionHolder -> {
12793 var session = sessionHolder .getSession ();
12894 TransactionConfig .Builder builder = TransactionConfig .builder ();
129- Optional .ofNullable (data .txMeta ).ifPresent (builder ::withMetadata );
13095
131- configureTimeout (builder );
132-
133- return Mono .fromDirect (session .beginTransaction (builder .build ()))
96+ return Mono .fromDirect (session .beginTransaction (buildTxConfig ()))
13497 .map (tx -> transaction (testkitState .addReactiveTransactionStreamsHolder (
13598 new ReactiveTransactionStreamsHolder (sessionHolder , tx ))));
13699 });
@@ -144,15 +107,8 @@ private Transaction transaction(String txId) {
144107
145108 @ Getter
146109 @ Setter
147- public static class SessionBeginTransactionBody {
110+ public static class SessionBeginTransactionBody
111+ extends AbstractTestkitRequestWithTransactionConfig .TransactionConfigBody {
148112 private String sessionId ;
149- private Map <String , Object > txMeta ;
150- private Integer timeout ;
151- private Boolean timeoutPresent = false ;
152-
153- public void setTimeout (Integer timeout ) {
154- this .timeout = timeout ;
155- timeoutPresent = true ;
156- }
157113 }
158114}
0 commit comments