Skip to content

Commit 8115e51

Browse files
martinbonninmartinbonnin
authored
Attempt at fixing a CI flake (#5961)
* Attempt at fixing a CI flake * Another attempt * remove debug * add comment
1 parent 8f526d8 commit 8115e51

File tree

3 files changed

+71
-77
lines changed

3 files changed

+71
-77
lines changed

‎gradle/libraries.toml

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,7 @@ kotlinx-coroutines = "1.8.0"
3232
kotlinx-datetime = "0.5.0"
3333
kotlinx-serialization-runtime = "1.6.2"
3434
ksp = "2.0.0-1.0.21"
35-
ktor = "2.3.7"
35+
ktor = "2.3.11"
3636
moshix = "0.14.1"
3737
node-fetch = "2.6.7"
3838
node-ws = "8.14.2"

‎libraries/apollo-engine-ktor/src/commonMain/kotlin/com/apollographql/apollo3/network/ws/KtorWebSocketEngine.kt

Lines changed: 63 additions & 71 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ import com.apollographql.apollo3.api.http.HttpHeader
55
import com.apollographql.apollo3.exception.ApolloNetworkException
66
import com.apollographql.apollo3.exception.ApolloWebSocketClosedException
77
import io.ktor.client.HttpClient
8+
import io.ktor.client.plugins.websocket.DefaultClientWebSocketSession
89
import io.ktor.client.plugins.websocket.WebSockets
910
import io.ktor.client.plugins.websocket.webSocket
1011
import io.ktor.client.request.headers
@@ -14,14 +15,14 @@ import io.ktor.http.URLProtocol
1415
import io.ktor.http.Url
1516
import io.ktor.websocket.CloseReason
1617
import io.ktor.websocket.Frame
17-
import io.ktor.websocket.close
1818
import io.ktor.websocket.readText
1919
import kotlinx.coroutines.CancellationException
2020
import kotlinx.coroutines.CoroutineScope
21-
import kotlinx.coroutines.Deferred
2221
import kotlinx.coroutines.Dispatchers
2322
import kotlinx.coroutines.SupervisorJob
2423
import kotlinx.coroutines.channels.Channel
24+
import kotlinx.coroutines.channels.ReceiveChannel
25+
import kotlinx.coroutines.coroutineScope
2526
import kotlinx.coroutines.launch
2627
import okio.ByteString
2728

@@ -37,6 +38,8 @@ class KtorWebSocketEngine(
3738
)
3839

3940
private val coroutineScope = CoroutineScope(Dispatchers.Default + SupervisorJob())
41+
private val receiveMessageChannel = Channel<String>(Channel.UNLIMITED)
42+
private val sendFrameChannel = Channel<Frame>(Channel.UNLIMITED)
4043

4144
override suspend fun open(
4245
url: String,
@@ -52,8 +55,6 @@ class KtorWebSocketEngine(
5255
/* URLProtocol.SOCKS */else -> throw UnsupportedOperationException("SOCKS is not a supported protocol")
5356
}
5457
}.build()
55-
val receiveMessageChannel = Channel<String>(Channel.UNLIMITED)
56-
val sendFrameChannel = Channel<Frame>(Channel.UNLIMITED)
5758
coroutineScope.launch {
5859
try {
5960
client.webSocket(
@@ -66,56 +67,36 @@ class KtorWebSocketEngine(
6667
url(newUrl)
6768
},
6869
) {
69-
launch {
70-
while (true) {
71-
val frame = sendFrameChannel.receive()
72-
try {
73-
send(frame)
74-
75-
// Also close the connection if the sent frame is a close frame
76-
if (frame is Frame.Close) {
77-
receiveMessageChannel.close()
78-
sendFrameChannel.close()
79-
break
80-
}
81-
} catch (e: Exception) {
82-
handleNetworkException(e, closeReason, receiveMessageChannel, sendFrameChannel)
83-
break
84-
}
70+
coroutineScope {
71+
launch {
72+
sendFrames(this@webSocket)
8573
}
86-
}
87-
while (true) {
88-
when (val frame = try {
89-
incoming.receive()
90-
} catch (e: Exception) {
91-
handleNetworkException(e, closeReason, receiveMessageChannel, sendFrameChannel)
92-
break
93-
}) {
94-
is Frame.Text -> {
95-
receiveMessageChannel.send(frame.readText())
96-
}
97-
98-
is Frame.Binary -> {
99-
receiveMessageChannel.send(frame.data.decodeToString())
74+
try {
75+
receiveFrames(incoming)
76+
} catch (e: Throwable) {
77+
val closeReason = closeReasonOrNull()
78+
val apolloException = if (closeReason != null) {
79+
ApolloWebSocketClosedException(
80+
code = closeReason.code.toInt(),
81+
reason = closeReason.message,
82+
cause = e
83+
)
84+
} else {
85+
ApolloNetworkException(
86+
message = "Web socket communication error",
87+
platformCause = e
88+
)
10089
}
101-
102-
is Frame.Ping -> {
103-
send(Frame.Pong(frame.data))
104-
}
105-
106-
is Frame.Pong -> {}
107-
is Frame.Close -> {
108-
close()
109-
receiveMessageChannel.close()
110-
}
111-
112-
else -> error("unknown frame type")
90+
receiveMessageChannel.close(apolloException)
91+
throw e
11392
}
11493
}
11594
}
116-
} catch (e: Exception) {
95+
} catch (e: Throwable) {
11796
receiveMessageChannel.close(ApolloNetworkException(message = "Web socket communication error", platformCause = e))
118-
sendFrameChannel.close(e)
97+
} finally {
98+
// Not 100% sure this can happen. Better safe than sorry. close() is idempotent so it shouldn't hurt
99+
receiveMessageChannel.close(ApolloNetworkException(message = "Web socket communication error", platformCause = null))
119100
}
120101
}
121102
return object : WebSocketConnection {
@@ -137,31 +118,42 @@ class KtorWebSocketEngine(
137118
}
138119
}
139120

140-
private suspend fun handleNetworkException(
141-
e: Exception,
142-
deferredCloseReason: Deferred<CloseReason?>,
143-
receiveMessageChannel: Channel<String>,
144-
sendFrameChannel: Channel<Frame>,
145-
) {
146-
if (e is CancellationException) throw e
147-
val closeReason = try {
148-
deferredCloseReason.await()
149-
} catch (e: Exception) {
121+
private suspend fun DefaultClientWebSocketSession.closeReasonOrNull(): CloseReason? {
122+
return try {
123+
closeReason.await()
124+
} catch (t: Throwable) {
125+
if (t is CancellationException) {
126+
throw t
127+
}
150128
null
151129
}
152-
val apolloException = if (closeReason != null) {
153-
ApolloWebSocketClosedException(
154-
code = closeReason.code.toInt(),
155-
reason = closeReason.message,
156-
cause = e
157-
)
158-
} else {
159-
ApolloNetworkException(
160-
message = "Web socket communication error",
161-
platformCause = e
162-
)
130+
}
131+
132+
private suspend fun sendFrames(session: DefaultClientWebSocketSession) {
133+
while (true) {
134+
val frame = sendFrameChannel.receive()
135+
session.send(frame)
136+
if (frame is Frame.Close) {
137+
// normal termination
138+
receiveMessageChannel.close()
139+
}
140+
}
141+
}
142+
143+
private suspend fun receiveFrames(incoming: ReceiveChannel<Frame>) {
144+
while (true) {
145+
val frame = incoming.receive()
146+
when (frame) {
147+
is Frame.Text -> {
148+
receiveMessageChannel.trySend(frame.readText())
149+
}
150+
151+
is Frame.Binary -> {
152+
receiveMessageChannel.trySend(frame.data.decodeToString())
153+
}
154+
155+
else -> error("unknown frame type")
156+
}
163157
}
164-
receiveMessageChannel.close(apolloException)
165-
sendFrameChannel.close(apolloException)
166158
}
167159
}

‎tests/engine/src/commonTest/kotlin/WebSocketEngineTest.kt

Lines changed: 7 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -3,15 +3,16 @@ import com.apollographql.apollo3.api.http.HttpHeader
33
import com.apollographql.apollo3.exception.ApolloException
44
import com.apollographql.apollo3.exception.ApolloNetworkException
55
import com.apollographql.apollo3.exception.ApolloWebSocketClosedException
6+
import com.apollographql.apollo3.mpp.Platform
7+
import com.apollographql.apollo3.mpp.platform
8+
import com.apollographql.apollo3.testing.internal.runTest
69
import com.apollographql.mockserver.CloseFrame
710
import com.apollographql.mockserver.DataMessage
811
import com.apollographql.mockserver.MockServer
912
import com.apollographql.mockserver.TextMessage
1013
import com.apollographql.mockserver.awaitWebSocketRequest
1114
import com.apollographql.mockserver.enqueueWebSocket
12-
import com.apollographql.apollo3.mpp.Platform
13-
import com.apollographql.apollo3.mpp.platform
14-
import com.apollographql.apollo3.testing.internal.runTest
15+
import kotlinx.coroutines.delay
1516
import okio.ByteString.Companion.encodeUtf8
1617
import kotlin.test.Test
1718
import kotlin.test.assertEquals
@@ -20,7 +21,6 @@ import kotlin.test.assertIs
2021
import kotlin.test.assertTrue
2122

2223
class WebSocketEngineTest {
23-
2424
/**
2525
* NSURLSession has a bug that sometimes skips the close frame
2626
* See https://developer.apple.com/forums/thread/679446
@@ -58,7 +58,6 @@ class WebSocketEngineTest {
5858
webSocketServer.close()
5959
}
6060

61-
6261
@Test
6362
fun binaryFrames() = runTest {
6463
if (platform() == Platform.Js) return@runTest // Binary frames are not supported by the JS WebSocketEngine
@@ -100,6 +99,9 @@ class WebSocketEngineTest {
10099
val responseBody = webSocketServer.enqueueWebSocket()
101100
val connection = webSocketEngine.open(webSocketServer.url())
102101

102+
// See https://youtrack.jetbrains.com/issue/KTOR-7099/Race-condition-in-darwin-websockets-client-runIncomingProcessor
103+
delay(1000)
104+
103105
responseBody.enqueueMessage(CloseFrame(4200, "Bye now"))
104106
val e = assertFailsWith<ApolloException> {
105107
connection.receive()

0 commit comments

Comments
 (0)