From 3bd832debe9a6a09888cf177363d1c663d55ac84 Mon Sep 17 00:00:00 2001 From: devxb Date: Wed, 7 Feb 2024 00:31:23 +0900 Subject: [PATCH 1/5] =?UTF-8?q?refactor:=20UndoManager=EB=A5=BC=20?= =?UTF-8?q?=EC=82=AD=EC=A0=9C=ED=95=98=EA=B3=A0,=20=EB=A9=94=EC=8B=9C?= =?UTF-8?q?=EC=A7=80=ED=81=90=EC=97=90=EC=84=9C=20=EC=A0=84=EB=B6=80=20?= =?UTF-8?q?=EA=B4=80=EB=A6=AC=ED=95=98=EB=8F=84=EB=A1=9D=20=EC=88=98?= =?UTF-8?q?=EC=A0=95=ED=95=9C=EB=8B=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 5 +- .../netx/api/TransactionRollbackEvent.kt | 2 +- .../AutoConfigureDistributedTransaction.kt | 3 +- .../engine/AbstractTransactionDispatcher.kt | 8 +- .../netx/engine/AbstractTransactionManager.kt | 7 -- .../org/rooftop/netx/engine/UndoManager.kt | 13 ---- .../redis/RedisStreamTransactionDispatcher.kt | 27 ++++++- .../redis/RedisStreamTransactionManager.kt | 3 - .../netx/redis/RedisTransactionConfigurer.kt | 4 - .../rooftop/netx/redis/RedisUndoConfigurer.kt | 44 ----------- .../rooftop/netx/redis/RedisUndoManager.kt | 36 --------- .../netx/redis/SpringEventPublisher.kt | 1 - .../netx/redis/RedisUndoManagerTest.kt | 77 ------------------- src/test/resources/application.properties | 3 - 14 files changed, 32 insertions(+), 201 deletions(-) delete mode 100644 src/main/kotlin/org/rooftop/netx/engine/UndoManager.kt delete mode 100644 src/main/kotlin/org/rooftop/netx/redis/RedisUndoConfigurer.kt delete mode 100644 src/main/kotlin/org/rooftop/netx/redis/RedisUndoManager.kt delete mode 100644 src/test/kotlin/org/rooftop/netx/redis/RedisUndoManagerTest.kt diff --git a/README.md b/README.md index d21c5b5..d796f68 100644 --- a/README.md +++ b/README.md @@ -3,6 +3,8 @@ > Distributed transaction library based on Choreography
+ + ![version 0.1.2](https://img.shields.io/badge/version-0.1.2-black?labelColor=black&style=flat-square) ![jdk 17](https://img.shields.io/badge/jdk-17-orange?labelColor=black&style=flat-square) @@ -46,9 +48,6 @@ class Application { | **netx.group** | pay-group | 분산 노드의 그룹입니다. 트랜잭션 이벤트는 같은 그룹내 하나의 노드로만 전송됩니다. | | **netx.node-id** | 1 | id 생성에 사용될 식별자입니다. 모든 서버는 반드시 다른 id를 할당받아야 하며, 1~256 만큼의 id를 설정할 수 있습니다. _`중복된 id 생성을 방지하기위해 twitter snowflake 알고리즘으로 id를 생성합니다.`_ | | **netx.node-name** | pay-1 | _`$netx.group`_ 에 참여할 서버의 이름입니다. 같은 그룹내에 중복된 이름이 존재하면 안됩니다. | -| **netx.undo.mode** | redis | 트랜잭션 undo 상태 저장에 사용할 저장소 구현체의 mode 입니다. | -| **netx.undo.host** | localhost | 트랜잭션 undo 상태 저장에 사용할 저장소의 host url 입니다. | -| **netx.undo.port** | 6380 | 트랜잭션 undo 상태 저장에 사용할 저장소의 port 입니다. | ### Usage example diff --git a/src/main/kotlin/org/rooftop/netx/api/TransactionRollbackEvent.kt b/src/main/kotlin/org/rooftop/netx/api/TransactionRollbackEvent.kt index addfdce..63de5c6 100644 --- a/src/main/kotlin/org/rooftop/netx/api/TransactionRollbackEvent.kt +++ b/src/main/kotlin/org/rooftop/netx/api/TransactionRollbackEvent.kt @@ -4,5 +4,5 @@ data class TransactionRollbackEvent( val transactionId: String, val nodeName: String, val cause: String?, - val undoState: String, + val undo: String, ) diff --git a/src/main/kotlin/org/rooftop/netx/autoconfig/AutoConfigureDistributedTransaction.kt b/src/main/kotlin/org/rooftop/netx/autoconfig/AutoConfigureDistributedTransaction.kt index 093beb9..42278e7 100644 --- a/src/main/kotlin/org/rooftop/netx/autoconfig/AutoConfigureDistributedTransaction.kt +++ b/src/main/kotlin/org/rooftop/netx/autoconfig/AutoConfigureDistributedTransaction.kt @@ -1,10 +1,9 @@ package org.rooftop.netx.autoconfig import org.rooftop.netx.redis.RedisTransactionConfigurer -import org.rooftop.netx.redis.RedisUndoConfigurer import org.springframework.boot.autoconfigure.ImportAutoConfiguration @Target(AnnotationTarget.CLASS) @Retention(AnnotationRetention.RUNTIME) -@ImportAutoConfiguration(RedisUndoConfigurer::class, RedisTransactionConfigurer::class) +@ImportAutoConfiguration(RedisTransactionConfigurer::class) annotation class AutoConfigureDistributedTransaction diff --git a/src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionDispatcher.kt b/src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionDispatcher.kt index 3450aac..36bcfa7 100644 --- a/src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionDispatcher.kt +++ b/src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionDispatcher.kt @@ -11,7 +11,6 @@ import reactor.core.publisher.Flux import reactor.core.publisher.Mono abstract class AbstractTransactionDispatcher( - private val undoManager: UndoManager, private val eventPublisher: EventPublisher, ) { @@ -53,21 +52,22 @@ abstract class AbstractTransactionDispatcher( } private fun publishRollback(transaction: Transaction): Mono { - return undoManager.find(transaction.id) + return findOwnTransaction(transaction) .doOnNext { eventPublisher.publish( TransactionRollbackEvent( transaction.id, transaction.serverId, transaction.cause, - it + it.undo ) ) } - .flatMap { undoManager.delete(transaction.id) } .map { transaction } } + protected abstract fun findOwnTransaction(transaction: Transaction): Mono + private fun publishStart(it: Transaction): Mono { return Mono.just(it) .doOnNext { diff --git a/src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionManager.kt b/src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionManager.kt index 0a77b6b..357e36f 100644 --- a/src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionManager.kt +++ b/src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionManager.kt @@ -11,13 +11,11 @@ abstract class AbstractTransactionManager( private val nodeName: String, private val eventPublisher: EventPublisher, private val transactionIdGenerator: TransactionIdGenerator = TransactionIdGenerator(nodeId), - private val undoManager: UndoManager, ) : TransactionManager { final override fun start(undo: String): Mono { return startTransaction() .subscribeTransaction() - .saveUndoState(undo) .contextWrite { it.put(CONTEXT_TX_KEY, transactionIdGenerator.generate()) } } @@ -36,7 +34,6 @@ abstract class AbstractTransactionManager( return exists(transactionId) .joinTransaction() .subscribeTransaction() - .saveUndoState(undo) .contextWrite { it.put(CONTEXT_TX_KEY, transactionId) } } @@ -56,10 +53,6 @@ abstract class AbstractTransactionManager( } } - private fun Mono.saveUndoState(undo: String): Mono { - return this.flatMap { undoManager.save(it, undo) } - } - final override fun rollback(transactionId: String, cause: String): Mono { return exists(transactionId) .publishTransaction(transaction { diff --git a/src/main/kotlin/org/rooftop/netx/engine/UndoManager.kt b/src/main/kotlin/org/rooftop/netx/engine/UndoManager.kt deleted file mode 100644 index ab12d9e..0000000 --- a/src/main/kotlin/org/rooftop/netx/engine/UndoManager.kt +++ /dev/null @@ -1,13 +0,0 @@ -package org.rooftop.netx.engine - -import reactor.core.publisher.Mono - -interface UndoManager { - - fun save(transactionId: String, undo: String): Mono - - fun find(transactionId: String): Mono - - fun delete(transactionId: String): Mono - -} diff --git a/src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionDispatcher.kt b/src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionDispatcher.kt index fc5e7d7..f1ca9e6 100644 --- a/src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionDispatcher.kt +++ b/src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionDispatcher.kt @@ -2,8 +2,8 @@ package org.rooftop.netx.redis import org.rooftop.netx.engine.AbstractTransactionDispatcher import org.rooftop.netx.engine.SubscribeTransactionEvent -import org.rooftop.netx.engine.UndoManager import org.rooftop.netx.idl.Transaction +import org.rooftop.netx.idl.TransactionState import org.springframework.context.ApplicationEventPublisher import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory import org.springframework.data.redis.connection.stream.Consumer @@ -12,6 +12,7 @@ import org.springframework.data.redis.connection.stream.StreamOffset import org.springframework.data.redis.core.ReactiveRedisTemplate import org.springframework.data.redis.stream.StreamReceiver import reactor.core.publisher.Flux +import reactor.core.publisher.Mono import reactor.core.scheduler.Schedulers import kotlin.time.Duration.Companion.hours import kotlin.time.toJavaDuration @@ -19,11 +20,10 @@ import kotlin.time.toJavaDuration class RedisStreamTransactionDispatcher( eventPublisher: ApplicationEventPublisher, connectionFactory: ReactiveRedisConnectionFactory, - undoManager: UndoManager, private val streamGroup: String, private val nodeName: String, private val reactiveRedisTemplate: ReactiveRedisTemplate, -) : AbstractTransactionDispatcher(undoManager, SpringEventPublisher(eventPublisher)) { +) : AbstractTransactionDispatcher(SpringEventPublisher(eventPublisher)) { private val options = StreamReceiver.StreamReceiverOptions.builder() .pollTimeout(1.hours.toJavaDuration()) @@ -39,6 +39,14 @@ class RedisStreamTransactionDispatcher( StreamOffset.create(event.transactionId, ReadOffset.from(">")) ).publishOn(Schedulers.parallel()) .map { Transaction.parseFrom(it.value["data"]?.toByteArray()) } + .flatMap { + when (it.state) { + TransactionState.TRANSACTION_STATE_ROLLBACK -> + findOwnTransaction(it) + + else -> Mono.just(it) + } + } } } @@ -47,4 +55,17 @@ class RedisStreamTransactionDispatcher( .createGroup(event.transactionId, ReadOffset.from("0"), streamGroup) .flatMapMany { Flux.just(it) } } + + override fun findOwnTransaction(transaction: Transaction): Mono { + return reactiveRedisTemplate.opsForStream() + .read(StreamOffset.create(transaction.id, ReadOffset.from("0"))) + .map { Transaction.parseFrom(it.value["data"]!!) } + .filter { it.group == streamGroup } + .filter { hasUndo(it) } + .next() + } + + private fun hasUndo(transaction: Transaction): Boolean = + transaction.state == TransactionState.TRANSACTION_STATE_JOIN + || transaction.state == TransactionState.TRANSACTION_STATE_START } diff --git a/src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionManager.kt b/src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionManager.kt index 2e37ef6..f213f47 100644 --- a/src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionManager.kt +++ b/src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionManager.kt @@ -1,7 +1,6 @@ package org.rooftop.netx.redis import org.rooftop.netx.engine.AbstractTransactionManager -import org.rooftop.netx.engine.UndoManager import org.rooftop.netx.idl.Transaction import org.springframework.context.ApplicationEventPublisher import org.springframework.data.domain.Range @@ -13,13 +12,11 @@ class RedisStreamTransactionManager( nodeId: Int, nodeName: String, applicationEventPublisher: ApplicationEventPublisher, - undoManager: UndoManager, private val reactiveRedisTemplate: ReactiveRedisTemplate, ) : AbstractTransactionManager( nodeId, nodeName, SpringEventPublisher(applicationEventPublisher), - undoManager = undoManager ) { override fun findAnyTransaction(transactionId: String): Mono { diff --git a/src/main/kotlin/org/rooftop/netx/redis/RedisTransactionConfigurer.kt b/src/main/kotlin/org/rooftop/netx/redis/RedisTransactionConfigurer.kt index 376852f..e4f8708 100644 --- a/src/main/kotlin/org/rooftop/netx/redis/RedisTransactionConfigurer.kt +++ b/src/main/kotlin/org/rooftop/netx/redis/RedisTransactionConfigurer.kt @@ -1,7 +1,6 @@ package org.rooftop.netx.redis import org.rooftop.netx.api.TransactionManager -import org.rooftop.netx.engine.UndoManager import org.rooftop.pay.infra.transaction.ByteArrayRedisSerializer import org.springframework.beans.factory.annotation.Value import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty @@ -22,7 +21,6 @@ class RedisTransactionConfigurer( @Value("\${netx.node-id}") private val nodeId: Int, @Value("\${netx.node-name}") private val nodeName: String, private val applicationEventPublisher: ApplicationEventPublisher, - private val undoManager: UndoManager, ) { @Bean @@ -32,7 +30,6 @@ class RedisTransactionConfigurer( nodeId, nodeName, applicationEventPublisher, - undoManager, reactiveRedisTemplate() ) @@ -42,7 +39,6 @@ class RedisTransactionConfigurer( RedisStreamTransactionDispatcher( applicationEventPublisher, reactiveRedisConnectionFactory(), - undoManager, group, nodeName, reactiveRedisTemplate() diff --git a/src/main/kotlin/org/rooftop/netx/redis/RedisUndoConfigurer.kt b/src/main/kotlin/org/rooftop/netx/redis/RedisUndoConfigurer.kt deleted file mode 100644 index 5d807b2..0000000 --- a/src/main/kotlin/org/rooftop/netx/redis/RedisUndoConfigurer.kt +++ /dev/null @@ -1,44 +0,0 @@ -package org.rooftop.netx.redis - -import org.springframework.beans.factory.annotation.Value -import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty -import org.springframework.context.annotation.Bean -import org.springframework.context.annotation.Configuration -import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory -import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory -import org.springframework.data.redis.core.ReactiveRedisTemplate -import org.springframework.data.redis.serializer.RedisSerializationContext -import org.springframework.data.redis.serializer.StringRedisSerializer - -@Configuration -class RedisUndoConfigurer( - @Value("\${netx.group}") private val group: String, - @Value("\${netx.undo.host}") private val netxUndoHost: String, - @Value("\${netx.undo.port}") private val netxUndoPort: String, -) { - - @Bean - @ConditionalOnProperty(prefix = "netx.undo", name = ["mode"], havingValue = "redis") - fun redisUndoManager(): RedisUndoManager = RedisUndoManager(group, redisUndoServer()) - - @Bean - @ConditionalOnProperty(prefix = "netx.undo", name = ["mode"], havingValue = "redis") - fun redisUndoServer(): ReactiveRedisTemplate { - val stringRedisSerializer = StringRedisSerializer() - - val context = - RedisSerializationContext.newSerializationContext(stringRedisSerializer) - .value(stringRedisSerializer) - .build() - - return ReactiveRedisTemplate(undoServerConnectionFactory(), context) - } - - @Bean - @ConditionalOnProperty(prefix = "netx.undo", name = ["mode"], havingValue = "redis") - fun undoServerConnectionFactory(): ReactiveRedisConnectionFactory { - val undoServerPort: String = System.getProperty("netx.undo.port") ?: netxUndoPort - - return LettuceConnectionFactory(netxUndoHost, undoServerPort.toInt()) - } -} diff --git a/src/main/kotlin/org/rooftop/netx/redis/RedisUndoManager.kt b/src/main/kotlin/org/rooftop/netx/redis/RedisUndoManager.kt deleted file mode 100644 index 11d18c5..0000000 --- a/src/main/kotlin/org/rooftop/netx/redis/RedisUndoManager.kt +++ /dev/null @@ -1,36 +0,0 @@ -package org.rooftop.netx.redis - -import org.rooftop.netx.engine.UndoManager -import org.springframework.data.redis.core.ReactiveRedisTemplate -import reactor.core.publisher.Mono - -class RedisUndoManager( - private val group: String, - private val reactiveRedisTemplate: ReactiveRedisTemplate, -) : UndoManager { - - override fun find(transactionId: String): Mono { - return reactiveRedisTemplate.opsForValue()["$group:$transactionId"] - .switchIfEmpty( - Mono.error { - throw IllegalStateException("Cannot find undo state \"$group:$transactionId\"") - } - ) - } - - override fun delete(transactionId: String): Mono { - return reactiveRedisTemplate.opsForValue().delete("$group:$transactionId") - } - - override fun save(transactionId: String, undo: String): Mono { - return reactiveRedisTemplate.opsForValue() - .set("$group:$transactionId", undo) - .flatMap { - when (it) { - true -> Mono.just(it) - false -> Mono.error { throw IllegalStateException("Error occurred during the undo process.") } - } - } - .map { transactionId } - } -} diff --git a/src/main/kotlin/org/rooftop/netx/redis/SpringEventPublisher.kt b/src/main/kotlin/org/rooftop/netx/redis/SpringEventPublisher.kt index 28ad8ad..aa84518 100644 --- a/src/main/kotlin/org/rooftop/netx/redis/SpringEventPublisher.kt +++ b/src/main/kotlin/org/rooftop/netx/redis/SpringEventPublisher.kt @@ -1,7 +1,6 @@ package org.rooftop.netx.redis import org.rooftop.netx.engine.EventPublisher -import org.rooftop.netx.engine.SubscribeTransactionEvent import org.springframework.context.ApplicationEventPublisher class SpringEventPublisher(private val eventPublisher: ApplicationEventPublisher) : EventPublisher { diff --git a/src/test/kotlin/org/rooftop/netx/redis/RedisUndoManagerTest.kt b/src/test/kotlin/org/rooftop/netx/redis/RedisUndoManagerTest.kt deleted file mode 100644 index 268f976..0000000 --- a/src/test/kotlin/org/rooftop/netx/redis/RedisUndoManagerTest.kt +++ /dev/null @@ -1,77 +0,0 @@ -package org.rooftop.netx.redis - -import io.kotest.core.spec.style.DescribeSpec -import io.kotest.matchers.shouldBe -import org.rooftop.netx.autoconfig.AutoConfigureDistributedTransaction -import org.rooftop.netx.engine.UndoManager -import org.springframework.test.context.ContextConfiguration -import org.springframework.test.context.TestPropertySource -import reactor.test.StepVerifier - -@AutoConfigureDistributedTransaction -@ContextConfiguration(classes = [RedisContainer::class]) -@TestPropertySource("classpath:application.properties") -internal class RedisUndoManagerTest( - private val undoManager: UndoManager, -) : DescribeSpec({ - - describe("find 메소드는") { - context("transactionId를 받으면,") { - - val transactionId = "TX-1" - val undoState = "id:1" - undoManager.save(transactionId, undoState).block() - - it("저장된 UndoState를 반환한다.") { - val result = undoManager.find(transactionId) - - StepVerifier.create(result) - .assertNext { - it shouldBe undoState - } - .verifyComplete() - } - } - - context("존재하지 않는 transactionId를 받으면,") { - val transactionId = "UNKNOWN_TX" - - it("IllegalStateException 을 던진다.") { - val result = undoManager.find(transactionId) - - StepVerifier.create(result) - .verifyErrorMessage("Cannot find undo state \"netx-group:UNKNOWN_TX\"") - } - } - } - - describe("delete 메소드는") { - context("transactionId를 받으면,") { - - val transactionId = "TX-2" - val undoState = "id:2" - undoManager.save(transactionId, undoState).block() - - it("undoState 를 삭제하고 true를 반환한다.") { - val result = undoManager.delete(transactionId) - - StepVerifier.create(result) - .expectNext(true) - .verifyComplete() - } - } - - context("존재하지 않는 transactionId를 받으면") { - - val transactionId = "UNKNOWN_TX" - - it("false 를 반환한다.") { - val result = undoManager.delete(transactionId) - - StepVerifier.create(result) - .expectNext(false) - .verifyComplete() - } - } - } -}) diff --git a/src/test/resources/application.properties b/src/test/resources/application.properties index 91f67be..b0cdb23 100644 --- a/src/test/resources/application.properties +++ b/src/test/resources/application.properties @@ -4,6 +4,3 @@ netx.port=6379 netx.group=netx-group netx.node-id=1 netx.node-name=netx-node -netx.undo.mode=redis -netx.undo.host=localhost -netx.undo.port=6379 From 9a4feffb754df38a736e71856a9b1dffeaad0d7f Mon Sep 17 00:00:00 2001 From: devxb Date: Thu, 8 Feb 2024 16:22:55 +0900 Subject: [PATCH 2/5] =?UTF-8?q?feat:=20=EC=B2=98=EB=A6=AC=ED=95=98?= =?UTF-8?q?=EC=A7=80=20=EB=AA=BB=ED=95=9C=20=EA=B3=A0=EC=95=84=20=ED=8A=B8?= =?UTF-8?q?=EB=9E=9C=EC=9E=AD=EC=85=98=EC=9D=84=20=EC=B0=BE=EC=95=84?= =?UTF-8?q?=EC=84=9C=20=EC=B2=98=EB=A6=AC=ED=95=98=EB=8A=94=20recovery?= =?UTF-8?q?=EA=B8=B0=EB=8A=A5=EC=9D=84=20=EC=B6=94=EA=B0=80=ED=95=9C?= =?UTF-8?q?=EB=8B=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 18 +++--- build.gradle | 2 +- gradle.properties | 3 + gradle/{mq.gradle => db.gradle} | 1 + idl | 2 +- .../engine/AbstractTransactionDispatcher.kt | 47 ++++++++------ .../netx/engine/AbstractTransactionManager.kt | 29 +++++++-- .../AbstractTransactionRetrySupporter.kt | 24 +++++++ .../org/rooftop/netx/engine/EventPublisher.kt | 6 -- .../netx/engine/SubscribeTransactionEvent.kt | 5 -- .../redis/RedisStreamTransactionDispatcher.kt | 43 +++++++------ .../redis/RedisStreamTransactionManager.kt | 13 ++-- .../netx/redis/RedisTransactionConfigurer.kt | 52 +++++++++++++--- .../redis/RedisTransactionRetrySupporter.kt | 62 +++++++++++++++++++ .../netx/redis/SpringEventPublisher.kt | 11 ---- .../org/rooftop/netx/redis/RedisContainer.kt | 4 -- .../RedisStreamTransactionManagerTest.kt | 2 + src/test/resources/application.properties | 2 + 18 files changed, 231 insertions(+), 95 deletions(-) rename gradle/{mq.gradle => db.gradle} (56%) create mode 100644 src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionRetrySupporter.kt delete mode 100644 src/main/kotlin/org/rooftop/netx/engine/EventPublisher.kt delete mode 100644 src/main/kotlin/org/rooftop/netx/engine/SubscribeTransactionEvent.kt create mode 100644 src/main/kotlin/org/rooftop/netx/redis/RedisTransactionRetrySupporter.kt delete mode 100644 src/main/kotlin/org/rooftop/netx/redis/SpringEventPublisher.kt diff --git a/README.md b/README.md index d796f68..d4bb319 100644 --- a/README.md +++ b/README.md @@ -40,14 +40,16 @@ class Application { #### Properties -| key | example | description | -|--------------------|-----------|------------------------------------------------------------------------------------------------------------------------------------| -| **netx.mode** | redis | 트랜잭션 관리에 사용할 메시지 큐 구현체의 mode 입니다. | -| **netx.host** | localhost | 트랜잭션 관리에 사용할 메시지 큐 의 host url 입니다. (ex. redis host) | -| **netx.port** | 6379 | 트랜잭션 관리에 사용할 메시지 큐의 port 입니다. | -| **netx.group** | pay-group | 분산 노드의 그룹입니다. 트랜잭션 이벤트는 같은 그룹내 하나의 노드로만 전송됩니다. | -| **netx.node-id** | 1 | id 생성에 사용될 식별자입니다. 모든 서버는 반드시 다른 id를 할당받아야 하며, 1~256 만큼의 id를 설정할 수 있습니다. _`중복된 id 생성을 방지하기위해 twitter snowflake 알고리즘으로 id를 생성합니다.`_ | -| **netx.node-name** | pay-1 | _`$netx.group`_ 에 참여할 서버의 이름입니다. 같은 그룹내에 중복된 이름이 존재하면 안됩니다. | +| key | example | description | +|-------------------------|-----------|----------------------------------------------------------------------------------------------------------------------------------------------------| +| **netx.mode** | redis | 트랜잭션 관리에 사용할 메시지 큐 구현체의 mode 입니다. | +| **netx.host** | localhost | 트랜잭션 관리에 사용할 메시지 큐 의 host url 입니다. (ex. redis host) | +| **netx.port** | 6379 | 트랜잭션 관리에 사용할 메시지 큐의 port 입니다. | +| **netx.group** | pay-group | 분산 노드의 그룹입니다. 트랜잭션 이벤트는 같은 그룹내 하나의 노드로만 전송됩니다. | +| **netx.node-id** | 1 | id 생성에 사용될 식별자입니다. 모든 서버는 반드시 다른 id를 할당받아야 하며, 1~256 만큼의 id를 설정할 수 있습니다. _`중복된 id 생성을 방지하기위해 twitter snowflake 알고리즘으로 id를 생성합니다.`_ | +| **netx.node-name** | pay-1 | _`netx.group`_ 에 참여할 서버의 이름입니다. 같은 그룹내에 중복된 이름이 존재하면 안됩니다. | +| **netx.recovery-milli** | 60000 | _`netx.recovery-milli`_ 마다 _`netx.orphan-milli`_ 동안 처리 되지 않는 트랜잭션을 찾아 재실행합니다. 기본값은 60000(60초) 입니다. | +| **netx.orphan-milli** | 10000 | 트랜잭션이 PENDING 상태가 되었지만 orphan-milli가 지나도 ACK 상태가 되지 않는경우 다른 노드에게 처리를 위임합니다. 기본값은 10000(10초) 입니다. | ### Usage example diff --git a/build.gradle b/build.gradle index c643f27..e9761c0 100644 --- a/build.gradle +++ b/build.gradle @@ -35,7 +35,7 @@ publishing { } } -apply from: "gradle/mq.gradle" +apply from: "gradle/db.gradle" apply from: "gradle/test.gradle" apply from: "gradle/core.gradle" apply from: "gradle/sonar.gradle" diff --git a/gradle.properties b/gradle.properties index 0a8618e..a26f5c1 100644 --- a/gradle.properties +++ b/gradle.properties @@ -34,6 +34,9 @@ snowflakeVersion=5.2.5 ### Lettuce ### lettuceVersion=6.3.0.RELEASE +### Redisson ### +redissonVersion=3.26.0 + ### TestContainer ### testContainerVersion=1.19.3 diff --git a/gradle/mq.gradle b/gradle/db.gradle similarity index 56% rename from gradle/mq.gradle rename to gradle/db.gradle index d2ea2c7..f2cfa17 100644 --- a/gradle/mq.gradle +++ b/gradle/db.gradle @@ -1,3 +1,4 @@ dependencies { implementation "io.lettuce:lettuce-core:${lettuceVersion}" + implementation "org.redisson:redisson:${redissonVersion}" } diff --git a/idl b/idl index d0ea8e0..ef759ce 160000 --- a/idl +++ b/idl @@ -1 +1 @@ -Subproject commit d0ea8e0b64253d3e7966e2b6d95a751f40fe27de +Subproject commit ef759ceb6d294dcf391b0563bb883db68bc787de diff --git a/src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionDispatcher.kt b/src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionDispatcher.kt index 36bcfa7..92e40f5 100644 --- a/src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionDispatcher.kt +++ b/src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionDispatcher.kt @@ -6,38 +6,43 @@ import org.rooftop.netx.api.TransactionRollbackEvent import org.rooftop.netx.api.TransactionStartEvent import org.rooftop.netx.idl.Transaction import org.rooftop.netx.idl.TransactionState -import org.springframework.context.event.EventListener +import org.springframework.context.ApplicationEventPublisher import reactor.core.publisher.Flux import reactor.core.publisher.Mono abstract class AbstractTransactionDispatcher( - private val eventPublisher: EventPublisher, + private val eventPublisher: ApplicationEventPublisher, ) { - @EventListener(SubscribeTransactionEvent::class) - fun subscribeStream(event: SubscribeTransactionEvent): Flux { - return receive(event) - .dispatch() + fun subscribeStream(transactionId: String): Flux> { + return receive(transactionId) + .flatMap { dispatchAndAck(it.first, it.second) } } - protected abstract fun receive(event: SubscribeTransactionEvent): Flux + protected abstract fun receive(transactionId: String): Flux> - private fun Flux.dispatch(): Flux { - return this.flatMap { - when (it.state) { - TransactionState.TRANSACTION_STATE_JOIN -> publishJoin(it) - TransactionState.TRANSACTION_STATE_COMMIT -> publishCommit(it) - TransactionState.TRANSACTION_STATE_ROLLBACK -> publishRollback(it) - TransactionState.TRANSACTION_STATE_START -> publishStart(it) - else -> error("Cannot find matched transaction state \"${it.state}\"") - } + fun dispatchAndAck(transaction: Transaction, messageId: String): Flux> { + return Flux.just(transaction to messageId) + .dispatch() + .ack() + } + + private fun Flux>.dispatch(): Flux> { + return this.flatMap { (transaction, messageId) -> + when (transaction.state) { + TransactionState.TRANSACTION_STATE_JOIN -> publishJoin(transaction) + TransactionState.TRANSACTION_STATE_COMMIT -> publishCommit(transaction) + TransactionState.TRANSACTION_STATE_ROLLBACK -> publishRollback(transaction) + TransactionState.TRANSACTION_STATE_START -> publishStart(transaction) + else -> error("Cannot find matched transaction state \"${transaction.state}\"") + }.map { transaction to messageId } } } private fun publishJoin(it: Transaction): Mono { return Mono.just(it) .doOnNext { - eventPublisher.publish( + eventPublisher.publishEvent( TransactionJoinEvent( it.id, it.serverId @@ -48,13 +53,13 @@ abstract class AbstractTransactionDispatcher( private fun publishCommit(it: Transaction): Mono { return Mono.just(it) - .doOnNext { eventPublisher.publish(TransactionCommitEvent(it.id, it.serverId)) } + .doOnNext { eventPublisher.publishEvent(TransactionCommitEvent(it.id, it.serverId)) } } private fun publishRollback(transaction: Transaction): Mono { return findOwnTransaction(transaction) .doOnNext { - eventPublisher.publish( + eventPublisher.publishEvent( TransactionRollbackEvent( transaction.id, transaction.serverId, @@ -71,7 +76,9 @@ abstract class AbstractTransactionDispatcher( private fun publishStart(it: Transaction): Mono { return Mono.just(it) .doOnNext { - eventPublisher.publish(TransactionStartEvent(it.id, it.serverId)) + eventPublisher.publishEvent(TransactionStartEvent(it.id, it.serverId)) } } + + protected abstract fun Flux>.ack(): Flux> } diff --git a/src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionManager.kt b/src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionManager.kt index 357e36f..12a1f10 100644 --- a/src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionManager.kt +++ b/src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionManager.kt @@ -5,59 +5,75 @@ import org.rooftop.netx.idl.Transaction import org.rooftop.netx.idl.TransactionState import org.rooftop.netx.idl.transaction import reactor.core.publisher.Mono +import reactor.core.scheduler.Schedulers abstract class AbstractTransactionManager( nodeId: Int, + private val nodeGroup: String, private val nodeName: String, - private val eventPublisher: EventPublisher, private val transactionIdGenerator: TransactionIdGenerator = TransactionIdGenerator(nodeId), + private val transactionDispatcher: AbstractTransactionDispatcher, + private val transactionRetrySupporter: AbstractTransactionRetrySupporter, ) : TransactionManager { final override fun start(undo: String): Mono { - return startTransaction() + return startTransaction(undo) .subscribeTransaction() + .watchTransaction() .contextWrite { it.put(CONTEXT_TX_KEY, transactionIdGenerator.generate()) } } - private fun startTransaction(): Mono { + private fun startTransaction(undo: String): Mono { return Mono.deferContextual { Mono.just(it[CONTEXT_TX_KEY]) } .flatMap { transactionId -> publishTransaction(transactionId, transaction { id = transactionId serverId = nodeName + group = nodeGroup this.state = TransactionState.TRANSACTION_STATE_START + this.undo = undo }) } } final override fun join(transactionId: String, undo: String): Mono { return exists(transactionId) - .joinTransaction() + .joinTransaction(undo) .subscribeTransaction() + .watchTransaction() .contextWrite { it.put(CONTEXT_TX_KEY, transactionId) } } - private fun Mono.joinTransaction(): Mono { + private fun Mono.joinTransaction(undo: String): Mono { return flatMap { transactionId -> publishTransaction(transactionId, transaction { id = transactionId serverId = nodeName + group = nodeGroup state = TransactionState.TRANSACTION_STATE_JOIN + this.undo = undo }) } } private fun Mono.subscribeTransaction(): Mono { return this.doOnSuccess { - eventPublisher.publish(SubscribeTransactionEvent(it)) + transactionDispatcher.subscribeStream(it) + .subscribeOn(Schedulers.parallel()) + .subscribe() } } + private fun Mono.watchTransaction(): Mono { + return this.flatMap { transactionRetrySupporter.watchTransaction(it) } + } + final override fun rollback(transactionId: String, cause: String): Mono { return exists(transactionId) .publishTransaction(transaction { id = transactionId serverId = nodeName + group = nodeGroup state = TransactionState.TRANSACTION_STATE_ROLLBACK this.cause = cause }) @@ -69,6 +85,7 @@ abstract class AbstractTransactionManager( .publishTransaction(transaction { id = transactionId serverId = nodeName + group = nodeGroup state = TransactionState.TRANSACTION_STATE_COMMIT }) .contextWrite { it.put(CONTEXT_TX_KEY, transactionId) } diff --git a/src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionRetrySupporter.kt b/src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionRetrySupporter.kt new file mode 100644 index 0000000..70e9305 --- /dev/null +++ b/src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionRetrySupporter.kt @@ -0,0 +1,24 @@ +package org.rooftop.netx.engine + +import org.rooftop.netx.idl.Transaction +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono +import reactor.core.scheduler.Schedulers +import kotlin.time.Duration.Companion.milliseconds +import kotlin.time.toJavaDuration + +abstract class AbstractTransactionRetrySupporter( + recoveryMilli: Long, +) { + + init { + Flux.interval(recoveryMilli.milliseconds.toJavaDuration()) + .publishOn(Schedulers.parallel()) + .flatMap { claimOrphanTransaction() } + .subscribe() + } + + abstract fun watchTransaction(transactionId: String): Mono + + abstract fun claimOrphanTransaction(): Flux> +} diff --git a/src/main/kotlin/org/rooftop/netx/engine/EventPublisher.kt b/src/main/kotlin/org/rooftop/netx/engine/EventPublisher.kt deleted file mode 100644 index c57d744..0000000 --- a/src/main/kotlin/org/rooftop/netx/engine/EventPublisher.kt +++ /dev/null @@ -1,6 +0,0 @@ -package org.rooftop.netx.engine - -fun interface EventPublisher { - - fun publish(event: Any) -} diff --git a/src/main/kotlin/org/rooftop/netx/engine/SubscribeTransactionEvent.kt b/src/main/kotlin/org/rooftop/netx/engine/SubscribeTransactionEvent.kt deleted file mode 100644 index 79e8447..0000000 --- a/src/main/kotlin/org/rooftop/netx/engine/SubscribeTransactionEvent.kt +++ /dev/null @@ -1,5 +0,0 @@ -package org.rooftop.netx.engine - -data class SubscribeTransactionEvent( - val transactionId: String -) diff --git a/src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionDispatcher.kt b/src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionDispatcher.kt index f1ca9e6..9d21605 100644 --- a/src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionDispatcher.kt +++ b/src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionDispatcher.kt @@ -1,7 +1,6 @@ package org.rooftop.netx.redis import org.rooftop.netx.engine.AbstractTransactionDispatcher -import org.rooftop.netx.engine.SubscribeTransactionEvent import org.rooftop.netx.idl.Transaction import org.rooftop.netx.idl.TransactionState import org.springframework.context.ApplicationEventPublisher @@ -20,10 +19,10 @@ import kotlin.time.toJavaDuration class RedisStreamTransactionDispatcher( eventPublisher: ApplicationEventPublisher, connectionFactory: ReactiveRedisConnectionFactory, - private val streamGroup: String, + private val nodeGroup: String, private val nodeName: String, private val reactiveRedisTemplate: ReactiveRedisTemplate, -) : AbstractTransactionDispatcher(SpringEventPublisher(eventPublisher)) { +) : AbstractTransactionDispatcher(eventPublisher) { private val options = StreamReceiver.StreamReceiverOptions.builder() .pollTimeout(1.hours.toJavaDuration()) @@ -31,36 +30,36 @@ class RedisStreamTransactionDispatcher( private val receiver = StreamReceiver.create(connectionFactory, options) - override fun receive(event: SubscribeTransactionEvent): Flux { - return createGroupIfNotExists(event) + override fun receive(transactionId: String): Flux> { + return createGroupIfNotExists(transactionId) .flatMap { - receiver.receiveAutoAck( - Consumer.from(streamGroup, nodeName), - StreamOffset.create(event.transactionId, ReadOffset.from(">")) + receiver.receive( + Consumer.from(nodeGroup, nodeName), + StreamOffset.create(transactionId, ReadOffset.from(">")) ).publishOn(Schedulers.parallel()) - .map { Transaction.parseFrom(it.value["data"]?.toByteArray()) } - .flatMap { - when (it.state) { + .map { Transaction.parseFrom(it.value["data"]?.toByteArray()) to it.id.value } + .flatMap { (transaction, messageId) -> + when (transaction.state) { TransactionState.TRANSACTION_STATE_ROLLBACK -> - findOwnTransaction(it) + findOwnTransaction(transaction).map { it to messageId } - else -> Mono.just(it) + else -> Mono.just(transaction to messageId) } } } } - private fun createGroupIfNotExists(event: SubscribeTransactionEvent): Flux { + private fun createGroupIfNotExists(transactionId: String): Flux { return reactiveRedisTemplate.opsForStream() - .createGroup(event.transactionId, ReadOffset.from("0"), streamGroup) + .createGroup(transactionId, ReadOffset.from("0"), nodeGroup) .flatMapMany { Flux.just(it) } } override fun findOwnTransaction(transaction: Transaction): Mono { - return reactiveRedisTemplate.opsForStream() + return reactiveRedisTemplate.opsForStream() .read(StreamOffset.create(transaction.id, ReadOffset.from("0"))) - .map { Transaction.parseFrom(it.value["data"]!!) } - .filter { it.group == streamGroup } + .map { Transaction.parseFrom(it.value["data"]?.toByteArray()) } + .filter { it.group == nodeGroup } .filter { hasUndo(it) } .next() } @@ -68,4 +67,12 @@ class RedisStreamTransactionDispatcher( private fun hasUndo(transaction: Transaction): Boolean = transaction.state == TransactionState.TRANSACTION_STATE_JOIN || transaction.state == TransactionState.TRANSACTION_STATE_START + + override fun Flux>.ack(): Flux> { + return this.flatMap { (transaction, messageId) -> + reactiveRedisTemplate.opsForStream() + .acknowledge(transaction.id, nodeGroup, messageId) + .flatMapMany { Flux.just(transaction to messageId) } + } + } } diff --git a/src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionManager.kt b/src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionManager.kt index f213f47..7c3a3e8 100644 --- a/src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionManager.kt +++ b/src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionManager.kt @@ -2,7 +2,6 @@ package org.rooftop.netx.redis import org.rooftop.netx.engine.AbstractTransactionManager import org.rooftop.netx.idl.Transaction -import org.springframework.context.ApplicationEventPublisher import org.springframework.data.domain.Range import org.springframework.data.redis.connection.stream.Record import org.springframework.data.redis.core.ReactiveRedisTemplate @@ -11,12 +10,16 @@ import reactor.core.publisher.Mono class RedisStreamTransactionManager( nodeId: Int, nodeName: String, - applicationEventPublisher: ApplicationEventPublisher, + nodeGroup: String, + transactionDispatcher: RedisStreamTransactionDispatcher, + transactionRetrySupporter: RedisTransactionRetrySupporter, private val reactiveRedisTemplate: ReactiveRedisTemplate, ) : AbstractTransactionManager( - nodeId, - nodeName, - SpringEventPublisher(applicationEventPublisher), + nodeId = nodeId, + nodeName = nodeName, + nodeGroup = nodeGroup, + transactionDispatcher = transactionDispatcher, + transactionRetrySupporter = transactionRetrySupporter, ) { override fun findAnyTransaction(transactionId: String): Mono { diff --git a/src/main/kotlin/org/rooftop/netx/redis/RedisTransactionConfigurer.kt b/src/main/kotlin/org/rooftop/netx/redis/RedisTransactionConfigurer.kt index e4f8708..b3b9030 100644 --- a/src/main/kotlin/org/rooftop/netx/redis/RedisTransactionConfigurer.kt +++ b/src/main/kotlin/org/rooftop/netx/redis/RedisTransactionConfigurer.kt @@ -1,5 +1,8 @@ package org.rooftop.netx.redis +import org.redisson.Redisson +import org.redisson.api.RedissonReactiveClient +import org.redisson.config.Config import org.rooftop.netx.api.TransactionManager import org.rooftop.pay.infra.transaction.ByteArrayRedisSerializer import org.springframework.beans.factory.annotation.Value @@ -17,9 +20,11 @@ import org.springframework.data.redis.serializer.StringRedisSerializer class RedisTransactionConfigurer( @Value("\${netx.host}") private val host: String, @Value("\${netx.port}") private val port: String, - @Value("\${netx.group}") private val group: String, + @Value("\${netx.group}") private val nodeGroup: String, @Value("\${netx.node-id}") private val nodeId: Int, @Value("\${netx.node-name}") private val nodeName: String, + @Value("\${netx.recovery-milli:60000}") private val recoveryMilli: Long, + @Value("\${netx.orphan-milli:10000}") private val orphanMilli: Long, private val applicationEventPublisher: ApplicationEventPublisher, ) { @@ -27,21 +32,36 @@ class RedisTransactionConfigurer( @ConditionalOnProperty(prefix = "netx", name = ["mode"], havingValue = "redis") fun redisStreamTransactionManager(): TransactionManager = RedisStreamTransactionManager( - nodeId, - nodeName, - applicationEventPublisher, - reactiveRedisTemplate() + nodeId = nodeId, + nodeName = nodeName, + nodeGroup = nodeGroup, + transactionDispatcher = redisStreamTransactionDispatcher(), + transactionRetrySupporter = redisTransactionRetrySupporter(), + reactiveRedisTemplate = reactiveRedisTemplate(), ) @Bean @ConditionalOnProperty(prefix = "netx", name = ["mode"], havingValue = "redis") fun redisStreamTransactionDispatcher(): RedisStreamTransactionDispatcher = RedisStreamTransactionDispatcher( - applicationEventPublisher, - reactiveRedisConnectionFactory(), - group, - nodeName, - reactiveRedisTemplate() + eventPublisher = applicationEventPublisher, + connectionFactory = reactiveRedisConnectionFactory(), + nodeGroup = nodeGroup, + nodeName = nodeName, + reactiveRedisTemplate = reactiveRedisTemplate() + ) + + @Bean + @ConditionalOnProperty(prefix = "netx", name = ["mode"], havingValue = "redis") + fun redisTransactionRetrySupporter(): RedisTransactionRetrySupporter = + RedisTransactionRetrySupporter( + nodeGroup = nodeGroup, + nodeName = nodeName, + reactiveRedisTemplate = reactiveRedisTemplate(), + redissonReactiveClient = redissonReactiveClient(), + transactionDispatcher = redisStreamTransactionDispatcher(), + orphanMilli = orphanMilli, + recoveryMilli = recoveryMilli, ) @Bean @@ -56,6 +76,18 @@ class RedisTransactionConfigurer( return ReactiveRedisTemplate(reactiveRedisConnectionFactory(), context) } + @Bean + @ConditionalOnProperty(prefix = "netx", name = ["mode"], havingValue = "redis") + fun redissonReactiveClient(): RedissonReactiveClient { + val port: String = System.getProperty("netx.port") ?: port + + return Redisson.create(Config() + .also { + it.useSingleServer() + .setAddress("redis://$host:$port") + }).reactive() + } + @Bean @ConditionalOnProperty(prefix = "netx", name = ["mode"], havingValue = "redis") fun byteArrayRedisSerializer(): ByteArrayRedisSerializer { diff --git a/src/main/kotlin/org/rooftop/netx/redis/RedisTransactionRetrySupporter.kt b/src/main/kotlin/org/rooftop/netx/redis/RedisTransactionRetrySupporter.kt new file mode 100644 index 0000000..b566229 --- /dev/null +++ b/src/main/kotlin/org/rooftop/netx/redis/RedisTransactionRetrySupporter.kt @@ -0,0 +1,62 @@ +package org.rooftop.netx.redis + +import org.redisson.api.RedissonReactiveClient +import org.rooftop.netx.engine.AbstractTransactionRetrySupporter +import org.rooftop.netx.idl.Transaction +import org.springframework.data.domain.Range +import org.springframework.data.redis.connection.RedisStreamCommands.XClaimOptions +import org.springframework.data.redis.core.ReactiveRedisTemplate +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono +import reactor.core.scheduler.Schedulers +import java.util.concurrent.TimeUnit + +class RedisTransactionRetrySupporter( + private val nodeGroup: String, + private val nodeName: String, + private val reactiveRedisTemplate: ReactiveRedisTemplate, + private val redissonReactiveClient: RedissonReactiveClient, + private val transactionDispatcher: RedisStreamTransactionDispatcher, + private val orphanMilli: Long, + recoveryMilli: Long, +) : AbstractTransactionRetrySupporter(recoveryMilli) { + + override fun watchTransaction(transactionId: String): Mono { + return reactiveRedisTemplate.opsForSet() + .add(nodeGroup, transactionId.toByteArray()) + .map { transactionId } + } + + override fun claimOrphanTransaction(): Flux> { + return reactiveRedisTemplate.opsForSet() + .members(nodeGroup) + .flatMap { claimTransactions(String(it)) } + .publishOn(Schedulers.parallel()) + .flatMap { transactionDispatcher.dispatchAndAck(it.first, it.second) } + } + + private fun claimTransactions(transactionId: String): Flux> { + return reactiveRedisTemplate.opsForStream() + .pending(transactionId, nodeGroup, Range.closed("-", "+"), Long.MAX_VALUE) + .filter { it.get().toList().isNotEmpty() } + .flatMap { pendingMessage -> + redissonReactiveClient.getLock("$nodeGroup-key") + .tryLock(0, orphanMilli, TimeUnit.MILLISECONDS) + .map { pendingMessage } + } + .flatMapMany { + reactiveRedisTemplate.opsForStream() + .claim( + transactionId, nodeGroup, nodeName, XClaimOptions + .minIdleMs(orphanMilli) + .ids(it.get().toList()) + ) + } + .map { Transaction.parseFrom(it.value["data"]) to it.id.toString() } + .flatMap { + redissonReactiveClient.getLock("$nodeGroup-key") + .unlock() + .thenMany { it } + } + } +} diff --git a/src/main/kotlin/org/rooftop/netx/redis/SpringEventPublisher.kt b/src/main/kotlin/org/rooftop/netx/redis/SpringEventPublisher.kt deleted file mode 100644 index aa84518..0000000 --- a/src/main/kotlin/org/rooftop/netx/redis/SpringEventPublisher.kt +++ /dev/null @@ -1,11 +0,0 @@ -package org.rooftop.netx.redis - -import org.rooftop.netx.engine.EventPublisher -import org.springframework.context.ApplicationEventPublisher - -class SpringEventPublisher(private val eventPublisher: ApplicationEventPublisher) : EventPublisher { - - override fun publish(event: Any) { - eventPublisher.publishEvent(event) - } -} diff --git a/src/test/kotlin/org/rooftop/netx/redis/RedisContainer.kt b/src/test/kotlin/org/rooftop/netx/redis/RedisContainer.kt index 8329778..1491200 100644 --- a/src/test/kotlin/org/rooftop/netx/redis/RedisContainer.kt +++ b/src/test/kotlin/org/rooftop/netx/redis/RedisContainer.kt @@ -17,9 +17,5 @@ class RedisContainer { "netx.port", redis.getMappedPort(6379).toString() ) - System.setProperty( - "netx.undo.port", - redis.getMappedPort(6379).toString() - ) } } diff --git a/src/test/kotlin/org/rooftop/netx/redis/RedisStreamTransactionManagerTest.kt b/src/test/kotlin/org/rooftop/netx/redis/RedisStreamTransactionManagerTest.kt index a81ecef..2c1906d 100644 --- a/src/test/kotlin/org/rooftop/netx/redis/RedisStreamTransactionManagerTest.kt +++ b/src/test/kotlin/org/rooftop/netx/redis/RedisStreamTransactionManagerTest.kt @@ -131,6 +131,8 @@ internal class RedisStreamTransactionManagerTest( eventually(5.minutes) { eventCapture.capturedCount(TransactionRollbackEvent::class) } + + Thread.sleep(10.minutes.inWholeMilliseconds) } } diff --git a/src/test/resources/application.properties b/src/test/resources/application.properties index b0cdb23..0c78093 100644 --- a/src/test/resources/application.properties +++ b/src/test/resources/application.properties @@ -4,3 +4,5 @@ netx.port=6379 netx.group=netx-group netx.node-id=1 netx.node-name=netx-node +netx.recovery-milli=5000 +netx.orphan-milli=1000 From 0acb5ef00e1a30b03428addf8ff2f34cccd64d9c Mon Sep 17 00:00:00 2001 From: devxb Date: Sat, 10 Feb 2024 13:36:59 +0900 Subject: [PATCH 3/5] =?UTF-8?q?fix:=20RedisRetrySupporter=EA=B0=80=20Pendi?= =?UTF-8?q?ng=20=EC=83=81=ED=83=9C=EC=9D=98=20=EB=A9=94=EC=8B=9C=EC=A7=80?= =?UTF-8?q?=EB=A5=BC=20=ED=8C=8C=EC=8B=B1=ED=95=98=EC=A7=80=20=EB=AA=BB?= =?UTF-8?q?=ED=95=98=EB=8A=94=20=EB=B2=84=EA=B7=B8=EB=A5=BC=20=EC=88=98?= =?UTF-8?q?=EC=A0=95=ED=95=9C=EB=8B=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../redis/RedisTransactionRetrySupporter.kt | 21 ++++++++++++------- 1 file changed, 14 insertions(+), 7 deletions(-) diff --git a/src/main/kotlin/org/rooftop/netx/redis/RedisTransactionRetrySupporter.kt b/src/main/kotlin/org/rooftop/netx/redis/RedisTransactionRetrySupporter.kt index b566229..045eff6 100644 --- a/src/main/kotlin/org/rooftop/netx/redis/RedisTransactionRetrySupporter.kt +++ b/src/main/kotlin/org/rooftop/netx/redis/RedisTransactionRetrySupporter.kt @@ -1,6 +1,7 @@ package org.rooftop.netx.redis import org.redisson.api.RedissonReactiveClient +import org.rooftop.netx.engine.AbstractTransactionDispatcher import org.rooftop.netx.engine.AbstractTransactionRetrySupporter import org.rooftop.netx.idl.Transaction import org.springframework.data.domain.Range @@ -16,7 +17,7 @@ class RedisTransactionRetrySupporter( private val nodeName: String, private val reactiveRedisTemplate: ReactiveRedisTemplate, private val redissonReactiveClient: RedissonReactiveClient, - private val transactionDispatcher: RedisStreamTransactionDispatcher, + private val transactionDispatcher: AbstractTransactionDispatcher, private val orphanMilli: Long, recoveryMilli: Long, ) : AbstractTransactionRetrySupporter(recoveryMilli) { @@ -27,7 +28,7 @@ class RedisTransactionRetrySupporter( .map { transactionId } } - override fun claimOrphanTransaction(): Flux> { + override fun handleOrphanTransaction(): Flux> { return reactiveRedisTemplate.opsForSet() .members(nodeGroup) .flatMap { claimTransactions(String(it)) } @@ -45,18 +46,24 @@ class RedisTransactionRetrySupporter( .map { pendingMessage } } .flatMapMany { - reactiveRedisTemplate.opsForStream() + reactiveRedisTemplate.opsForStream() .claim( transactionId, nodeGroup, nodeName, XClaimOptions .minIdleMs(orphanMilli) - .ids(it.get().toList()) + .ids(it.get().map { eachMessage -> eachMessage.id.value }.toList()) ) } - .map { Transaction.parseFrom(it.value["data"]) to it.id.toString() } - .flatMap { + .map { Transaction.parseFrom(it.value["data"]?.toByteArray()) to it.id.toString() } + .flatMap { transactionWithMessageId -> redissonReactiveClient.getLock("$nodeGroup-key") .unlock() - .thenMany { it } + .flatMapMany { Flux.just(transactionWithMessageId) } + } + .doOnError { + redissonReactiveClient.getLock("$nodeGroup-key") + .unlock() + .subscribeOn(Schedulers.parallel()) + .subscribe() } } } From ca7b2e216025a1ce0f0141c10c4382858f051afa Mon Sep 17 00:00:00 2001 From: devxb Date: Sat, 10 Feb 2024 14:32:42 +0900 Subject: [PATCH 4/5] =?UTF-8?q?feat:=20pending=EB=90=98=EC=97=88=EC=A7=80?= =?UTF-8?q?=EB=A7=8C,=20ack=EB=90=98=EC=A7=80=EC=95=8A=EC=9D=80=20?= =?UTF-8?q?=ED=8A=B8=EB=9E=9C=EC=9E=AD=EC=85=98=EC=9D=84=20=EC=B0=BE?= =?UTF-8?q?=EC=95=84=EC=84=9C=20=EC=B2=98=EB=A6=AC=ED=95=98=EB=8A=94=20?= =?UTF-8?q?=EA=B8=B0=EB=8A=A5=EC=9D=84=20=EC=B6=94=EA=B0=80=ED=95=9C?= =?UTF-8?q?=EB=8B=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../AbstractTransactionRetrySupporter.kt | 4 +- .../redis/RedisStreamTransactionManager.kt | 6 +- .../redis/RedisTransactionRetrySupporter.kt | 2 +- .../NoAckRedisStreamTransactionDispatcher.kt | 73 +++++++++++ .../redis/NoAckRedisTransactionConfigurer.kt | 117 ++++++++++++++++++ .../org/rooftop/netx/redis/RedisAssertions.kt | 23 ++++ .../RedisTransactionRetrySupporterTest.kt | 38 ++++++ src/test/resources/application.properties | 2 +- 8 files changed, 259 insertions(+), 6 deletions(-) create mode 100644 src/test/kotlin/org/rooftop/netx/redis/NoAckRedisStreamTransactionDispatcher.kt create mode 100644 src/test/kotlin/org/rooftop/netx/redis/NoAckRedisTransactionConfigurer.kt create mode 100644 src/test/kotlin/org/rooftop/netx/redis/RedisAssertions.kt create mode 100644 src/test/kotlin/org/rooftop/netx/redis/RedisTransactionRetrySupporterTest.kt diff --git a/src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionRetrySupporter.kt b/src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionRetrySupporter.kt index 70e9305..3f063c9 100644 --- a/src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionRetrySupporter.kt +++ b/src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionRetrySupporter.kt @@ -14,11 +14,11 @@ abstract class AbstractTransactionRetrySupporter( init { Flux.interval(recoveryMilli.milliseconds.toJavaDuration()) .publishOn(Schedulers.parallel()) - .flatMap { claimOrphanTransaction() } + .flatMap { handleOrphanTransaction() } .subscribe() } abstract fun watchTransaction(transactionId: String): Mono - abstract fun claimOrphanTransaction(): Flux> + protected abstract fun handleOrphanTransaction(): Flux> } diff --git a/src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionManager.kt b/src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionManager.kt index 7c3a3e8..2a46bf8 100644 --- a/src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionManager.kt +++ b/src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionManager.kt @@ -1,6 +1,8 @@ package org.rooftop.netx.redis +import org.rooftop.netx.engine.AbstractTransactionDispatcher import org.rooftop.netx.engine.AbstractTransactionManager +import org.rooftop.netx.engine.AbstractTransactionRetrySupporter import org.rooftop.netx.idl.Transaction import org.springframework.data.domain.Range import org.springframework.data.redis.connection.stream.Record @@ -11,8 +13,8 @@ class RedisStreamTransactionManager( nodeId: Int, nodeName: String, nodeGroup: String, - transactionDispatcher: RedisStreamTransactionDispatcher, - transactionRetrySupporter: RedisTransactionRetrySupporter, + transactionDispatcher: AbstractTransactionDispatcher, + transactionRetrySupporter: AbstractTransactionRetrySupporter, private val reactiveRedisTemplate: ReactiveRedisTemplate, ) : AbstractTransactionManager( nodeId = nodeId, diff --git a/src/main/kotlin/org/rooftop/netx/redis/RedisTransactionRetrySupporter.kt b/src/main/kotlin/org/rooftop/netx/redis/RedisTransactionRetrySupporter.kt index 045eff6..62373d7 100644 --- a/src/main/kotlin/org/rooftop/netx/redis/RedisTransactionRetrySupporter.kt +++ b/src/main/kotlin/org/rooftop/netx/redis/RedisTransactionRetrySupporter.kt @@ -56,7 +56,7 @@ class RedisTransactionRetrySupporter( .map { Transaction.parseFrom(it.value["data"]?.toByteArray()) to it.id.toString() } .flatMap { transactionWithMessageId -> redissonReactiveClient.getLock("$nodeGroup-key") - .unlock() + .forceUnlock() .flatMapMany { Flux.just(transactionWithMessageId) } } .doOnError { diff --git a/src/test/kotlin/org/rooftop/netx/redis/NoAckRedisStreamTransactionDispatcher.kt b/src/test/kotlin/org/rooftop/netx/redis/NoAckRedisStreamTransactionDispatcher.kt new file mode 100644 index 0000000..6f79e54 --- /dev/null +++ b/src/test/kotlin/org/rooftop/netx/redis/NoAckRedisStreamTransactionDispatcher.kt @@ -0,0 +1,73 @@ +package org.rooftop.netx.redis + +import org.rooftop.netx.engine.AbstractTransactionDispatcher +import org.rooftop.netx.idl.Transaction +import org.rooftop.netx.idl.TransactionState +import org.springframework.context.ApplicationEventPublisher +import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory +import org.springframework.data.redis.connection.stream.Consumer +import org.springframework.data.redis.connection.stream.ReadOffset +import org.springframework.data.redis.connection.stream.StreamOffset +import org.springframework.data.redis.core.ReactiveRedisTemplate +import org.springframework.data.redis.stream.StreamReceiver +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono +import reactor.core.scheduler.Schedulers +import kotlin.time.Duration.Companion.hours +import kotlin.time.toJavaDuration + +class NoAckRedisStreamTransactionDispatcher( + eventPublisher: ApplicationEventPublisher, + connectionFactory: ReactiveRedisConnectionFactory, + private val nodeGroup: String, + private val nodeName: String, + private val reactiveRedisTemplate: ReactiveRedisTemplate, +) : AbstractTransactionDispatcher(eventPublisher) { + + private val options = StreamReceiver.StreamReceiverOptions.builder() + .pollTimeout(1.hours.toJavaDuration()) + .build() + + private val receiver = StreamReceiver.create(connectionFactory, options) + + override fun receive(transactionId: String): Flux> { + return createGroupIfNotExists(transactionId) + .flatMap { + receiver.receive( + Consumer.from(nodeGroup, nodeName), + StreamOffset.create(transactionId, ReadOffset.from(">")) + ).publishOn(Schedulers.parallel()) + .map { Transaction.parseFrom(it.value["data"]?.toByteArray()) to it.id.value } + .flatMap { (transaction, messageId) -> + when (transaction.state) { + TransactionState.TRANSACTION_STATE_ROLLBACK -> + findOwnTransaction(transaction).map { it to messageId } + + else -> Mono.just(transaction to messageId) + } + } + } + } + + private fun createGroupIfNotExists(transactionId: String): Flux { + return reactiveRedisTemplate.opsForStream() + .createGroup(transactionId, ReadOffset.from("0"), nodeGroup) + .flatMapMany { Flux.just(it) } + } + + override fun findOwnTransaction(transaction: Transaction): Mono { + return reactiveRedisTemplate.opsForStream() + .read(StreamOffset.create(transaction.id, ReadOffset.from("0"))) + .map { Transaction.parseFrom(it.value["data"]?.toByteArray()) } + .filter { it.group == nodeGroup } + .filter { hasUndo(it) } + .next() + } + + private fun hasUndo(transaction: Transaction): Boolean = + transaction.state == TransactionState.TRANSACTION_STATE_JOIN + || transaction.state == TransactionState.TRANSACTION_STATE_START + + override fun Flux>.ack(): Flux> = this +} + diff --git a/src/test/kotlin/org/rooftop/netx/redis/NoAckRedisTransactionConfigurer.kt b/src/test/kotlin/org/rooftop/netx/redis/NoAckRedisTransactionConfigurer.kt new file mode 100644 index 0000000..64d8cbe --- /dev/null +++ b/src/test/kotlin/org/rooftop/netx/redis/NoAckRedisTransactionConfigurer.kt @@ -0,0 +1,117 @@ +package org.rooftop.netx.redis + +import org.redisson.Redisson +import org.redisson.api.RedissonReactiveClient +import org.redisson.config.Config +import org.rooftop.netx.api.TransactionManager +import org.rooftop.pay.infra.transaction.ByteArrayRedisSerializer +import org.springframework.beans.factory.annotation.Value +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty +import org.springframework.boot.test.context.TestConfiguration +import org.springframework.context.ApplicationEventPublisher +import org.springframework.context.annotation.Bean +import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory +import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory +import org.springframework.data.redis.core.ReactiveRedisTemplate +import org.springframework.data.redis.serializer.RedisSerializationContext +import org.springframework.data.redis.serializer.StringRedisSerializer + +@TestConfiguration +class NoAckRedisTransactionConfigurer( + @Value("\${netx.host}") private val host: String, + @Value("\${netx.port}") private val port: String, + @Value("\${netx.group}") private val nodeGroup: String, + @Value("\${netx.node-id}") private val nodeId: Int, + @Value("\${netx.node-name}") private val nodeName: String, + @Value("\${netx.recovery-milli:60000}") private val recoveryMilli: Long, + @Value("\${netx.orphan-milli:10000}") private val orphanMilli: Long, + private val applicationEventPublisher: ApplicationEventPublisher, +) { + + @Bean + @ConditionalOnProperty(prefix = "netx", name = ["mode"], havingValue = "redis") + fun redisStreamTransactionManager(): TransactionManager = + RedisStreamTransactionManager( + nodeId = nodeId, + nodeName = nodeName, + nodeGroup = nodeGroup, + transactionDispatcher = noAckRedisStreamTransactionDispatcher(), + transactionRetrySupporter = redisTransactionRetrySupporter(), + reactiveRedisTemplate = reactiveRedisTemplate(), + ) + + @Bean + @ConditionalOnProperty(prefix = "netx", name = ["mode"], havingValue = "redis") + fun redisStreamTransactionDispatcher(): RedisStreamTransactionDispatcher = + RedisStreamTransactionDispatcher( + eventPublisher = applicationEventPublisher, + connectionFactory = reactiveRedisConnectionFactory(), + nodeGroup = nodeGroup, + nodeName = nodeName, + reactiveRedisTemplate = reactiveRedisTemplate() + ) + + @Bean + @ConditionalOnProperty(prefix = "netx", name = ["mode"], havingValue = "redis") + fun noAckRedisStreamTransactionDispatcher(): NoAckRedisStreamTransactionDispatcher = + NoAckRedisStreamTransactionDispatcher( + eventPublisher = applicationEventPublisher, + connectionFactory = reactiveRedisConnectionFactory(), + nodeGroup = nodeGroup, + nodeName = nodeName, + reactiveRedisTemplate = reactiveRedisTemplate() + ) + + @Bean + @ConditionalOnProperty(prefix = "netx", name = ["mode"], havingValue = "redis") + fun redisTransactionRetrySupporter(): RedisTransactionRetrySupporter = + RedisTransactionRetrySupporter( + nodeGroup = nodeGroup, + nodeName = nodeName, + reactiveRedisTemplate = reactiveRedisTemplate(), + redissonReactiveClient = redissonReactiveClient(), + transactionDispatcher = redisStreamTransactionDispatcher(), + orphanMilli = orphanMilli, + recoveryMilli = recoveryMilli, + ) + + @Bean + @ConditionalOnProperty(prefix = "netx", name = ["mode"], havingValue = "redis") + fun reactiveRedisTemplate(): ReactiveRedisTemplate { + val builder = RedisSerializationContext.newSerializationContext( + StringRedisSerializer() + ) + + val context = builder.value(byteArrayRedisSerializer()).build() + + return ReactiveRedisTemplate(reactiveRedisConnectionFactory(), context) + } + + @Bean + @ConditionalOnProperty(prefix = "netx", name = ["mode"], havingValue = "redis") + fun redissonReactiveClient(): RedissonReactiveClient { + val port: String = System.getProperty("netx.port") ?: port + + return Redisson.create( + Config() + .also { + it.useSingleServer() + .setAddress("redis://$host:$port") + }).reactive() + } + + @Bean + @ConditionalOnProperty(prefix = "netx", name = ["mode"], havingValue = "redis") + fun byteArrayRedisSerializer(): ByteArrayRedisSerializer { + return ByteArrayRedisSerializer() + } + + @Bean + @ConditionalOnProperty(prefix = "netx", name = ["mode"], havingValue = "redis") + fun reactiveRedisConnectionFactory(): ReactiveRedisConnectionFactory { + val port: String = System.getProperty("netx.port") ?: port + + return LettuceConnectionFactory(host, port.toInt()) + } +} + diff --git a/src/test/kotlin/org/rooftop/netx/redis/RedisAssertions.kt b/src/test/kotlin/org/rooftop/netx/redis/RedisAssertions.kt new file mode 100644 index 0000000..288bdb6 --- /dev/null +++ b/src/test/kotlin/org/rooftop/netx/redis/RedisAssertions.kt @@ -0,0 +1,23 @@ +package org.rooftop.netx.redis + +import io.kotest.matchers.shouldBe +import org.springframework.beans.factory.annotation.Value +import org.springframework.boot.test.context.TestComponent +import org.springframework.data.domain.Range +import org.springframework.data.redis.core.ReactiveRedisOperations + +@TestComponent +internal class RedisAssertions( + private val reactiveRedisOperations: ReactiveRedisOperations, + @Value("\${netx.group}") private val nodeGroup: String, +) { + + fun pendingMessageCountShouldBe(transactionId: String, count: Long) { + val pendingMessageCount = reactiveRedisOperations.opsForStream() + .pending(transactionId, nodeGroup, Range.closed("-", "+"), Long.MAX_VALUE) + .map { it.get().toList().size } + .block() + + pendingMessageCount shouldBe count + } +} diff --git a/src/test/kotlin/org/rooftop/netx/redis/RedisTransactionRetrySupporterTest.kt b/src/test/kotlin/org/rooftop/netx/redis/RedisTransactionRetrySupporterTest.kt new file mode 100644 index 0000000..147575a --- /dev/null +++ b/src/test/kotlin/org/rooftop/netx/redis/RedisTransactionRetrySupporterTest.kt @@ -0,0 +1,38 @@ +package org.rooftop.netx.redis + +import io.kotest.assertions.nondeterministic.eventually +import io.kotest.core.annotation.DisplayName +import io.kotest.core.spec.style.DescribeSpec +import org.rooftop.netx.api.TransactionManager +import org.springframework.test.context.ContextConfiguration +import org.springframework.test.context.TestPropertySource +import kotlin.time.Duration.Companion.minutes + +@ContextConfiguration( + classes = [ + RedisContainer::class, + RedisAssertions::class, + NoAckRedisTransactionConfigurer::class, + ] +) +@TestPropertySource("classpath:application.properties") +@DisplayName("RedisTransactionRetrySupporter 클래스의") +internal class RedisTransactionRetrySupporterTest( + private val redisAssertions: RedisAssertions, + private val transactionManager: TransactionManager, +) : DescribeSpec({ + + describe("handleOrphanTransaction 메소드는") { + context("pending되었지만, ack되지 않은 트랜잭션이 있다면,") { + it("해당 트랜잭션을 찾아서 처리하고, ack 상태로 변경한다.") { + val transactionId = transactionManager.start("undo").block()!! + + Thread.sleep(3_000) + + eventually(10.minutes) { + redisAssertions.pendingMessageCountShouldBe(transactionId, 0) + } + } + } + } +}) diff --git a/src/test/resources/application.properties b/src/test/resources/application.properties index 0c78093..fdea166 100644 --- a/src/test/resources/application.properties +++ b/src/test/resources/application.properties @@ -4,5 +4,5 @@ netx.port=6379 netx.group=netx-group netx.node-id=1 netx.node-name=netx-node -netx.recovery-milli=5000 +netx.recovery-milli=1000 netx.orphan-milli=1000 From 83d45eabfd7915c1ce1def173affe7a687c8f211 Mon Sep 17 00:00:00 2001 From: devxb Date: Sat, 10 Feb 2024 15:16:11 +0900 Subject: [PATCH 5/5] =?UTF-8?q?build:=20netx-idl=20=EB=B2=84=EC=A0=84=20?= =?UTF-8?q?=EC=B5=9C=EC=8B=A0=ED=99=94?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- idl | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/idl b/idl index ef759ce..fdb6042 160000 --- a/idl +++ b/idl @@ -1 +1 @@ -Subproject commit ef759ceb6d294dcf391b0563bb883db68bc787de +Subproject commit fdb6042d5406ff052d07ca2997cf8f74c0608eaf