changed LimitingFlowCollector.requests to AtomicLong

This avoids Int overflow when client is misbehaving and is sending multiple RequestN frames with n=Int.MAX_VALUE

This closes #213

Author: yuriykulikov

rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/internal/Limiter.kt

@@ -32,27 +32,62 @@ internal suspend inline fun Flow<Payload>.collectLimiting(limiter: Limiter, cros
     }
 }
 
-//TODO revisit 2 atomics and sync object
+/**
+ * Maintains the amount of requests which the client is ready to consume and
+ * prevents sending further updates by suspending the sending coroutine
+ * if this amount reaches 0.
+ *
+ * ### Operation
+ *
+ * Each [useRequest] call decrements the maintained requests amount.
+ * Calling coroutine is suspended when this amount reaches 0.
+ * The coroutine is resumed when [updateRequests] is called.
+ *
+ * ### Unbounded mode
+ *
+ * Limiter enters an unbounded mode when:
+ * * [Limiter] is created passing `Int.MAX_VALUE` as `initial`
+ * * client sends a `RequestN` frame with `Int.MAX_VALUE`
+ * * Internal Long counter overflows
+ *
+ * In unbounded mode Limiter will assume that the client
+ * is able to process requests without limitations, all further
+ * [updateRequests] will be NOP and [useRequest] will never suspend.
+ */
 internal class Limiter(initial: Int) : SynchronizedObject() {
-    private val requests = atomic(initial)
-    private val awaiter = atomic<CancellableContinuation<Unit>?>(null)
+    private val requests: AtomicLong = atomic(initial.toLong())
+    private val unbounded: AtomicBoolean = atomic(initial == Int.MAX_VALUE)
+    private var awaiter: CancellableContinuation<Unit>? = null
 
     fun updateRequests(n: Int) {
-        if (n <= 0) return
+        if (n <= 0 || unbounded.value) return
         synchronized(this) {
-            requests += n
-            awaiter.getAndSet(null)?.takeIf(CancellableContinuation<Unit>::isActive)?.resume(Unit)
+            val updatedRequests = requests.value + n.toLong()
+            if (updatedRequests < 0) {
+                unbounded.value = true
+                requests.value = Long.MAX_VALUE
+            } else {
+                requests.value = updatedRequests
+            }
+
+            if (awaiter?.isActive == true) {
+                awaiter?.resume(Unit)
+                awaiter = null
+            }
         }
     }
 
     suspend fun useRequest() {
-        if (requests.getAndDecrement() > 0) {
+        if (unbounded.value || requests.decrementAndGet() >= 0) {
             currentCoroutineContext().ensureActive()
         } else {
-            suspendCancellableCoroutine<Unit> {
+            suspendCancellableCoroutine<Unit> { continuation ->
                 synchronized(this) {
-                    awaiter.value = it
-                    if (requests.value >= 0 && it.isActive) it.resume(Unit)
+                    if (requests.value >= 0 && continuation.isActive) {
+                        continuation.resume(Unit)
+                    } else {
+                        this.awaiter = continuation
+                    }
                 }
             }
         }

rsocket-core/src/commonTest/kotlin/io/rsocket/kotlin/core/RSocketTest.kt

@@ -23,6 +23,7 @@ import io.rsocket.kotlin.keepalive.*
 import io.rsocket.kotlin.payload.*
 import io.rsocket.kotlin.test.*
 import io.rsocket.kotlin.transport.local.*
+import kotlinx.atomicfu.atomic
 import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.*
 import kotlinx.coroutines.flow.*
@@ -192,6 +193,52 @@ class RSocketTest : SuspendTest, TestWithLeakCheck {
         assertTrue(channel.receiveCatching().isClosed)
     }
 
+    @Test
+    fun testStreamInitialUnbounded() = test {
+        val requester = start(RSocketRequestHandler {
+            requestStream {
+                (0..9).asFlow().map {
+                    payload(it.toString())
+                }
+            }
+        })
+        requester.requestStream(payload("HELLO"))
+                .flowOn(PrefetchStrategy(Int.MAX_VALUE, 0))
+                .test {
+                    repeat(10) {
+                        awaitItem().close()
+                    }
+                    awaitComplete()
+                }
+    }
+
+    @Test
+    fun testStreamRequestNUnbounded() = test {
+        class UnboundedAfterNStrategy(private val initial: Int) : RequestStrategy {
+            override fun provide(): RequestStrategy.Element = Element()
+            inner class Element : RequestStrategy.Element {
+                private val requested = atomic(initial)
+                override suspend fun firstRequest(): Int = initial
+                override suspend fun nextRequest(): Int {
+                    val requestUnbounded = requested.getAndDecrement() == 0
+                    return if (requestUnbounded) Int.MAX_VALUE else 0
+                }
+            }
+        }
+
+        start(RSocketRequestHandler {
+            requestStream {
+                (0..9).asFlow().map { payload(it.toString()) }
+            }
+        })
+                .requestStream(payload("HELLO"))
+                .flowOn(UnboundedAfterNStrategy(initial = 5))
+                .test {
+                    repeat(10) { awaitItem().close() }
+                    awaitComplete()
+                }
+    }
+
     @Test
     fun testChannel() = test {
         val awaiter = Job()