wrap stream ids on reaching maximum value (#68)

Author: mostroverkhov

rsocket-core/src/main/kotlin/io/rsocket/kotlin/internal/RSocketRequester.kt

@@ -90,7 +90,7 @@ internal class RSocketRequester(
 
     private fun handleFireAndForget(payload: Payload): Completable {
         return Completable.fromRunnable {
-            val streamId = streamIds.nextStreamId()
+            val streamId = streamIds.nextStreamId(receivers)
             val requestFrame = Frame.Request.from(
                     streamId,
                     FrameType.FIRE_AND_FORGET,
@@ -102,7 +102,7 @@ internal class RSocketRequester(
 
     private fun handleRequestResponse(payload: Payload): Single<Payload> {
         return Single.defer {
-            val streamId = streamIds.nextStreamId()
+            val streamId = streamIds.nextStreamId(receivers)
             val requestFrame = Frame.Request.from(
                     streamId, FrameType.REQUEST_RESPONSE, payload, 1)
 
@@ -119,7 +119,7 @@ internal class RSocketRequester(
 
     private fun handleRequestStream(payload: Payload): Flowable<Payload> {
         return Flowable.defer {
-            val streamId = streamIds.nextStreamId()
+            val streamId = streamIds.nextStreamId(receivers)
             val receiver = StreamReceiver.create()
             receivers[streamId] = receiver
             val reqN = Cond()
@@ -148,7 +148,7 @@ internal class RSocketRequester(
     private fun handleChannel(request: Flowable<Payload>): Flowable<Payload> {
         return Flowable.defer {
             val receiver = StreamReceiver.create()
-            val streamId = streamIds.nextStreamId()
+            val streamId = streamIds.nextStreamId(receivers)
             val reqN = Cond()
 
             receiver.doOnRequestIfActive { requestN ->
@@ -259,7 +259,7 @@ internal class RSocketRequester(
                 }
                 else -> unsupportedFrame(streamId, frame)
             }
-        } ?: missingReceiver(streamId, type, frame)
+        }
     }
 
     private fun unsupportedFrame(streamId: Int, frame: Frame) {
@@ -268,21 +268,6 @@ internal class RSocketRequester(
                         "$streamId : $frame"))
     }
 
-    private fun missingReceiver(streamId: Int, type: FrameType, frame: Frame) {
-        if (!streamIds.isBeforeOrCurrent(streamId)) {
-            val err = if (type === FrameType.ERROR) {
-                IllegalStateException(
-                        "Client received error for non-existent stream: " +
-                                "$streamId Message: ${frame.dataUtf8}")
-            } else {
-                IllegalStateException(
-                        "Client received message for non-existent stream: " +
-                                "$streamId, frame type: $type")
-            }
-            errorConsumer(err)
-        }
-    }
-
     private inner class Lifecycle {
         private val terminated = AtomicReference<Throwable>()
 

rsocket-core/src/main/kotlin/io/rsocket/kotlin/internal/StreamIds.kt

@@ -16,19 +16,38 @@
 
 package io.rsocket.kotlin.internal
 
-internal sealed class StreamIds(private var streamId: Int) {
+import java.util.concurrent.atomic.AtomicLongFieldUpdater
 
-    @Synchronized
-    fun nextStreamId(): Int {
-        streamId += 2
+
+internal sealed class StreamIds(streamId: Int) {
+
+    @JvmField
+    @Volatile
+    internal var streamId: Long = streamId.toLong()
+
+    fun nextStreamId(streamIds: Map<Int, *>): Int {
+        var streamId: Int
+        do {
+            val next = STREAM_ID.addAndGet(this, 2)
+            if (next <= MAX_STREAM_ID) {
+                return next.toInt()
+            }
+            streamId = (next and MASK).toInt()
+        } while (streamId == 0 || streamIds.containsKey(streamId))
         return streamId
     }
 
-    @Synchronized
-    fun isBeforeOrCurrent(streamId: Int): Boolean =
-            this.streamId >= streamId && streamId > 0
+    companion object {
+        private val STREAM_ID = AtomicLongFieldUpdater.newUpdater(StreamIds::class.java, "streamId")
+        private const val MASK: Long = 0x7FFFFFFF
+        internal const val MAX_STREAM_ID = Int.MAX_VALUE
+
+
+    }
 }
 
 internal class ClientStreamIds : StreamIds(-1)
 
 internal class ServerStreamIds : StreamIds(0)
+
+internal class TestStreamIds(streamId: Int) : StreamIds(streamId)

rsocket-core/src/test/kotlin/io/rsocket/kotlin/RSocketRequesterTest.kt

@@ -42,21 +42,6 @@ class RSocketRequesterTest {
     @get:Rule
     val rule = ClientSocketRule()
 
-    @Test(timeout = 2000)
-    fun testInvalidFrameOnStream0() {
-
-        rule.receiver.onNext(Frame.RequestN.from(0, 10))
-        val errors = rule.errors
-
-        assertThat("Unexpected errors.",
-                errors,
-                hasSize<Throwable>(1))
-        assertThat(
-                "Unexpected error received.",
-                errors,
-                contains(instanceOf<Throwable>(IllegalStateException::class.java)))
-    }
-
     @Test(timeout = 2000)
     fun testStreamInitialN() {
         val stream = rule.requester.requestStream(DefaultPayload.EMPTY)

rsocket-core/src/test/kotlin/io/rsocket/kotlin/StreamIdsTest.kt

@@ -18,71 +18,62 @@ package io.rsocket.kotlin
 
 import io.rsocket.kotlin.internal.ClientStreamIds
 import io.rsocket.kotlin.internal.ServerStreamIds
+import io.rsocket.kotlin.internal.StreamIds
+import io.rsocket.kotlin.internal.TestStreamIds
 import org.junit.Assert.assertEquals
-import org.junit.Assert.assertFalse
-import org.junit.Assert.assertTrue
-
 import org.junit.Test
+import java.util.*
+import java.util.concurrent.ConcurrentHashMap
 
 class StreamIdsTest {
 
     @Test
     fun testClientSequence() {
+        val map = Collections.emptyMap<Int, Any>()
         val s = ClientStreamIds()
-        assertEquals(1, s.nextStreamId().toLong())
-        assertEquals(3, s.nextStreamId().toLong())
-        assertEquals(5, s.nextStreamId().toLong())
+        assertEquals(1, s.nextStreamId(map).toLong())
+        assertEquals(3, s.nextStreamId(map).toLong())
+        assertEquals(5, s.nextStreamId(map).toLong())
     }
 
     @Test
     fun testServerSequence() {
+        val map = Collections.emptyMap<Int, Any>()
         val s = ServerStreamIds()
-        assertEquals(2, s.nextStreamId().toLong())
-        assertEquals(4, s.nextStreamId().toLong())
-        assertEquals(6, s.nextStreamId().toLong())
+        assertEquals(2, s.nextStreamId(map).toLong())
+        assertEquals(4, s.nextStreamId(map).toLong())
+        assertEquals(6, s.nextStreamId(map).toLong())
     }
 
     @Test
-    fun testClientIsValid() {
-        val s = ClientStreamIds()
-
-        assertFalse(s.isBeforeOrCurrent(1))
-        assertFalse(s.isBeforeOrCurrent(3))
-
-        s.nextStreamId()
-        assertTrue(s.isBeforeOrCurrent(1))
-        assertFalse(s.isBeforeOrCurrent(3))
-
-        s.nextStreamId()
-        assertTrue(s.isBeforeOrCurrent(3))
+    fun testClientSequenceWrap() {
+        val map = ConcurrentHashMap<Int, Any>()
+        val s = TestStreamIds(Integer.MAX_VALUE - 2)
 
-        // negative
-        assertFalse(s.isBeforeOrCurrent(-1))
-        // connection
-        assertFalse(s.isBeforeOrCurrent(0))
-        // server also accepted (checked externally)
-        assertTrue(s.isBeforeOrCurrent(2))
+        assertEquals(2147483647, s.nextStreamId(map).toLong())
+        assertEquals(1, s.nextStreamId(map).toLong())
+        assertEquals(3, s.nextStreamId(map).toLong())
     }
 
     @Test
-    fun testServerIsValid() {
-        val s = ServerStreamIds()
-
-        assertFalse(s.isBeforeOrCurrent(2))
-        assertFalse(s.isBeforeOrCurrent(4))
-
-        s.nextStreamId()
-        assertTrue(s.isBeforeOrCurrent(2))
-        assertFalse(s.isBeforeOrCurrent(4))
+    fun testServerSequenceWrap() {
+        val map = ConcurrentHashMap<Int, Any>()
+        val s = TestStreamIds(Integer.MAX_VALUE - 3)
 
-        s.nextStreamId()
-        assertTrue(s.isBeforeOrCurrent(4))
+        assertEquals(2147483646, s.nextStreamId(map).toLong())
+        assertEquals(2, s.nextStreamId(map).toLong())
+        assertEquals(4, s.nextStreamId(map).toLong())
+    }
 
-        // negative
-        assertFalse(s.isBeforeOrCurrent(-2))
-        // connection
-        assertFalse(s.isBeforeOrCurrent(0))
-        // client also accepted (checked externally)
-        assertTrue(s.isBeforeOrCurrent(1))
+    @Test
+    fun testSequenceSkipsExistingStreamIds() {
+        val map = ConcurrentHashMap<Int, Any>()
+        map.put(5, Any())
+        map.put(9, Any())
+        val s = TestStreamIds(StreamIds.MAX_STREAM_ID)
+        assertEquals(1, s.nextStreamId(map).toLong())
+        assertEquals(3, s.nextStreamId(map).toLong())
+        assertEquals(7, s.nextStreamId(map).toLong())
+        assertEquals(11, s.nextStreamId(map).toLong())
     }
 }