Drop isClosed

Author: whyoleg

rsocket-core/api/rsocket-core.api

@@ -763,14 +763,12 @@ public abstract interface class io/rsocket/kotlin/transport/RSocketMultiplexedCo
 }
 
 public abstract interface class io/rsocket/kotlin/transport/RSocketMultiplexedConnection$Stream : kotlinx/coroutines/CoroutineScope {
-	public abstract fun isClosedForSend ()Z
 	public abstract fun receiveFrame (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public abstract fun sendFrame (Lkotlinx/io/Buffer;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public abstract fun setSendPriority (I)V
 }
 
 public abstract interface class io/rsocket/kotlin/transport/RSocketSequentialConnection : io/rsocket/kotlin/transport/RSocketConnection {
-	public abstract fun isClosedForSend ()Z
 	public abstract fun receiveFrame (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public abstract fun sendFrame (ILkotlinx/io/Buffer;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 }

rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/connection/ConnectionEstablishmentHandler.kt

@@ -89,7 +89,7 @@ internal abstract class ConnectionEstablishmentHandler(
             }
         } catch (cause: Throwable) {
             connection.close()
-            withContext(NonCancellable) {
+            nonCancellable {
                 connection.sendError(
                     when (cause) {
                         is RSocketError -> cause

rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/connection/LoggingConnection.kt

@@ -40,7 +40,6 @@ private class SequentialLoggingConnection(
     private val delegate: RSocketSequentialConnection,
     private val logger: Logger,
 ) : RSocketSequentialConnection {
-    override val isClosedForSend: Boolean get() = delegate.isClosedForSend
     override val coroutineContext: CoroutineContext get() = delegate.coroutineContext
 
     override suspend fun sendFrame(streamId: Int, frame: Buffer) {
@@ -86,7 +85,6 @@ private class MultiplexedLoggingStream(
     private val delegate: RSocketMultiplexedConnection.Stream,
     private val logger: Logger,
 ) : RSocketMultiplexedConnection.Stream {
-    override val isClosedForSend: Boolean get() = delegate.isClosedForSend
     override val coroutineContext: CoroutineContext get() = delegate.coroutineContext
 
     override fun setSendPriority(priority: Int) {

rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/connection/MultiplexedConnection.kt

@@ -226,7 +226,6 @@ internal class MultiplexedConnection(
         streamId: Int,
         private val stream: RSocketMultiplexedConnection.Stream,
     ) : OperationOutbound(streamId, frameCodec) {
-        override val isClosed: Boolean get() = stream.isClosedForSend
         override suspend fun sendFrame(frame: Buffer): Unit = stream.sendFrame(frame)
     }
 

rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/connection/OldConnection.kt

@@ -32,8 +32,6 @@ internal class OldConnection(
 ) : RSocketSequentialConnection {
     private val outboundQueue = PrioritizationFrameQueue(Channel.BUFFERED)
 
-    override val isClosedForSend: Boolean get() = outboundQueue.isClosedForSend
-
     override val coroutineContext: CoroutineContext get() = connection.coroutineContext
 
     init {

rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/connection/SequentialConnection.kt

@@ -144,7 +144,6 @@ internal class SequentialConnection(
     }
 
     private inner class Outbound(streamId: Int) : OperationOutbound(streamId, frameCodec) {
-        override val isClosed: Boolean get() = !isActive || connection.isClosedForSend
         override suspend fun sendFrame(frame: Buffer): Unit = connection.sendFrame(streamId, frame)
     }
 

rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/operation/OperationOutbound.kt

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2024 the original author or authors.
+ * Copyright 2015-2025 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -34,8 +34,6 @@ internal abstract class OperationOutbound(
     // TODO: decide on it
     // private var firstRequestFrameSent: Boolean = false
 
-    abstract val isClosed: Boolean
-
     protected abstract suspend fun sendFrame(frame: Buffer)
     private suspend fun sendFrame(frame: Frame): Unit = sendFrame(frameCodec.encodeFrame(frame))
 

rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/operation/RequesterFireAndForgetOperation.kt

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2024 the original author or authors.
+ * Copyright 2015-2025 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -17,6 +17,7 @@
 package io.rsocket.kotlin.operation
 
 import io.rsocket.kotlin.frame.*
+import io.rsocket.kotlin.internal.io.*
 import io.rsocket.kotlin.payload.*
 import kotlinx.coroutines.*
 import kotlin.coroutines.*
@@ -36,7 +37,7 @@ internal class RequesterFireAndForgetOperation(
             requestSentCont.resume(Unit)
         } catch (cause: Throwable) {
             if (requestSentCont.isActive) requestSentCont.resumeWithException(cause)
-            if (!outbound.isClosed) withContext(NonCancellable) { outbound.sendCancel() }
+            nonCancellable { outbound.sendCancel() }
             throw cause
         }
     }

rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/operation/RequesterRequestChannelOperation.kt

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2024 the original author or authors.
+ * Copyright 2015-2025 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -18,6 +18,7 @@ package io.rsocket.kotlin.operation
 
 import io.rsocket.kotlin.frame.*
 import io.rsocket.kotlin.internal.*
+import io.rsocket.kotlin.internal.io.*
 import io.rsocket.kotlin.payload.*
 import kotlinx.atomicfu.*
 import kotlinx.coroutines.*
@@ -58,11 +59,11 @@ internal class RequesterRequestChannelOperation(
                 try {
                     while (true) outbound.sendRequestN(responsePayloads.nextRequestN() ?: break)
                 } catch (cause: Throwable) {
-                    if (!currentCoroutineContext().isActive || !outbound.isClosed) throw cause
+                    if (!currentCoroutineContext().isActive) throw cause
                 }
             }
         } catch (cause: Throwable) {
-            if (!outbound.isClosed) withContext(NonCancellable) {
+            nonCancellable {
                 when (val error = failure) {
                     null -> outbound.sendCancel()
                     else -> outbound.sendError(error)

rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/operation/RequesterRequestResponseOperation.kt

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2024 the original author or authors.
+ * Copyright 2015-2025 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -17,6 +17,7 @@
 package io.rsocket.kotlin.operation
 
 import io.rsocket.kotlin.frame.*
+import io.rsocket.kotlin.internal.io.*
 import io.rsocket.kotlin.payload.*
 import kotlinx.coroutines.*
 
@@ -35,7 +36,7 @@ internal class RequesterRequestResponseOperation(
             responseDeferred.join()
         } catch (cause: Throwable) {
             // TODO: we don't need to send cancel if we have sent no frames
-            if (!outbound.isClosed) withContext(NonCancellable) { outbound.sendCancel() }
+            nonCancellable { outbound.sendCancel() }
             throw cause
         }
     }