Skip to content

Commit 6ab445d

Browse files
onoderisonoderis
committed
Add reconnection to youtube
1 parent 78f1998 commit 6ab445d

4 files changed

Lines changed: 71 additions & 70 deletions

File tree

‎src/main/kotlin/failchat/youtube/YoutubeChatClient.kt‎

Lines changed: 60 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -17,29 +17,31 @@ import failchat.chat.findTyped
1717
import failchat.chat.handlers.BraceEscaper
1818
import failchat.util.CoroutineExceptionLogger
1919
import failchat.util.value
20+
import failchat.youtube.LiveChatResponse.Action
2021
import kotlinx.coroutines.CoroutineScope
2122
import kotlinx.coroutines.Dispatchers
22-
import kotlinx.coroutines.Job
2323
import kotlinx.coroutines.cancel
24+
import kotlinx.coroutines.delay
25+
import kotlinx.coroutines.isActive
2426
import kotlinx.coroutines.launch
2527
import mu.KotlinLogging
2628
import java.util.concurrent.atomic.AtomicReference
29+
import kotlin.coroutines.cancellation.CancellationException
2730

2831
class YoutubeChatClient(
2932
override val callbacks: ChatClientCallbacks,
3033
private val youtubeClient: YoutubeClient,
3134
private val messageIdGenerator: MessageIdGenerator,
3235
private val history: ChatMessageHistory,
3336
private val videoId: String
34-
) : ChatClient, CoroutineScope by CoroutineScope(Dispatchers.Default) {
37+
) : ChatClient, CoroutineScope by CoroutineScope(Dispatchers.Default + CoroutineExceptionLogger) {
3538

3639
private companion object {
3740
val logger = KotlinLogging.logger {}
3841
val roleToBadgeMap = mapOf(
3942
RoleBadges.verified.description to RoleBadges.verified,
4043
"Owner" to RoleBadges.streamer,
4144
RoleBadges.moderator.description to RoleBadges.moderator
42-
4345
)
4446
val roleBadgeToColorMap = mapOf(
4547
RoleBadges.streamer to YoutubeColors.streamer,
@@ -65,57 +67,79 @@ class YoutubeChatClient(
6567
if (!statusChanged) {
6668
error("Chat client status: ${atomicStatus.value}")
6769
}
68-
logger.info { "Starting youtube client" }
70+
logger.info("Starting youtube client")
6971

70-
val job = launch {
71-
val initialParameters = youtubeClient.getNewLiveChatSessionData(videoId)
72-
logger.info { "Initial youtube parameters: $initialParameters" }
72+
launchWatcher()
73+
}
7374

74-
val statusUpdated = atomicStatus.compareAndSet(ChatClientStatus.CONNECTING, ChatClientStatus.CONNECTED)
75-
if (statusUpdated) {
76-
callbacks.onStatusUpdate(StatusUpdate(Origin.YOUTUBE, OriginStatus.CONNECTED))
75+
private fun launchWatcher() = launch {
76+
while (isActive) {
77+
try {
78+
listenForMessages()
79+
} catch (e: CancellationException) {
80+
// do nothing
81+
} catch (e: Throwable) {
82+
logger.error(e) { "Error occurred in youtube chat listener" }
83+
atomicStatus.set(ChatClientStatus.ERROR)
84+
callbacks.onStatusUpdate(StatusUpdate(Origin.YOUTUBE, OriginStatus.DISCONNECTED))
85+
delay(5000)
7786
}
87+
}
88+
atomicStatus.set(ChatClientStatus.OFFLINE)
89+
logger.info("Youtube watcher was stopped")
90+
}
7891

79-
highlightHandler.setChannelTitle(initialParameters.channelName)
92+
private suspend fun listenForMessages() {
93+
var parameters = youtubeClient.getNewLiveChatSessionData(videoId)
94+
logger.info { "Initial youtube parameters: $parameters" }
8095

81-
val actionChannel = with(youtubeClient) {
82-
pollLiveChatActions(initialParameters)
83-
}
84-
for (action in actionChannel) {
85-
if (action.isModerationAction()) {
86-
val channelIdToDeleteMessages = action.markChatItemsByAuthorAsDeletedAction!!.externalChannelId
96+
atomicStatus.set(ChatClientStatus.CONNECTED)
97+
callbacks.onStatusUpdate(StatusUpdate(Origin.YOUTUBE, OriginStatus.CONNECTED))
8798

88-
val messagesToDelete = history.findTyped<YoutubeMessage> {
89-
channelIdToDeleteMessages == it.author.id
90-
}
91-
messagesToDelete.forEach {
92-
callbacks.onChatMessageDeleted(it)
93-
}
99+
highlightHandler.setChannelTitle(parameters.channelName)
94100

101+
while (isActive) {
102+
val liveChatContinuation = youtubeClient.getLiveChatResponse(parameters)
103+
.continuationContents
104+
.liveChatContinuation
105+
106+
for (action in liveChatContinuation.actions) {
107+
if (action.isModerationAction()) {
108+
handleModerationAction(action)
95109
} else {
96-
val message = action.toChatMessage() ?: continue
97-
messageHandlers.forEach {
98-
it.handleMessage(message)
99-
}
100-
callbacks.onChatMessage(message)
110+
handleChatMessageAction(action)
101111
}
102112
}
113+
114+
val continuationData = liveChatContinuation.continuations.first().anyContinuation()
115+
parameters = parameters.copy(nextContinuation = continuationData.continuation)
116+
117+
delay(continuationData.timeoutMs.toLong())
103118
}
119+
}
104120

105-
job.invokeOnCompletion { e ->
106-
if (e != null) {
107-
atomicStatus.set(ChatClientStatus.ERROR)
108-
} else {
109-
atomicStatus.set(ChatClientStatus.OFFLINE)
110-
}
111-
callbacks.onStatusUpdate(StatusUpdate(Origin.YOUTUBE, OriginStatus.DISCONNECTED))
121+
private fun handleChatMessageAction(action: Action) {
122+
val message = action.toChatMessage() ?: return
123+
messageHandlers.forEach {
124+
it.handleMessage(message)
125+
}
126+
callbacks.onChatMessage(message)
127+
}
128+
129+
private suspend fun handleModerationAction(action: Action) {
130+
val channelIdToDeleteMessages = action.markChatItemsByAuthorAsDeletedAction!!.externalChannelId
131+
132+
val messagesToDelete = history.findTyped<YoutubeMessage> {
133+
channelIdToDeleteMessages == it.author.id
134+
}
135+
messagesToDelete.forEach {
136+
callbacks.onChatMessageDeleted(it)
112137
}
113138
}
114139

115140
override fun stop() {
116141
logger.info { "Stopping youtube client" }
117142
cancel()
118-
atomicStatus.value = ChatClientStatus.OFFLINE
119143
}
120144

121145
private fun LiveChatResponse.Action.toChatMessage(): YoutubeMessage? {
@@ -199,5 +223,4 @@ class YoutubeChatClient(
199223
description = liveChatAuthorBadgeRenderer.tooltip
200224
)
201225
}
202-
203226
}

‎src/main/kotlin/failchat/youtube/YoutubeClient.kt‎

Lines changed: 0 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -12,11 +12,6 @@ import io.ktor.http.HttpMethod
1212
import io.ktor.http.isSuccess
1313
import io.ktor.http.takeFrom
1414
import io.ktor.util.toByteArray
15-
import kotlinx.coroutines.CoroutineScope
16-
import kotlinx.coroutines.channels.Channel
17-
import kotlinx.coroutines.channels.ReceiveChannel
18-
import kotlinx.coroutines.delay
19-
import kotlinx.coroutines.launch
2015

2116
class YoutubeClient(
2217
private val httpClient: HttpClient,
@@ -132,28 +127,4 @@ class YoutubeClient(
132127
throw YoutubeClientException(cause = e)
133128
}
134129
}
135-
136-
fun CoroutineScope.pollLiveChatActions(initialParameters: LiveChatRequestParameters): ReceiveChannel<LiveChatResponse.Action> {
137-
val channel = Channel<LiveChatResponse.Action>(50)
138-
139-
launch {
140-
var parameters = initialParameters
141-
142-
while (true) {
143-
val response = getLiveChatResponse(parameters)
144-
145-
response.continuationContents.liveChatContinuation.actions.forEach { action ->
146-
channel.send(action)
147-
}
148-
149-
val continuationDto = response.continuationContents.liveChatContinuation.continuations.first().anyContinuation()
150-
parameters = parameters.copy(nextContinuation = continuationDto.continuation)
151-
152-
delay(continuationDto.timeoutMs.toLong())
153-
}
154-
}
155-
156-
return channel
157-
}
158-
159130
}

‎src/main/kotlin/failchat/youtube/YoutubeHighlightHandler.kt‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,11 @@
11
package failchat.youtube
22

33
import failchat.chat.MessageHandler
4-
import failchat.util.LateinitVal
4+
import java.util.concurrent.atomic.AtomicReference
55

66
class YoutubeHighlightHandler : MessageHandler<YoutubeMessage> {
77

8-
private val appealedChannelTitle = LateinitVal<String>()
8+
private val appealedChannelTitle = AtomicReference<String?>()
99

1010
override fun handleMessage(message: YoutubeMessage) {
1111
appealedChannelTitle.get()?.let {
@@ -18,5 +18,4 @@ class YoutubeHighlightHandler : MessageHandler<YoutubeMessage> {
1818
fun setChannelTitle(channelTitle: String) {
1919
appealedChannelTitle.set("@$channelTitle")
2020
}
21-
2221
}

‎src/test/kotlin/failchat/youtube/YoutubeClientTest.kt‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ import io.kotest.matchers.ints.shouldBeGreaterThanOrEqual
66
import kotlinx.coroutines.runBlocking
77
import org.junit.Ignore
88
import org.junit.Test
9+
import kotlin.test.assertTrue
910

1011
@Ignore
1112
class YoutubeClientTest {
@@ -15,10 +16,10 @@ class YoutubeClientTest {
1516
objectMapper = testObjectMapper,
1617
youtubeHtmlParser = YoutubeHtmlParser(objectMapper = testObjectMapper)
1718
)
19+
private val videoId = "jfKfPfyJRdk"
1820

1921
@Test
2022
fun getViewersCountTest() = runBlocking<Unit> {
21-
val videoId = "5qap5aO4i9A"
2223
val innertubeApiKey = client.getNewLiveChatSessionData(videoId).innertubeApiKey
2324

2425
val count = client.getViewersCount(videoId, innertubeApiKey)
@@ -27,4 +28,11 @@ class YoutubeClientTest {
2728
println(count)
2829
}
2930

31+
@Test
32+
fun getMessagesTest() = runBlocking<Unit> {
33+
val params = client.getNewLiveChatSessionData(videoId)
34+
val response = client.getLiveChatResponse(params)
35+
36+
assertTrue(response.continuationContents.liveChatContinuation.actions.isNotEmpty())
37+
}
3038
}

0 commit comments

Comments
 (0)