Skip to content

fix: 자신이 발행한 transaction을 찾지 못하는 버그 수정 #32

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
devxb merged 2 commits into from
Feb 18, 2024
Merged

fix: 자신이 발행한 transaction을 찾지 못하는 버그 수정 #32

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e818366
자신이 발행한 transaction을 찾지 못하는 버그를 수정한다
devxb Feb 18, 2024
2fad59e
refactor: 사용하지 않는 메소드를 삭제한다
devxb Feb 18, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionDispatcher.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,22 +61,22 @@ abstract class AbstractTransactionDispatcher {
)
)

TransactionState.TRANSACTION_STATE_ROLLBACK -> findOwnTransaction(transaction)
TransactionState.TRANSACTION_STATE_ROLLBACK -> findOwnUndo(transaction)
.map {
TransactionRollbackEvent(
transaction.id,
transaction.serverId,
transaction.group,
transaction.cause,
it.undo,
it,
)
}

else -> throw cannotFindMatchedTransactionEventException
}
}

protected abstract fun findOwnTransaction(transaction: Transaction): Mono<Transaction>
protected abstract fun findOwnUndo(transaction: Transaction): Mono<String>

protected abstract fun ack(
transaction: Transaction,
Expand Down
20 changes: 7 additions & 13 deletions src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionDispatcher.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,8 +10,6 @@ import org.rooftop.netx.idl.Transaction
import org.rooftop.netx.idl.TransactionState
import org.rooftop.netx.meta.TransactionHandler
import org.springframework.context.ApplicationContext
import org.springframework.data.redis.connection.stream.ReadOffset
import org.springframework.data.redis.connection.stream.StreamOffset
import org.springframework.data.redis.core.ReactiveRedisTemplate
import reactor.core.publisher.Mono
import kotlin.reflect.KClass
Expand Down Expand Up @@ -64,19 +62,15 @@ class RedisStreamTransactionDispatcher(
}
}

override fun findOwnTransaction(transaction: Transaction): Mono<Transaction> {
return reactiveRedisTemplate.opsForStream<String, String>()
.read(StreamOffset.create(STREAM_KEY, ReadOffset.from("0")))
.map { Transaction.parseFrom(it.value["data"]?.toByteArray()) }
.filter { it.group == nodeGroup }
.filter { hasUndo(it) }
.next()
override fun findOwnUndo(transaction: Transaction): Mono<String> {
return reactiveRedisTemplate.opsForHash<String, String>()[transaction.id, nodeGroup]
.switchIfEmpty(
Mono.error {
error("Cannot find undo state in transaction hashes key \"${transaction.id}\"")
}
)
}

private fun hasUndo(transaction: Transaction): Boolean =
transaction.state == TransactionState.TRANSACTION_STATE_JOIN
|| transaction.state == TransactionState.TRANSACTION_STATE_START

override fun ack(transaction: Transaction, messageId: String): Mono<Pair<Transaction, String>> {
return reactiveRedisTemplate.opsForStream<String, ByteArray>()
.acknowledge(STREAM_KEY, nodeGroup, messageId)
Expand Down
44 changes: 23 additions & 21 deletions src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionManager.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,23 +1,17 @@
package org.rooftop.netx.redis

import org.redisson.api.RedissonReactiveClient
import org.rooftop.netx.engine.AbstractTransactionManager
import org.rooftop.netx.engine.AbstractTransactionRetrySupporter
import org.rooftop.netx.idl.Transaction
import org.rooftop.netx.idl.TransactionState
import org.springframework.data.redis.connection.stream.Record
import org.springframework.data.redis.core.ReactiveRedisTemplate
import reactor.core.publisher.Mono
import reactor.core.scheduler.Schedulers
import java.util.concurrent.TimeUnit

class RedisStreamTransactionManager(
nodeId: Int,
nodeName: String,
nodeGroup: String,
transactionRetrySupporter: AbstractTransactionRetrySupporter,
private val nodeGroup: String,
private val reactiveRedisTemplate: ReactiveRedisTemplate<String, ByteArray>,
private val redissonReactiveClient: RedissonReactiveClient,
) : AbstractTransactionManager(
nodeId = nodeId,
nodeName = nodeName,
Expand All @@ -26,13 +20,13 @@ class RedisStreamTransactionManager(

override fun findAnyTransaction(transactionId: String): Mono<TransactionState> {
return reactiveRedisTemplate
.opsForValue()[transactionId]
.opsForHash<String, String>()[transactionId, STATE_KEY]
.switchIfEmpty(
Mono.error {
error("Cannot find exists transaction id \"$transactionId\"")
}
)
.map { TransactionState.valueOf(String(it)) }
.map { TransactionState.valueOf(it) }
}

override fun publishTransaction(transactionId: String, transaction: Transaction): Mono<String> {
Expand All @@ -42,24 +36,32 @@ class RedisStreamTransactionManager(
.withStreamKey(STREAM_KEY)
)
.flatMap {
redissonReactiveClient.getLock("$transactionId-key")
.tryLock(10, TimeUnit.MINUTES)
}
.flatMap {
reactiveRedisTemplate.opsForValue()
.set(transactionId, transaction.state.name.toByteArray())
}
.doFinally {
redissonReactiveClient.getLock("$transactionId-key")
.forceUnlock()
.subscribeOn(Schedulers.parallel())
.subscribe()
if (hasUndo(transaction)) {
return@flatMap reactiveRedisTemplate.opsForHash<String, String>()
.putAll(
transactionId, mapOf(
STATE_KEY to transaction.state.name,
nodeGroup to transaction.undo
)
)
}
reactiveRedisTemplate.opsForHash<String, String>()
.putAll(
transactionId, mapOf(
STATE_KEY to transaction.state.name,
)
)
}
.map { transactionId }
}

private fun hasUndo(transaction: Transaction): Boolean =
transaction.state == TransactionState.TRANSACTION_STATE_JOIN
|| transaction.state == TransactionState.TRANSACTION_STATE_START

private companion object {
private const val DATA = "data"
private const val STREAM_KEY = "NETX_STREAM"
private const val STATE_KEY = "TX_STATE"
}
}
2 changes: 0 additions & 2 deletions src/main/kotlin/org/rooftop/netx/redis/RedisTransactionConfigurer.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,9 +34,7 @@ class RedisTransactionConfigurer(
nodeId = nodeId,
nodeName = nodeName,
nodeGroup = nodeGroup,
transactionRetrySupporter = redisTransactionRetrySupporter(),
reactiveRedisTemplate = reactiveRedisTemplate(),
redissonReactiveClient = redissonReactiveClient(),
)

@Bean
Expand Down
18 changes: 7 additions & 11 deletions src/test/kotlin/org/rooftop/netx/redis/NoAckRedisStreamTransactionDispatcher.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,19 +61,15 @@ class NoAckRedisStreamTransactionDispatcher(
}
}

override fun findOwnTransaction(transaction: Transaction): Mono<Transaction> {
return reactiveRedisTemplate.opsForStream<String, String>()
.read(StreamOffset.create(transaction.id, ReadOffset.from("0")))
.map { Transaction.parseFrom(it.value["data"]?.toByteArray()) }
.filter { it.group == nodeGroup }
.filter { hasUndo(it) }
.next()
override fun findOwnUndo(transaction: Transaction): Mono<String> {
return reactiveRedisTemplate.opsForHash<String, String>()[transaction.id, nodeGroup]
.switchIfEmpty(
Mono.error {
error("Cannot find undo state in transaction hashes key \"${transaction.id}\"")
}
)
}

private fun hasUndo(transaction: Transaction): Boolean =
transaction.state == TransactionState.TRANSACTION_STATE_JOIN
|| transaction.state == TransactionState.TRANSACTION_STATE_START

override fun ack(transaction: Transaction, messageId: String): Mono<Pair<Transaction, String>> =
Mono.just(transaction to messageId)

Expand Down
2 changes: 0 additions & 2 deletions src/test/kotlin/org/rooftop/netx/redis/NoAckRedisTransactionConfigurer.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,9 +34,7 @@ class NoAckRedisTransactionConfigurer(
nodeId = nodeId,
nodeName = nodeName,
nodeGroup = nodeGroup,
transactionRetrySupporter = redisTransactionRetrySupporter(),
reactiveRedisTemplate = reactiveRedisTemplate(),
redissonReactiveClient = redissonReactiveClient(),
)

@Bean
Expand Down