feat: Java support in TransactionEvent (#74)

gradle.properties

@@ -39,3 +39,6 @@ testContainerVersion=1.19.3
 
 ### Jackson ###
 jacksonVersion=2.16.1
+
+### Awaitility ###
+awaitilityVersion=3.0.0

gradle/test.gradle

@@ -10,4 +10,7 @@ dependencies {
     testImplementation "io.kotest.extensions:kotest-extensions-spring:${kotestExtensionSpringVersion}"
 
     testImplementation "org.testcontainers:testcontainers:${testContainerVersion}"
+
+    testImplementation "org.awaitility:awaitility:${awaitilityVersion}"
+
 }

src/main/kotlin/org/rooftop/netx/api/TransactionEvent.kt

@@ -9,6 +9,9 @@ sealed class TransactionEvent(
     private val event: String?,
     private val codec: Codec,
 ) {
+
+    fun <T : Any> decodeEvent(type: Class<T>): T = decodeEvent(type.kotlin)
+
     fun <T : Any> decodeEvent(type: KClass<T>): T =
         codec.decode(
             event ?: throw NullPointerException("Cannot decode event cause event is null"),

src/main/kotlin/org/rooftop/netx/api/TransactionRollbackEvent.kt

@@ -12,5 +12,7 @@ class TransactionRollbackEvent(
     private val codec: Codec,
 ) : TransactionEvent(transactionId, nodeName, group, event, codec) {
 
+    fun <T : Any> decodeUndo(type: Class<T>): T = decodeUndo(type.kotlin)
+
     fun <T : Any> decodeUndo(type: KClass<T>): T = codec.decode(undo, type)
 }

src/test/java/org/rooftop/netx/javasupports/Event.java

@@ -0,0 +1,7 @@
+package org.rooftop.netx.javasupports;
+
+public record Event(
+    Long event
+) {
+
+}

src/test/java/org/rooftop/netx/javasupports/NetxJavaSupportsTest.java

@@ -0,0 +1,75 @@
+package org.rooftop.netx.javasupports;
+
+import java.util.concurrent.TimeUnit;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.rooftop.netx.api.TransactionManager;
+import org.rooftop.netx.idl.TransactionState;
+import org.rooftop.netx.meta.EnableDistributedTransaction;
+import org.rooftop.netx.redis.RedisContainer;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.TestPropertySource;
+import org.springframework.test.context.junit.jupiter.SpringExtension;
+
+@EnableDistributedTransaction
+@ExtendWith(SpringExtension.class)
+@ContextConfiguration(classes = {
+    RedisContainer.class,
+    NetxJavaSupportsTest.class,
+    TransactionEventListeners.class,
+})
+@DisplayName("NetxJavaSupportsTest")
+@TestPropertySource("classpath:application.properties")
+class NetxJavaSupportsTest {
+
+    private static final Undo NEGATIVE_UNDO = new Undo(-1L);
+    private static final Undo POSITIVE_UNDO = new Undo(1L);
+    private static final Event NEGATIVE_EVENT = new Event(-1L);
+    private static final Event POSITIVE_EVENT = new Event(1L);
+
+    @Autowired
+    private TransactionManager transactionManager;
+
+    @Autowired
+    private TransactionEventListeners transactionEventListeners;
+
+    @BeforeEach
+    void clear() {
+        transactionEventListeners.clear();
+    }
+
+    @Test
+    @DisplayName("Scenario-1. Start -> Join -> Commit")
+    void Scenario1_Start_Join_Commit() {
+        String transactionId = transactionManager.syncStart(NEGATIVE_UNDO, NEGATIVE_EVENT);
+        transactionManager.syncJoin(transactionId, NEGATIVE_UNDO, NEGATIVE_EVENT);
+        transactionManager.syncCommit(transactionId);
+
+        Awaitility.waitAtMost(5, TimeUnit.SECONDS)
+            .untilAsserted(() -> {
+                transactionEventListeners.assertTransactionCount(TransactionState.TRANSACTION_STATE_START, 1);
+                transactionEventListeners.assertTransactionCount(TransactionState.TRANSACTION_STATE_JOIN, 1);
+                transactionEventListeners.assertTransactionCount(TransactionState.TRANSACTION_STATE_COMMIT, 1);
+            });
+    }
+
+    @Test
+    @DisplayName("Scenario-2. Start -> Join -> Rollback")
+    void Transaction_Start_Join_Rollback() {
+        String transactionId = transactionManager.syncStart(POSITIVE_UNDO, POSITIVE_EVENT);
+        transactionManager.syncJoin(transactionId, POSITIVE_UNDO, POSITIVE_EVENT);
+        transactionManager.syncRollback(transactionId, "Scenario-2. Start -> Join -> Rollback");
+
+        Awaitility.waitAtMost(5, TimeUnit.SECONDS)
+            .untilAsserted(() -> {
+                transactionEventListeners.assertTransactionCount(TransactionState.TRANSACTION_STATE_START, 1);
+                transactionEventListeners.assertTransactionCount(TransactionState.TRANSACTION_STATE_JOIN, 1);
+                transactionEventListeners.assertTransactionCount(TransactionState.TRANSACTION_STATE_ROLLBACK, 1);
+            });
+    }
+
+}

src/test/java/org/rooftop/netx/javasupports/TransactionClient.java

@@ -0,0 +1,30 @@
+package org.rooftop.netx.javasupports;
+
+import org.rooftop.netx.api.TransactionManager;
+import org.springframework.boot.test.context.TestComponent;
+
+@TestComponent
+public class TransactionClient {
+
+    private final TransactionManager transactionManager;
+
+    public TransactionClient(TransactionManager transactionManager) {
+        this.transactionManager = transactionManager;
+    }
+
+    public String startTransaction(Undo undo, Event event) {
+        return transactionManager.syncStart(undo, event);
+    }
+
+    public void joinTransaction(String transactionId, Undo undo, Event event) {
+        transactionManager.syncJoin(transactionId, undo, event);
+    }
+
+    public void commitTransaction(String transactionId) {
+        transactionManager.syncCommit(transactionId);
+    }
+
+    public void rollbackTransaction(String transactionId, String cause) {
+        transactionManager.syncRollback(transactionId, cause);
+    }
+}

src/test/java/org/rooftop/netx/javasupports/TransactionEventListeners.java

@@ -0,0 +1,74 @@
+package org.rooftop.netx.javasupports;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import org.assertj.core.api.Assertions;
+import org.rooftop.netx.api.DecodeException;
+import org.rooftop.netx.api.TransactionCommitEvent;
+import org.rooftop.netx.api.TransactionCommitListener;
+import org.rooftop.netx.api.TransactionJoinEvent;
+import org.rooftop.netx.api.TransactionJoinListener;
+import org.rooftop.netx.api.TransactionRollbackEvent;
+import org.rooftop.netx.api.TransactionRollbackListener;
+import org.rooftop.netx.api.TransactionStartEvent;
+import org.rooftop.netx.api.TransactionStartListener;
+import org.rooftop.netx.idl.TransactionState;
+import org.rooftop.netx.meta.TransactionHandler;
+import reactor.core.publisher.Mono;
+
+@TransactionHandler
+public class TransactionEventListeners {
+
+    private final Map<TransactionState, Integer> receivedTransactions = new ConcurrentHashMap<>();
+
+    public void clear() {
+        receivedTransactions.clear();
+    }
+
+    public void assertTransactionCount(TransactionState transactionState, int count) {
+        Assertions.assertThat(receivedTransactions.getOrDefault(transactionState, 0))
+            .isEqualTo(count);
+    }
+
+    @TransactionStartListener(
+        event = Event.class,
+        noRetryFor = IllegalArgumentException.class
+    )
+    public void listenTransactionStartEvent(TransactionStartEvent transactionStartEvent) {
+        incrementTransaction(TransactionState.TRANSACTION_STATE_START);
+        Event event = transactionStartEvent.decodeEvent(Event.class);
+        if (event.event() < 0) {
+            throw new IllegalArgumentException();
+        }
+    }
+
+    @TransactionJoinListener(
+        event = Event.class,
+        noRetryFor = IllegalArgumentException.class
+    )
+    public void listenTransactionJoinEvent(TransactionJoinEvent transactionJoinEvent) {
+        incrementTransaction(TransactionState.TRANSACTION_STATE_JOIN);
+        Event event = transactionJoinEvent.decodeEvent(Event.class);
+        if (event.event() < 0) {
+            throw new IllegalArgumentException();
+        }
+    }
+
+    @TransactionCommitListener
+    public Mono<Long> listenTransactionCommitEvent(TransactionCommitEvent transactionCommitEvent) {
+        incrementTransaction(TransactionState.TRANSACTION_STATE_COMMIT);
+        return Mono.just(1L);
+    }
+
+    @TransactionRollbackListener(noRetryFor = DecodeException.class)
+    public String listenTransactionRollbackEvent(TransactionRollbackEvent transactionRollbackEvent) {
+        incrementTransaction(TransactionState.TRANSACTION_STATE_ROLLBACK);
+        transactionRollbackEvent.decodeUndo(Undo.class);
+        return "listenTransactionRollbackEvent";
+    }
+
+    private void incrementTransaction(TransactionState transactionState) {
+        receivedTransactions.put(transactionState,
+            receivedTransactions.getOrDefault(transactionState, 0) + 1);
+    }
+}

src/test/java/org/rooftop/netx/javasupports/Undo.java

@@ -0,0 +1,7 @@
+package org.rooftop.netx.javasupports;
+
+public record Undo(
+    Long undo
+) {
+
+}