Avoid crashing when TLS TCP socket is closed (#3690)

* Avoid crashing when TLS TCP socket is closed

When using a plain TCP socket with TLS, two worker jobs are spawned:

- cio-tls-input-loop
- cio-tls-output-loop

These worker jobs live in their own scope, and exceptions they throw can't
easily be caught by the caller: the only option is to install a
`CoroutineExceptionHandler` for the whole TLS socket, which is quite hacky
and may hide bugs.

When the TCP socket is closed, this is only caught when trying to write
on the output channel, which throws a `ClosedSendChannelException`.
We now catch that exception and cleanly stop the background job.

Fixes https://youtrack.jetbrains.com/issue/KTOR-5178/TLSSocket-cannot-catch-the-exception-thrown-by-appDataOutputLoop
Fixes https://youtrack.jetbrains.com/issue/KTOR-4360/Android-Impossible-to-catch-the-ClosedSendChannelException-when-TLS-connection-socket-is-closed

* fixup! Avoid crashing when TLS TCP socket is closed

Author: t-bast

ktor-network/ktor-network-tls/jvm/src/io/ktor/network/tls/TLSClientSessionJvm.kt

@@ -48,7 +48,6 @@ private class TLSSocket(
             appDataOutputLoop(this.channel)
         }
 
-    @OptIn(ExperimentalCoroutinesApi::class)
     private suspend fun appDataInputLoop(pipe: ByteWriteChannel) {
         try {
             input.consumeEach { record ->
@@ -80,6 +79,8 @@ private class TLSSocket(
                 buffer.flip()
                 output.send(TLSRecord(TLSRecordType.ApplicationData, packet = buildPacket { writeFully(buffer) }))
             }
+        } catch (_: ClosedSendChannelException) {
+            // The socket was already closed, we should ignore that error.
         } finally {
             output.close()
         }

ktor-network/ktor-network-tls/jvm/test/io/ktor/network/tls/tests/ConnectionTest.kt

@@ -57,6 +57,20 @@ class ConnectionTest {
         Unit
     }
 
+    @Test
+    fun tlsWithCloseTest(): Unit = runBlocking {
+        val selectorManager = ActorSelectorManager(Dispatchers.IO)
+        val socket = aSocket(selectorManager)
+            .tcp()
+            .connect("www.google.com", port = 443)
+            .tls(Dispatchers.Default)
+
+        val channel = socket.openWriteChannel(autoFlush = true)
+        socket.close()
+        assertEquals(42, channel.writeAvailable(ByteArray(42)))
+        assertTrue(channel.isClosedForWrite)
+    }
+
     @Test
     @Ignore
     fun clientCertificatesAuthTest() {