Skip to content

Commit ed60b27

Browse files
SyrentSyrent
committed
feat: update RedisPublisher and RedisSubscriber for improved connection handling and message processing
1 parent 92cd059 commit ed60b27

3 files changed

Lines changed: 216 additions & 95 deletions

File tree

‎build.gradle.kts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@ plugins {
66

77
allprojects {
88
group = "org.sayandev"
9-
version = "1.8.9.96"
9+
version = "1.8.9.99"
1010
description = "A modular Kotlin framework for Minecraft: JE"
1111

1212
plugins.apply("maven-publish")

‎stickynote-core/src/main/kotlin/org/sayandev/stickynote/core/messaging/publisher/RedisPublisher.kt‎

Lines changed: 101 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -7,12 +7,13 @@ import org.sayandev.stickynote.core.coroutine.dispatcher.AsyncDispatcher
77
import org.sayandev.stickynote.core.messaging.publisher.PayloadWrapper.Companion.asJson
88
import org.sayandev.stickynote.core.messaging.publisher.PayloadWrapper.Companion.asPayloadWrapper
99
import org.sayandev.stickynote.core.messaging.publisher.PayloadWrapper.Companion.typedPayload
10-
import org.sayandev.stickynote.core.messaging.subscriber.Subscriber
11-
import org.sayandev.stickynote.core.messaging.subscriber.Subscriber.Companion
1210
import org.sayandev.stickynote.core.utils.CoroutineUtils.launch
1311
import redis.clients.jedis.JedisPool
1412
import redis.clients.jedis.JedisPubSub
13+
import redis.clients.jedis.exceptions.JedisException
1514
import java.util.*
15+
import java.util.concurrent.atomic.AtomicBoolean
16+
import java.util.logging.Level
1617
import java.util.logging.Logger
1718

1819
abstract class RedisPublisher<P, S>(
@@ -29,52 +30,104 @@ abstract class RedisPublisher<P, S>(
2930
name
3031
) {
3132
val channel = "$namespace:$name"
32-
33-
private val subJedis = redis.resource
34-
// private val pubJedis = redis.resource
33+
private var subJedis = redis.resource
34+
private var subscriberThread: Thread? = null
35+
private val isSubscribed = AtomicBoolean(false)
36+
private val shouldReconnect = AtomicBoolean(true)
37+
private val pubSub = createPubSub()
3538

3639
init {
37-
val pubSub = object : JedisPubSub() {
40+
startSubscriber()
41+
}
42+
43+
private fun createPubSub(): JedisPubSub {
44+
return object : JedisPubSub() {
3845
override fun onMessage(channel: String, message: String) {
3946
if (channel != this@RedisPublisher.channel) return
40-
val result = message.asPayloadWrapper<S>()
41-
when (result.state) {
42-
PayloadWrapper.State.FORWARD -> {
43-
val wrappedPayload = message.asPayloadWrapper<P>()
44-
if (wrappedPayload.excludeSource && isSource(wrappedPayload.uniqueId)) return
45-
val payloadResult = handle(wrappedPayload.typedPayload(payloadClass)) ?: return
46-
val localJedis = redis.resource
47-
try {
48-
localJedis.publish(
49-
channel.toByteArray(),
50-
PayloadWrapper(
51-
wrappedPayload.uniqueId,
52-
payloadResult,
53-
PayloadWrapper.State.RESPOND,
54-
wrappedPayload.source,
55-
wrappedPayload.target,
56-
wrappedPayload.excludeSource
57-
).asJson().toByteArray()
58-
)
59-
} finally {
60-
localJedis.close()
61-
}
62-
}
63-
PayloadWrapper.State.RESPOND -> {
64-
for (publisher in HANDLER_LIST.filterIsInstance<RedisPublisher<P, S>>()) {
65-
if (publisher.id() == channel) {
66-
publisher.payloads[result.uniqueId]?.apply {
67-
this.complete(result.typedPayload(resultClass))
68-
publisher.payloads.remove(result.uniqueId)
69-
}
70-
}
71-
}
47+
try {
48+
val result = message.asPayloadWrapper<S>()
49+
when (result.state) {
50+
PayloadWrapper.State.FORWARD -> handleForward(message)
51+
PayloadWrapper.State.RESPOND -> handleResponse(result)
52+
PayloadWrapper.State.PROXY -> {}
7253
}
73-
PayloadWrapper.State.PROXY -> {}
54+
} catch (e: Exception) {
55+
logger.log(Level.WARNING, "Error processing message: ${e.message}")
56+
}
57+
}
58+
}
59+
}
60+
61+
private fun handleForward(message: String) {
62+
val wrappedPayload = message.asPayloadWrapper<P>()
63+
if (wrappedPayload.excludeSource && isSource(wrappedPayload.uniqueId)) return
64+
val payloadResult = handle(wrappedPayload.typedPayload(payloadClass)) ?: return
65+
66+
val localJedis = redis.resource
67+
try {
68+
localJedis.publish(
69+
channel.toByteArray(),
70+
PayloadWrapper(
71+
wrappedPayload.uniqueId,
72+
payloadResult,
73+
PayloadWrapper.State.RESPOND,
74+
wrappedPayload.source,
75+
wrappedPayload.target,
76+
wrappedPayload.excludeSource
77+
).asJson().toByteArray()
78+
)
79+
} finally {
80+
localJedis.close()
81+
}
82+
}
83+
84+
private fun handleResponse(result: PayloadWrapper<S>) {
85+
for (publisher in HANDLER_LIST.filterIsInstance<RedisPublisher<P, S>>()) {
86+
if (publisher.id() == channel) {
87+
publisher.payloads[result.uniqueId]?.apply {
88+
this.complete(result.typedPayload(resultClass))
89+
publisher.payloads.remove(result.uniqueId)
7490
}
7591
}
7692
}
77-
Thread({ subJedis.subscribe(pubSub, channel) }, "redis-pub-sub-thread-${channel}-${UUID.randomUUID().toString().split("-").first()}").start()
93+
}
94+
95+
private fun startSubscriber() {
96+
if (!shouldReconnect.get() || isSubscribed.get()) return
97+
98+
synchronized(this) {
99+
if (isSubscribed.get()) return
100+
101+
subscriberThread?.interrupt()
102+
subscriberThread = Thread({
103+
while (shouldReconnect.get()) {
104+
try {
105+
subJedis = redis.resource
106+
isSubscribed.set(true)
107+
subJedis.subscribe(pubSub, channel)
108+
} catch (e: JedisException) {
109+
logger.log(Level.WARNING, "Redis connection lost: ${e.message}")
110+
isSubscribed.set(false)
111+
safeCloseJedis()
112+
Thread.sleep(5000) // Wait before reconnecting
113+
} catch (e: Exception) {
114+
logger.log(Level.SEVERE, "Unexpected error in subscriber: ${e.message}")
115+
isSubscribed.set(false)
116+
safeCloseJedis()
117+
Thread.sleep(5000)
118+
}
119+
}
120+
}, "redis-pub-sub-thread-${channel}-${UUID.randomUUID().toString().split("-").first()}")
121+
subscriberThread?.start()
122+
}
123+
}
124+
125+
private fun safeCloseJedis() {
126+
try {
127+
subJedis.close()
128+
} catch (e: Exception) {
129+
logger.log(Level.WARNING, "Error closing Jedis connection: ${e.message}")
130+
}
78131
}
79132

80133
override suspend fun publish(payload: PayloadWrapper<P>): CompletableDeferred<S> {
@@ -104,6 +157,13 @@ abstract class RedisPublisher<P, S>(
104157
return HANDLER_LIST.flatMap { publisher -> publisher.payloads.keys }.contains(uniqueId)
105158
}
106159

160+
fun shutdown() {
161+
shouldReconnect.set(false)
162+
pubSub.unsubscribe()
163+
safeCloseJedis()
164+
subscriberThread?.interrupt()
165+
}
166+
107167
companion object {
108168
const val TIMEOUT_SECONDS = 5L
109169

@@ -116,4 +176,4 @@ abstract class RedisPublisher<P, S>(
116176
}
117177
}
118178
}
119-
}
179+
}

‎stickynote-core/src/main/kotlin/org/sayandev/stickynote/core/messaging/subscriber/RedisSubscriber.kt‎

Lines changed: 114 additions & 53 deletions
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,10 @@ import org.sayandev.stickynote.core.utils.CoroutineUtils.launch
1313
import redis.clients.jedis.Jedis
1414
import redis.clients.jedis.JedisPool
1515
import redis.clients.jedis.JedisPubSub
16+
import redis.clients.jedis.exceptions.JedisException
1617
import java.util.*
18+
import java.util.concurrent.atomic.AtomicBoolean
19+
import java.util.logging.Level
1720
import java.util.logging.Logger
1821
import kotlin.coroutines.CoroutineContext
1922

@@ -28,78 +31,129 @@ abstract class RedisSubscriber<P, S>(
2831
) : Subscriber<P, S>(namespace, name) {
2932

3033
val channel = "$namespace:$name"
31-
val subJedis: Jedis = redis.resource
32-
// val pubJedis: Jedis = redis.resource
34+
private var subJedis: Jedis = redis.resource
35+
private var subscriberThread: Thread? = null
36+
private val isSubscribed = AtomicBoolean(false)
37+
private val shouldReconnect = AtomicBoolean(true)
38+
private val pubSub = createPubSub()
3339

3440
init {
35-
val pubSub = object : JedisPubSub() {
41+
startSubscriber()
42+
}
43+
44+
private fun createPubSub(): JedisPubSub {
45+
return object : JedisPubSub() {
3646
override fun onMessage(channel: String, message: String) {
3747
if (channel != this@RedisSubscriber.channel) return
38-
val payloadWrapper = message.asPayloadWrapper<P>()
39-
40-
when (payloadWrapper.state) {
41-
PayloadWrapper.State.PROXY -> {
42-
val isVelocity = runCatching { Class.forName("com.velocitypowered.api.proxy.ProxyServer") != null }.isSuccess
43-
if (!isVelocity) return
44-
45-
launch(dispatcher) {
46-
val result = (HANDLER_LIST.find {it.namespace == this@RedisSubscriber.namespace && it.name == this@RedisSubscriber.name } as Subscriber<P, S>)
47-
.onSubscribe(payloadWrapper.typedPayload(payloadClass))
48-
result.await()
49-
publish(
50-
PayloadWrapper(
51-
payloadWrapper.uniqueId,
52-
result.getCompleted(),
53-
PayloadWrapper.State.RESPOND,
54-
payloadWrapper.source
55-
)
56-
)
57-
}
58-
}
48+
try {
49+
val payloadWrapper = message.asPayloadWrapper<P>()
50+
handleMessage(payloadWrapper)
51+
} catch (e: Exception) {
52+
logger.log(Level.WARNING, "Error processing message: ${e.message}")
53+
}
54+
}
55+
}
56+
}
5957

60-
PayloadWrapper.State.FORWARD -> {
61-
if (payloadWrapper.excludeSource && isSource(payloadWrapper.uniqueId)) return
62-
launch(dispatcher) {
63-
val result =
64-
(HANDLER_LIST.find { it.namespace == this@RedisSubscriber.namespace && it.name == this@RedisSubscriber.name } as? Subscriber<P, S>)?.onSubscribe(
65-
payloadWrapper.typedPayload(payloadClass)
66-
)
67-
if (payloadWrapper.target == "PROCESSED") return@launch;
68-
publish(
69-
PayloadWrapper(
70-
payloadWrapper.uniqueId,
71-
result?.getCompleted() ?: payloadWrapper.payload,
72-
if (result != null) PayloadWrapper.State.RESPOND else payloadWrapper.state,
73-
payloadWrapper.source,
74-
"PROCESSED"
75-
)
76-
)
77-
}
78-
}
79-
PayloadWrapper.State.RESPOND -> {
80-
/*launch(dispatcher) {
81-
(HANDLER_LIST.find { publisher -> publisher.id() == channel } as? Subscriber<P, S>)
82-
?.onSubscribe(payloadWrapper.typedPayload(// TODO: Result class))
83-
}*/
58+
private fun handleMessage(payloadWrapper: PayloadWrapper<P>) {
59+
when (payloadWrapper.state) {
60+
PayloadWrapper.State.PROXY -> handleProxyMessage(payloadWrapper)
61+
PayloadWrapper.State.FORWARD -> handleForwardMessage(payloadWrapper)
62+
PayloadWrapper.State.RESPOND -> {} // Handle response if needed
63+
}
64+
}
65+
66+
private fun handleProxyMessage(payloadWrapper: PayloadWrapper<P>) {
67+
val isVelocity = runCatching { Class.forName("com.velocitypowered.api.proxy.ProxyServer") != null }.isSuccess
68+
if (!isVelocity) return
69+
70+
launch(dispatcher) {
71+
val result = (HANDLER_LIST.find { it.namespace == namespace && it.name == name } as Subscriber<P, S>)
72+
.onSubscribe(payloadWrapper.typedPayload(payloadClass))
73+
result.await()
74+
publish(
75+
PayloadWrapper(
76+
payloadWrapper.uniqueId,
77+
result.getCompleted(),
78+
PayloadWrapper.State.RESPOND,
79+
payloadWrapper.source
80+
)
81+
)
82+
}
83+
}
84+
85+
private fun handleForwardMessage(payloadWrapper: PayloadWrapper<P>) {
86+
if (payloadWrapper.excludeSource && isSource(payloadWrapper.uniqueId)) return
87+
launch(dispatcher) {
88+
val result = (HANDLER_LIST.find { it.namespace == namespace && it.name == name } as? Subscriber<P, S>)
89+
?.onSubscribe(payloadWrapper.typedPayload(payloadClass))
90+
if (payloadWrapper.target == "PROCESSED") return@launch
91+
publish(
92+
PayloadWrapper(
93+
payloadWrapper.uniqueId,
94+
result?.getCompleted() ?: payloadWrapper.payload,
95+
if (result != null) PayloadWrapper.State.RESPOND else payloadWrapper.state,
96+
payloadWrapper.source,
97+
"PROCESSED"
98+
)
99+
)
100+
}
101+
}
102+
103+
private fun startSubscriber() {
104+
if (!shouldReconnect.get() || isSubscribed.get()) return
105+
106+
synchronized(this) {
107+
if (isSubscribed.get()) return
108+
109+
subscriberThread?.interrupt()
110+
subscriberThread = Thread({
111+
while (shouldReconnect.get()) {
112+
try {
113+
subJedis = redis.resource
114+
isSubscribed.set(true)
115+
subJedis.subscribe(pubSub, channel)
116+
} catch (e: JedisException) {
117+
logger.log(Level.WARNING, "Redis connection lost: ${e.message}")
118+
isSubscribed.set(false)
119+
safeCloseJedis()
120+
Thread.sleep(5000)
121+
} catch (e: Exception) {
122+
logger.log(Level.SEVERE, "Unexpected error in subscriber: ${e.message}")
123+
isSubscribed.set(false)
124+
safeCloseJedis()
125+
Thread.sleep(5000)
84126
}
85127
}
86-
}
87-
};
88-
Thread({ subJedis.subscribe(pubSub, channel) }, "redis-sub-sub-thread-${channel}-${UUID.randomUUID().toString().split("-").first()}").start()
128+
}, "redis-sub-thread-${channel}-${UUID.randomUUID().toString().split("-").first()}")
129+
subscriberThread?.start()
130+
}
131+
}
132+
133+
private fun safeCloseJedis() {
134+
try {
135+
subJedis.close()
136+
} catch (e: Exception) {
137+
logger.log(Level.WARNING, "Error closing Jedis connection: ${e.message}")
138+
}
89139
}
90140

91141
private suspend fun publish(payload: PayloadWrapper<*>) {
92142
val publication = CompletableDeferred<Unit>()
93143
launch(dispatcher) {
94144
delay(TIMEOUT_SECONDS * 1000)
95145
if (publication.isActive) {
96-
publication.completeExceptionally(IllegalStateException("Failed to publish payload in subscriber after ${RedisPublisher.TIMEOUT_SECONDS} seconds. Payload: $payload (channel: ${id()})"))
146+
publication.completeExceptionally(IllegalStateException("Failed to publish payload in subscriber after $TIMEOUT_SECONDS seconds. Payload: $payload (channel: ${id()})"))
97147
}
98148
}
99149

100150
val localJedis = redis.resource
101151
try {
102152
localJedis.publish(channel.toByteArray(), payload.asJson().toByteArray())
153+
publication.complete(Unit)
154+
} catch (e: Exception) {
155+
logger.log(Level.WARNING, "Error publishing message: ${e.message}")
156+
publication.completeExceptionally(e)
103157
} finally {
104158
localJedis.close()
105159
}
@@ -109,7 +163,14 @@ abstract class RedisSubscriber<P, S>(
109163
return Publisher.HANDLER_LIST.flatMap { publisher -> publisher.payloads.keys }.contains(uniqueId)
110164
}
111165

166+
fun shutdown() {
167+
shouldReconnect.set(false)
168+
pubSub.unsubscribe()
169+
safeCloseJedis()
170+
subscriberThread?.interrupt()
171+
}
172+
112173
companion object {
113174
const val TIMEOUT_SECONDS = 5L
114175
}
115-
}
176+
}

0 commit comments

Comments
 (0)