Merge branch 'release/0.9.8'

Author: mostroverkhov

README.md

@@ -13,6 +13,8 @@ It enables following symmetric interaction models:
 *  channel (bi-directional streams)
 *  per-stream and per-RSocket metadata 
 
+#### RPC style interactions support is provided by [RSOCKET RPC - KOTLIN](https://github.com/rsocket/rsocket-rpc-kotlin) 
+
 ## Build and Binaries
 
    ```groovy

build.gradle

@@ -41,9 +41,10 @@ subprojects {
         api 'io.netty:netty-buffer'
         api 'io.reactivex.rxjava2:rxjava'
         api 'org.jetbrains.kotlin:kotlin-stdlib-jdk7'
+        api 'org.slf4j:slf4j-api'
+
         compileOnly 'com.google.code.findbugs:jsr305'
 
-        testImplementation 'org.slf4j:slf4j-api'
         testImplementation 'junit:junit'
         testImplementation 'org.mockito:mockito-core'
         testImplementation 'org.hamcrest:hamcrest-library'

gradle.properties

@@ -15,7 +15,7 @@
 #
 
 group=io.rsocket.kotlin
-version=0.9.7
+version=0.9.8
 
 buildScanPluginVersion=2.4.1
 dependencyManagementPluginVersion=1.0.8.RELEASE

rsocket-core/gradle/dependency-locks/compileClasspath.lockfile

@@ -10,3 +10,4 @@ org.jetbrains.kotlin:kotlin-stdlib-jdk7:1.3.50
 org.jetbrains.kotlin:kotlin-stdlib:1.3.50
 org.jetbrains:annotations:13.0
 org.reactivestreams:reactive-streams:1.0.2
+org.slf4j:slf4j-api:1.7.28

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

@@ -23,6 +23,7 @@ import io.reactivex.processors.UnicastProcessor
 import io.rsocket.kotlin.*
 import io.rsocket.kotlin.exceptions.ApplicationException
 import io.rsocket.kotlin.exceptions.ChannelRequestException
+import io.rsocket.kotlin.internal.util.reactiveStreamsRequestN
 import org.reactivestreams.Publisher
 import org.reactivestreams.Subscriber
 import org.reactivestreams.Subscription
@@ -121,10 +122,11 @@ internal class RSocketRequester(
             val streamId = streamIds.nextStreamId(receivers)
             val receiver = StreamReceiver.create()
             receivers[streamId] = receiver
-            val reqN = Cond()
+            var isFirstRequestN = true
 
             receiver.doOnRequestIfActive { requestN ->
-                val frame = if (reqN.first()) {
+                val frame = if (isFirstRequestN) {
+                    isFirstRequestN = false
                     Frame.Request.from(
                             streamId,
                             FrameType.REQUEST_STREAM,
@@ -148,35 +150,37 @@ internal class RSocketRequester(
         return Flowable.defer {
             val receiver = StreamReceiver.create()
             val streamId = streamIds.nextStreamId(receivers)
-            val reqN = Cond()
+            var firstReqN = true
+            var firstReqPayload = true
 
             receiver.doOnRequestIfActive { requestN ->
 
-                if (reqN.first()) {
-                    val wrappedRequest = request.compose {
-                        val sender = RequestingPublisher.wrap(it)
-                        sender.request(1)
-                        senders[streamId] = sender
-                        receivers[streamId] = receiver
-                        sender
-                    }.publish().autoConnect(2)
+                if (firstReqN) {
+                    firstReqN = false
 
-                    val first = wrappedRequest.take(1)
-                            .map { payload ->
-                                Frame.Request.from(
-                                        streamId,
-                                        FrameType.REQUEST_CHANNEL,
-                                        payload,
-                                        requestN)
+                    val requestFrames = request
+                            .compose {
+                                val sender = RequestingPublisher.wrap(it)
+                                sender.request(1)
+                                senders[streamId] = sender
+                                receivers[streamId] = receiver
+                                sender
                             }
-                    val rest = wrappedRequest.skip(1)
                             .map { payload ->
-                                Frame.PayloadFrame.from(
-                                        streamId,
-                                        FrameType.NEXT,
-                                        payload)
+                                if (firstReqPayload) {
+                                    firstReqPayload = false
+                                    Frame.Request.from(
+                                            streamId,
+                                            FrameType.REQUEST_CHANNEL,
+                                            payload,
+                                            requestN)
+                                } else {
+                                    Frame.PayloadFrame.from(
+                                            streamId,
+                                            FrameType.NEXT,
+                                            payload)
+                                }
                             }
-                    val requestFrames = Flowable.concatArrayEager(first, rest)
                     requestFrames.subscribe(
                             ChannelRequestSubscriber(
                                     { payload -> frameSender.send(payload) },
@@ -248,10 +252,7 @@ internal class RSocketRequester(
                 FrameType.NEXT -> receiver.onNext(DefaultPayload(frame))
                 FrameType.REQUEST_N -> {
                     val sender = senders[streamId]
-                    sender?.let {
-                        val n = Frame.RequestN.requestN(frame).toLong()
-                        it.request(n)
-                    }
+                    sender?.request(reactiveStreamsRequestN(Frame.RequestN.requestN(frame)))
                 }
                 FrameType.COMPLETE -> {
                     receiver.onComplete()
@@ -320,18 +321,6 @@ internal class RSocketRequester(
         }
     }
 
-    private class Cond {
-        private var first = true
-
-        fun first(): Boolean =
-                if (first) {
-                    first = false
-                    true
-                } else {
-                    false
-                }
-    }
-
     private class ChannelRequestSubscriber(private val next: (Frame) -> Unit,
                                            private val error: (Throwable) -> Unit,
                                            private val complete: (Boolean) -> Unit)