Improve Lease support with sources for sent and received Leases (#37)

* Introduce FrameSender which releases queued frames on cancel/termination

basic interactions stress tests

cleanup readme

* remove commented code

* shorten method name & update readme

* Improve Lease support with sources for sent and received Leases

* complete sent and received Lease streams on connection close

* Lease: add initialAllowedRequests property

* Simplified RSocketFactory api, added javadocs

Author: mostroverkhov

rsocket-core/src/main/kotlin/io/rsocket/kotlin/Availability.kt

@@ -16,8 +16,8 @@ package io.rsocket.kotlin
 interface Availability {
 
     /**
-     * @return a positive numbers representing the availability of the entity. Higher is better, 0.0
-     * means not available
+     * @return a positive numbers representing the availability of the entity.
+     * Higher is better, 0.0 means not available
      */
     fun availability(): Double
 }

rsocket-core/src/main/kotlin/io/rsocket/kotlin/ClientOptions.kt

@@ -1,7 +1,23 @@
 package io.rsocket.kotlin
 
-class ClientOptions : Options<ClientOptions>() {
+class ClientOptions {
 
-    override fun copy(): ClientOptions =
-            ClientOptions().streamRequestLimit(streamRequestLimit())
+    private var streamRequestLimit: Int = 128
+
+    fun streamRequestLimit(streamRequestLimit: Int): ClientOptions {
+        assertRequestLimit(streamRequestLimit)
+        this.streamRequestLimit = streamRequestLimit
+        return this
+    }
+
+    internal fun streamRequestLimit(): Int = streamRequestLimit
+
+    fun copy(): ClientOptions = ClientOptions()
+            .streamRequestLimit(streamRequestLimit)
+
+    private fun assertRequestLimit(streamRequestLimit: Int) {
+        if (streamRequestLimit <= 0) {
+            throw IllegalArgumentException("stream request limit must be positive")
+        }
+    }
 }

rsocket-core/src/main/kotlin/io/rsocket/kotlin/DefaultPayload.kt

@@ -81,7 +81,7 @@ class DefaultPayload @JvmOverloads constructor(data: ByteBuffer,
             return _metadata
         }
 
-    override fun hasMetadata(): Boolean = _metadata != null
+    override val hasMetadata: Boolean = _metadata != null
 
     companion object {
 

rsocket-core/src/main/kotlin/io/rsocket/kotlin/Duration.kt

@@ -2,6 +2,9 @@ package io.rsocket.kotlin
 
 import java.util.concurrent.TimeUnit
 
+/**
+ * Represents duration with different time units
+ */
 data class Duration(private val value: Long, val unit: TimeUnit) {
 
     val millis = unit.toMillis(value)

rsocket-core/src/main/kotlin/io/rsocket/kotlin/Frame.kt

@@ -236,7 +236,7 @@ class Frame private constructor(private val handle: Handle<Frame>) : ByteBufHold
                 metadataMimeType: String,
                 dataMimeType: String,
                 payload: Payload): Frame {
-            val metadata = if (payload.hasMetadata())
+            val metadata = if (payload.hasMetadata)
                 Unpooled.wrappedBuffer(payload.metadata)
             else
                 Unpooled.EMPTY_BUFFER
@@ -424,7 +424,7 @@ class Frame private constructor(private val handle: Handle<Frame>) : ByteBufHold
             if (initialRequestN < 1) {
                 throw IllegalStateException("initial request n must be greater than 0")
             }
-            val metadata = if (payload.hasMetadata()) Unpooled.wrappedBuffer(payload.metadata) else null
+            val metadata = if (payload.hasMetadata) Unpooled.wrappedBuffer(payload.metadata) else null
             val data = Unpooled.wrappedBuffer(payload.data)
 
             val frame = RECYCLER.get()
@@ -531,8 +531,8 @@ class Frame private constructor(private val handle: Handle<Frame>) : ByteBufHold
         fun from(streamId: Int,
                  type: FrameType,
                  payload: Payload,
-                 flags: Int = if (payload.hasMetadata()) FLAGS_M else 0): Frame {
-            val metadata = if (payload.hasMetadata()) Unpooled.wrappedBuffer(payload.metadata) else null
+                 flags: Int = if (payload.hasMetadata) FLAGS_M else 0): Frame {
+            val metadata = if (payload.hasMetadata) Unpooled.wrappedBuffer(payload.metadata) else null
             val data = Unpooled.wrappedBuffer(payload.data)
             return from(streamId, type, metadata, data, flags)
         }

rsocket-core/src/main/kotlin/io/rsocket/kotlin/FrameType.kt

@@ -15,7 +15,7 @@
  */
 package io.rsocket.kotlin
 
-/** Types of [Frame] that can be sent.  */
+/** RSocket [Frame] types */
 enum class FrameType(val encodedType: Int, private val flags: Int = 0) {
     // blank type that is not defined
     UNDEFINED(0x00),
@@ -44,15 +44,13 @@ enum class FrameType(val encodedType: Int, private val flags: Int = 0) {
                     Flags.IS_REQUEST_TYPE or
                     Flags.HAS_INITIAL_REQUEST_N or
                     Flags.IS_FRAGMENTABLE),
-    // Requester mid-stream
     REQUEST_N(0x08),
     CANCEL(0x09, Flags.CAN_HAVE_METADATA),
-    // Responder
     PAYLOAD(0x0A, Flags.CAN_HAVE_METADATA_AND_DATA),
     ERROR(0x0B, Flags.CAN_HAVE_METADATA_AND_DATA),
-    // Requester & Responder
+    // Requester & Responder metadata push
     METADATA_PUSH(0x0C, Flags.CAN_HAVE_METADATA),
-    // Resumption frames, not yet implemented
+    // Resumption frames
     RESUME(0x0D),
     RESUME_OK(0x0E),
     // synthetic types from Responder for use by the rest of the machinery

rsocket-core/src/main/kotlin/io/rsocket/kotlin/InterceptorOptions.kt

@@ -3,11 +3,23 @@ package io.rsocket.kotlin
 import io.rsocket.kotlin.interceptors.DuplexConnectionInterceptor
 import io.rsocket.kotlin.interceptors.RSocketInterceptor
 
+/**
+ * Configures RSocket interceptors
+ */
 interface InterceptorOptions {
 
+    /**
+     * @param interceptor adds [DuplexConnectionInterceptor]
+     */
     fun connection(interceptor: DuplexConnectionInterceptor)
 
+    /**
+     * @param interceptor adds [RSocketInterceptor] for requester [RSocket]
+     */
     fun requester(interceptor: RSocketInterceptor)
 
+    /**
+     * @param interceptor adds [RSocketInterceptor] for handler (responder) [RSocket]
+     */
     fun handler(interceptor: RSocketInterceptor)
 }

rsocket-core/src/main/kotlin/io/rsocket/kotlin/KeepAlive.kt

@@ -1,8 +1,18 @@
 package io.rsocket.kotlin
 
+/**
+ * Configures keep-alive feature of RSocket
+ */
 interface KeepAlive {
 
+    /**
+     * @return time between KEEPALIVE frames that the client will send
+     */
     fun keepAliveInterval(): Duration
 
+    /**
+     * @return time that a client will allow a server to not respond to a
+     * KEEPALIVE before it is assumed to be dead.
+     */
     fun keepAliveMaxLifeTime(): Duration
 }

rsocket-core/src/main/kotlin/io/rsocket/kotlin/KeepAliveData.kt

@@ -2,10 +2,19 @@ package io.rsocket.kotlin
 
 import java.nio.ByteBuffer
 
+/**
+ * Provides means to handle data sent by client with KEEPALIVE frame
+ */
 interface KeepAliveData {
 
+    /**
+     * @return supplier of Keep-alive data [ByteBuffer] sent by client
+     */
     fun producer(): () -> ByteBuffer
 
+    /**
+     * @return consumer of Keep-alive data [ByteBuffer] returned by server
+     */
     fun handler(): (ByteBuffer) -> Unit
 }
 

rsocket-core/src/main/kotlin/io/rsocket/kotlin/KeepAliveOptions.kt

@@ -2,32 +2,60 @@ package io.rsocket.kotlin
 
 import io.rsocket.kotlin.internal.EmptyKeepAliveData
 
+/**
+ * Configures Keep-alive feature of RSocket
+ */
 class KeepAliveOptions : KeepAlive {
     private var interval: Duration = Duration.ofMillis(100)
     private var maxLifeTime: Duration = Duration.ofSeconds(1)
     private var keepAliveData: KeepAliveData = EmptyKeepAliveData()
 
+    /**
+     * @param interval time between KEEPALIVE frames that the client will send
+     * @return this [KeepAliveOptions]
+     */
     fun keepAliveInterval(interval: Duration): KeepAliveOptions {
         assertDuration(interval, "keepAliveInterval")
         this.interval = interval
         return this
     }
 
+    /**
+     * @return time between KEEPALIVE frames that the client will send
+     */
     override fun keepAliveInterval() = interval
 
+    /**
+     * @param maxLifetime time that a client will allow a server to not respond to a
+     * KEEPALIVE before it is assumed to be dead.
+     */
     fun keepAliveMaxLifeTime(maxLifetime: Duration): KeepAliveOptions {
         assertDuration(maxLifetime, "keepAliveMaxLifeTime")
         this.maxLifeTime = maxLifetime
         return this
     }
 
+    /**
+     * @return time that a client will allow a server to not respond to a
+     * KEEPALIVE before it is assumed to be dead.
+     */
     override fun keepAliveMaxLifeTime() = maxLifeTime
 
+    /**
+     * Provides means to handle data sent by client with KEEPALIVE frame
+     *
+     * @param keepAliveData utility for handling data sent by client with KEEPALIVE
+     * frame
+     * @return this KeepAliveOptions
+     */
     fun keepAliveData(keepAliveData: KeepAliveData): KeepAliveOptions {
         this.keepAliveData = keepAliveData
         return this
     }
 
+    /**
+     * @return utility for handling data sent by client with KEEPALIVE
+     */
     fun keepAliveData(): KeepAliveData = keepAliveData
 
     fun copy(): KeepAliveOptions = KeepAliveOptions()