reduce allocations in okhttp transport (#69)

Author: mostroverkhov

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

@@ -22,18 +22,6 @@ import io.rsocket.kotlin.internal.frame.ErrorFrameFlyweight
 
 internal object Exceptions {
 
-    /**
-     * Creates [Throwable] with no stack trace
-     */
-    fun <T : Throwable> noStacktrace(ex: T): T {
-        ex.stackTrace = arrayOf(StackTraceElement(
-                ex.javaClass.name,
-                "<init>",
-                null,
-                -1))
-        return ex
-    }
-
     /**
      * Creates [RuntimeException] from given Error [Frame]
      */

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

@@ -23,7 +23,6 @@ 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.Exceptions.noStacktrace
 import org.reactivestreams.Publisher
 import org.reactivestreams.Subscriber
 import org.reactivestreams.Subscription
@@ -358,6 +357,6 @@ internal class RSocketRequester(
     }
 
     companion object {
-        private val closedException = noStacktrace(ClosedChannelException())
+        private val closedException = ClosedChannelException()
     }
 }

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

@@ -26,7 +26,6 @@ import io.rsocket.kotlin.internal.RSocketResponder.DisposableSubscription.Compan
 import io.rsocket.kotlin.exceptions.ApplicationException
 import io.rsocket.kotlin.internal.frame.FrameHeaderFlyweight.FLAGS_C
 import io.rsocket.kotlin.internal.frame.FrameHeaderFlyweight.FLAGS_M
-import io.rsocket.kotlin.internal.Exceptions.noStacktrace
 import io.rsocket.kotlin.DefaultPayload
 import org.reactivestreams.Publisher
 import org.reactivestreams.Subscriber
@@ -298,6 +297,6 @@ internal class RSocketResponder(
     }
 
     companion object {
-        private val closedException = noStacktrace(ClosedChannelException())
+        private val closedException = ClosedChannelException()
     }
 }

rsocket-core/src/test/kotlin/io/rsocket/kotlin/util/ExceptionsTest.kt

@@ -16,8 +16,7 @@
 
 package io.rsocket.transport.okhttp
 
-import io.netty.buffer.ByteBuf
-import io.netty.buffer.Unpooled
+import io.netty.buffer.ByteBufAllocator
 import io.reactivex.Completable
 import io.reactivex.Flowable
 import io.reactivex.Single
@@ -35,9 +34,7 @@ internal class OkWebsocket(client: OkHttpClient,
     private val isOpen = AtomicBoolean()
     @Volatile
     private var failErr: ClosedChannelException? = null
-    private val defFailErr by lazy {
-        noStacktrace(ClosedChannelException())
-    }
+    private val defFailErr by lazy { ClosedChannelException() }
     private val connection = BehaviorProcessor.create<OkHttpWebSocketConnection>()
     private val frames = UnicastProcessor.create<Frame>()
 
@@ -48,23 +45,15 @@ internal class OkWebsocket(client: OkHttpClient,
         }
 
         override fun onMessage(webSocket: WebSocket?, bytes: ByteString) {
-            val msgBuffer = Unpooled.wrappedBuffer(bytes.asByteBuffer())
-            val frameBuffer = writeFrame(msgBuffer)
-
-            frames.onNext(Frame.from(frameBuffer))
-        }
+            val messageSize = bytes.size()
+            val buffer = allocator.buffer(messageSize + frameLengthSize)
 
-        private fun writeFrame(msgBuffer: ByteBuf): ByteBuf {
-            val msgSize = msgBuffer.readableBytes()
-            val frameSize = msgSize + frameLengthSize
-            val frameBuffer = Unpooled.buffer(frameSize, frameSize)
+            buffer.writeByte(messageSize shr 16)
+            buffer.writeByte(messageSize shr 8)
+            buffer.writeByte(messageSize)
+            buffer.writeBytes(bytes.toByteArray())
 
-            frameBuffer.writeByte(msgSize shr 16)
-            frameBuffer.writeByte(msgSize shr 8)
-            frameBuffer.writeByte(msgSize)
-            frameBuffer.writeBytes(msgBuffer)
-
-            return frameBuffer
+            frames.onNext(Frame.from(buffer))
         }
 
         override fun onClosed(webSocket: WebSocket?,
@@ -98,14 +87,12 @@ internal class OkWebsocket(client: OkHttpClient,
 
     fun send(frames: Publisher<Frame>): Completable =
             Flowable.fromPublisher(frames)
-                    .map { it.content() }
-                    .map { byteBuf ->
-                        val byteString = ByteString.of(
-                                byteBuf.skipBytes(frameLengthSize).nioBuffer())
-                        byteBuf.release()
-                        byteString
-                    }
-                    .flatMapCompletable { ws.sendAsync(it) }
+                    .map { frame ->
+                        val content = frame.content()
+                        val contentByteString = ByteString.of(content.skipBytes(frameLengthSize).nioBuffer())
+                        frame.release()
+                        ws.sendOrThrowOnFailure(contentByteString)
+                    }.ignoreElements()
 
     fun close(): Completable = Completable.create { e ->
         ws.close(normalClose, "close")
@@ -118,27 +105,16 @@ internal class OkWebsocket(client: OkHttpClient,
 
     internal fun isOpen(): Boolean = isOpen.get()
 
-    private fun WebSocket.sendAsync(bytes: ByteString): Completable =
-            Completable.create { e ->
-                if (send(bytes))
-                    e.onComplete()
-                else
-                    e.onError(failErr ?: defFailErr)
-            }
+    private fun WebSocket.sendOrThrowOnFailure(bytes: ByteString) {
+        if (!send(bytes)) {
+            throw (failErr ?: defFailErr)
+        }
+    }
 
     companion object {
         private const val normalClose = 1000
         private const val frameLengthSize = 3
-
-        private fun <T : Throwable> noStacktrace(ex: T): T {
-            ex.stackTrace = arrayOf(StackTraceElement(
-                    ex.javaClass.name,
-                    "<init>",
-                    null,
-                    -1))
-            return ex
-        }
-
+        private val allocator = ByteBufAllocator.DEFAULT
     }
 }
 

rsocket-transport-okhttp/src/main/kotlin/io/rsocket/transport/okhttp/OkWebsocket.kt

@@ -0,0 +1,33 @@
+#
+# Copyright 2015-2018 the original author or authors.
+# <p>
+#  Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+#  the License. You may obtain a copy of the License at
+#  <p>
+#  http://www.apache.org/licenses/LICENSE-2.0
+#  <p>
+#  Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+#  an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+#  specific language governing permissions and limitations under the License.
+#
+
+
+#
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+log4j.rootLogger=INFO, stdout
+
+log4j.appender.stdout=org.apache.log4j.ConsoleAppender
+log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
+log4j.appender.stdout.layout.ConversionPattern=%d{dd MMM yyyy HH:mm:ss,SSS} %5p [%t] (%F:%L) - %m%n