Skip to content

Commit d43ac79

Browse files
makeevrsergmakeevrserg
committed
[FIX] Fix okHttpClient discord
1 parent c8f58e2 commit d43ac79

6 files changed

Lines changed: 85 additions & 67 deletions

File tree

‎gradle.properties‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,7 @@ klibs.java.ktarget=21
1313
# Project
1414
klibs.project.name=MessageBridge
1515
klibs.project.group=ru.astrainteractive.messagebridge
16-
klibs.project.version.string=0.24.0
16+
klibs.project.version.string=0.24.1
1717
klibs.project.description=Bridge for TG and Discord
1818
klibs.project.developers=makeevrserg|Makeev Roman|makeevrserg@gmail.com
1919
klibs.project.url=https://empireprojekt.ru

‎instances/bukkit/src/main/kotlin/ru/astrainteractive/messagebridge/di/RootModule.kt‎

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,5 @@
11
package ru.astrainteractive.messagebridge.di
22

3-
import kotlinx.coroutines.GlobalScope
43
import kotlinx.coroutines.launch
54
import ru.astrainteractive.astralibs.command.api.brigadier.command.MultiplatformCommand
65
import ru.astrainteractive.astralibs.command.api.brigadier.command.PaperMultiplatformCommands
@@ -84,7 +83,7 @@ class RootModule(
8483

8584
val lifecycle = Lifecycle.Lambda(
8685
onEnable = {
87-
GlobalScope.launch(coreModule.dispatchers.IO) {
86+
coreModule.ioScope.launch {
8887
BEventChannel.consume(ServerOpenBEvent)
8988
}
9089
lifecycles.forEach(Lifecycle::onEnable)
@@ -93,10 +92,10 @@ class RootModule(
9392
lifecycles.forEach(Lifecycle::onReload)
9493
},
9594
onDisable = {
96-
GlobalScope.launch(coreModule.dispatchers.IO) {
95+
coreModule.ioScope.launch {
9796
BEventChannel.consume(ServerClosedBEvent)
9897
}
99-
lifecycles.forEach(Lifecycle::onDisable)
98+
lifecycles.reversed().forEach(Lifecycle::onDisable)
10099
}
101100
)
102101
}

‎instances/forge/src/main/kotlin/ru/astrainteractive/messagebridge/di/RootModule.kt‎

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,5 @@
11
package ru.astrainteractive.messagebridge.di
22

3-
import kotlinx.coroutines.GlobalScope
43
import kotlinx.coroutines.launch
54
import net.minecraftforge.fml.loading.FMLPaths
65
import ru.astrainteractive.astralibs.command.api.brigadier.command.MultiplatformCommand
@@ -90,7 +89,7 @@ class RootModule(
9089

9190
val lifecycle = Lifecycle.Lambda(
9291
onEnable = {
93-
GlobalScope.launch(coreModule.dispatchers.IO) {
92+
coreModule.ioScope.launch {
9493
BEventChannel.consume(ServerOpenBEvent)
9594
}
9695
lifecycles.forEach(Lifecycle::onEnable)
@@ -99,10 +98,10 @@ class RootModule(
9998
lifecycles.forEach(Lifecycle::onReload)
10099
},
101100
onDisable = {
102-
GlobalScope.launch(coreModule.dispatchers.IO) {
101+
coreModule.ioScope.launch {
103102
BEventChannel.consume(ServerClosedBEvent)
104103
}
105-
lifecycles.forEach(Lifecycle::onDisable)
104+
lifecycles.reversed().forEach(Lifecycle::onDisable)
106105
}
107106
)
108107
}

‎instances/neoforge/src/main/kotlin/ru/astrainteractive/messagebridge/di/RootModule.kt‎

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,5 @@
11
package ru.astrainteractive.messagebridge.di
22

3-
import kotlinx.coroutines.GlobalScope
43
import kotlinx.coroutines.launch
54
import net.neoforged.fml.loading.FMLPaths
65
import ru.astrainteractive.astralibs.command.api.brigadier.command.MultiplatformCommand
@@ -91,7 +90,7 @@ class RootModule(
9190

9291
val lifecycle = Lifecycle.Lambda(
9392
onEnable = {
94-
GlobalScope.launch(coreModule.dispatchers.IO) {
93+
coreModule.ioScope.launch {
9594
BEventChannel.consume(ServerOpenBEvent)
9695
}
9796
lifecycles.forEach(Lifecycle::onEnable)
@@ -100,10 +99,10 @@ class RootModule(
10099
lifecycles.forEach(Lifecycle::onReload)
101100
},
102101
onDisable = {
103-
GlobalScope.launch(coreModule.dispatchers.IO) {
102+
coreModule.ioScope.launch {
104103
BEventChannel.consume(ServerClosedBEvent)
105104
}
106-
lifecycles.forEach(Lifecycle::onDisable)
105+
lifecycles.reversed().forEach(Lifecycle::onDisable)
107106
}
108107
)
109108
}

‎modules/link/src/main/kotlin/ru/astrainteractive/messagebridge/link/database/di/LinkDatabaseModule.kt‎

Lines changed: 16 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -1,20 +1,20 @@
11
package ru.astrainteractive.messagebridge.link.database.di
22

33
import kotlinx.coroutines.CoroutineScope
4-
import kotlinx.coroutines.GlobalScope
54
import kotlinx.coroutines.flow.Flow
6-
import kotlinx.coroutines.flow.firstOrNull
5+
import kotlinx.coroutines.flow.SharingStarted
6+
import kotlinx.coroutines.flow.flatMapLatest
77
import kotlinx.coroutines.flow.flowOf
8-
import kotlinx.coroutines.launch
8+
import kotlinx.coroutines.flow.onEach
9+
import kotlinx.coroutines.flow.shareIn
910
import org.jetbrains.exposed.v1.jdbc.Database
1011
import org.jetbrains.exposed.v1.jdbc.SchemaUtils
1112
import org.jetbrains.exposed.v1.jdbc.transactions.TransactionManager
1213
import org.jetbrains.exposed.v1.jdbc.transactions.transaction
1314
import ru.astrainteractive.astralibs.lifecycle.Lifecycle
14-
import ru.astrainteractive.klibs.mikro.core.coroutines.mapCached
1515
import ru.astrainteractive.klibs.mikro.core.dispatchers.KotlinDispatchers
1616
import ru.astrainteractive.klibs.mikro.exposed.model.DatabaseConfiguration
17-
import ru.astrainteractive.klibs.mikro.exposed.util.connect
17+
import ru.astrainteractive.klibs.mikro.exposed.util.connectAsFlow
1818
import ru.astrainteractive.messagebridge.link.database.table.LinkedPlayerTable
1919
import java.io.File
2020

@@ -27,23 +27,17 @@ interface LinkDatabaseModule {
2727
dataFolder: File,
2828
dispatchers: KotlinDispatchers
2929
) : LinkDatabaseModule {
30-
override val databaseFlow: Flow<Database> = flowOf(
31-
value = DatabaseConfiguration.H2(dataFolder.resolve("linking").absolutePath)
32-
).mapCached(ioScope, dispatcher = dispatchers.IO) { dbConfig, previous ->
33-
previous?.connector?.invoke()?.close()
34-
previous?.run(TransactionManager::closeAndUnregister)
35-
val database = dbConfig.connect()
36-
TransactionManager.manager.defaultIsolationLevel = java.sql.Connection.TRANSACTION_SERIALIZABLE
37-
transaction(database) {
38-
SchemaUtils.create(LinkedPlayerTable)
39-
}
40-
database
41-
}
30+
override val databaseFlow: Flow<Database> =
31+
flowOf(DatabaseConfiguration.H2(dataFolder.resolve("linking").absolutePath))
32+
.flatMapLatest { databaseConfiguration -> databaseConfiguration.connectAsFlow() }
33+
.onEach { database ->
34+
TransactionManager.manager.defaultIsolationLevel = java.sql.Connection.TRANSACTION_SERIALIZABLE
35+
transaction(database) {
36+
SchemaUtils.create(LinkedPlayerTable)
37+
}
38+
}
39+
.shareIn(ioScope, SharingStarted.Eagerly, 1)
4240

43-
override val lifecycle: Lifecycle = Lifecycle.Lambda(
44-
onDisable = {
45-
GlobalScope.launch { databaseFlow.firstOrNull()?.run(TransactionManager::closeAndUnregister) }
46-
}
47-
)
41+
override val lifecycle: Lifecycle = Lifecycle.Lambda()
4842
}
4943
}

‎modules/messenger/discord/src/main/kotlin/ru/astrainteractive/messagebridge/messenger/discord/di/JdaMessengerModule.kt‎

Lines changed: 59 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -1,13 +1,14 @@
11
package ru.astrainteractive.messagebridge.messenger.discord.di
22

33
import com.neovisionaries.ws.client.WebSocketFactory
4-
import kotlinx.coroutines.GlobalScope
4+
import kotlinx.coroutines.CancellationException
55
import kotlinx.coroutines.cancel
66
import kotlinx.coroutines.channels.awaitClose
77
import kotlinx.coroutines.delay
88
import kotlinx.coroutines.flow.SharingStarted
99
import kotlinx.coroutines.flow.callbackFlow
1010
import kotlinx.coroutines.flow.combine
11+
import kotlinx.coroutines.flow.distinctUntilChanged
1112
import kotlinx.coroutines.flow.filterNotNull
1213
import kotlinx.coroutines.flow.first
1314
import kotlinx.coroutines.flow.firstOrNull
@@ -45,6 +46,7 @@ import ru.astrainteractive.messagebridge.messenger.discord.messaging.DiscordMess
4546
import ru.astrainteractive.messagebridge.messenger.discord.messaging.DiscordTopicUpdater
4647
import java.net.InetSocketAddress
4748
import java.net.Proxy
49+
import java.util.concurrent.TimeUnit
4850
import kotlin.time.Duration.Companion.seconds
4951

5052
class JdaMessengerModule(
@@ -53,57 +55,85 @@ class JdaMessengerModule(
5355
onlinePlayersProvider: OnlinePlayersProvider
5456
) : Logger by JUtiltLogger("MessageBridge-JdaMessengerModule").withoutParentHandlers() {
5557

56-
private val jdaFlow = coreModule.configKrate.cachedStateFlow
57-
.map { pluginConfiguration -> pluginConfiguration.jdaConfig }
58-
.flatMapLatest { config ->
58+
private val okHttpClientFlow = coreModule.configKrate.cachedStateFlow
59+
.map { pluginConfiguration -> pluginConfiguration.jdaConfig.proxy }
60+
.distinctUntilChanged()
61+
.flatMapLatest { proxy ->
62+
callbackFlow {
63+
val okHttpClient = if (proxy == null) {
64+
OkHttpClient.Builder().build()
65+
} else {
66+
@Suppress("MagicNumber")
67+
OkHttpClient.Builder()
68+
.connectTimeout(10, TimeUnit.SECONDS)
69+
.writeTimeout(10, TimeUnit.SECONDS)
70+
.readTimeout(60, TimeUnit.SECONDS)
71+
.callTimeout(75, TimeUnit.SECONDS)
72+
.pingInterval(15, TimeUnit.SECONDS)
73+
.retryOnConnectionFailure(true)
74+
.proxy(Proxy(Proxy.Type.HTTP, InetSocketAddress(proxy.host, proxy.port)))
75+
.proxyAuthenticator { route, response ->
76+
var builder = response.request.newBuilder()
77+
if (route?.socketAddress?.hostString == proxy.host) {
78+
val credential: String = Credentials.basic(proxy.username, proxy.password)
79+
builder.header("Proxy-Authorization", credential)
80+
}
81+
builder.build()
82+
}
83+
.build()
84+
}
85+
send(okHttpClient)
86+
87+
awaitClose {
88+
okHttpClient.dispatcher.executorService.shutdown()
89+
okHttpClient.connectionPool.evictAll()
90+
okHttpClient.cache?.close()
91+
}
92+
}
93+
}
94+
.shareIn(coreModule.ioScope, SharingStarted.Lazily, 1)
95+
96+
private val jdaFlow = combine(
97+
flow = okHttpClientFlow,
98+
flow2 = coreModule.configKrate.cachedStateFlow
99+
.map { pluginConfiguration -> pluginConfiguration.jdaConfig },
100+
transform = { okHttpClient, config ->
59101
callbackFlow {
60102
val builder = JDABuilder.createLight(config.token).apply {
61103
enableIntents(GatewayIntent.MESSAGE_CONTENT)
62104
enableIntents(GatewayIntent.DIRECT_MESSAGES)
63105
enableIntents(GatewayIntent.GUILD_MESSAGES)
64106
setActivity(Activity.playing(config.activity))
65107
config.proxy?.let { proxy ->
66-
val credential: String = Credentials.basic(proxy.username, proxy.password)
67108
setWebsocketFactory(
68109
WebSocketFactory()
69110
.setVerifyHostname(false)
70-
.also {
71-
it.proxySettings.setHost(proxy.host)
72-
it.proxySettings.setPort(proxy.port)
73-
it.proxySettings.setCredentials(proxy.username, proxy.password)
74-
}
75-
)
76-
setHttpClientBuilder(
77-
OkHttpClient.Builder()
78-
.proxy(Proxy(Proxy.Type.HTTP, InetSocketAddress(proxy.host, proxy.port)))
79-
.proxyAuthenticator { route, response ->
80-
var builder = response.request.newBuilder()
81-
if (route?.socketAddress?.hostString == proxy.host) {
82-
builder = builder.header("Proxy-Authorization", credential)
83-
}
84-
builder.build()
111+
.also { webSocketFactory ->
112+
webSocketFactory.proxySettings.setHost(proxy.host)
113+
webSocketFactory.proxySettings.setPort(proxy.port)
114+
webSocketFactory.proxySettings.setCredentials(proxy.username, proxy.password)
85115
}
86116
)
117+
setHttpClient(okHttpClient)
87118
}
88119
}
89120

90-
val jda = flow { emit(builder.build().awaitReady()) }
91-
.retryWhen { t, _ ->
92-
error { "#jdaFlow could not create JDA: ${t.localizedMessage}" }
93-
delay(5.seconds)
94-
true
95-
}
96-
.first()
97-
121+
val jda = builder.build().awaitReady()
98122
send(jda)
99123

100124
awaitClose {
101125
jda.shutdownNow()
102126
jda.awaitShutdown()
103127
jda.registeredListeners.forEach(jda::removeEventListener)
104128
}
129+
}.retryWhen { t, _ ->
130+
error { "#jdaFlow could not create JDA: ${t.localizedMessage}" }
131+
delay(5.seconds)
132+
val shouldRetry = t !is CancellationException
133+
shouldRetry
105134
}
106135
}
136+
).flattenConcat().shareIn(coreModule.ioScope, SharingStarted.Lazily, 1)
107137

108138
private val webhookClient = combine(
109139
flow = jdaFlow,
@@ -165,12 +195,9 @@ class JdaMessengerModule(
165195
onDisable = {
166196
discordMessageController.cancel()
167197
messageEventListener.cancel()
168-
GlobalScope.launch(coreModule.dispatchers.IO) {
198+
coreModule.ioScope.launch {
169199
jdaFlow.firstOrNull()?.let { jda ->
170200
messageEventListener.onDisable(jda)
171-
jda.registeredListeners.forEach(jda::removeEventListener)
172-
jda.shutdownNow()
173-
jda.awaitShutdown()
174201
}
175202
}
176203
}

0 commit comments

Comments
 (0)