update ktor

build.gradle.kts

@@ -14,6 +14,8 @@
  * limitations under the License.
  */
 
+import com.github.benmanes.gradle.versions.updates.*
+
 buildscript {
     repositories {
         mavenCentral()
@@ -29,6 +31,10 @@ plugins {
     alias(libs.plugins.versionUpdates)
 }
 
+tasks.named<DependencyUpdatesTask>("dependencyUpdates") {
+    rejectVersionIf { candidate.group == "io.ktor" && candidate.version.contains("ide-debug") }
+}
+
 configureYarn()
 
 subprojects {

buildSrc/settings.gradle.kts

@@ -4,14 +4,12 @@ pluginManagement {
     repositories {
         gradlePluginPortal()
         mavenCentral()
-        maven("https://maven.pkg.jetbrains.space/kotlin/p/kotlin/dev")
     }
 }
 
 dependencyResolutionManagement {
     repositories {
         mavenCentral()
-        maven("https://maven.pkg.jetbrains.space/kotlin/p/kotlin/dev")
     }
 
     versionCatalogs {

gradle/libs.versions.toml

@@ -1,19 +1,19 @@
 [versions]
-kotlin = "1.6.20-M1-106"
+kotlin = "1.6.20"
 
-kotlinx-atomicfu = "0.17.0"
-kotlinx-coroutines = "1.6.0"
+kotlinx-atomicfu = "0.17.1"
+kotlinx-coroutines = "1.6.1"
 kotlinx-benchmark = "0.4.2"
 
-ktor = "2.0.0-eap-317"
+ktor = "2.0.0"
 
 turbine = "0.7.0"
 
-versionUpdates = "0.41.0"
+versionUpdates = "0.42.0"
 
-rsocket-java = "1.1.1"
+rsocket-java = "1.1.2"
 
-jmh = "1.34"
+jmh = "1.35"
 
 [libraries]
 kotlinx-coroutines-core = { module = "org.jetbrains.kotlinx:kotlinx-coroutines-core", version.ref = "kotlinx-coroutines" }
@@ -30,7 +30,8 @@ ktor-client-core = { module = "io.ktor:ktor-client-core", version.ref = "ktor" }
 ktor-client-websockets = { module = "io.ktor:ktor-client-websockets", version.ref = "ktor" }
 ktor-client-cio = { module = "io.ktor:ktor-client-cio", version.ref = "ktor" }
 ktor-client-okhttp = { module = "io.ktor:ktor-client-okhttp", version.ref = "ktor" }
-ktor-server-core = { module = "io.ktor:ktor-server", version.ref = "ktor" }
+ktor-client-darwin = { module = "io.ktor:ktor-client-darwin", version.ref = "ktor" }
+ktor-server-core = { module = "io.ktor:ktor-server-core", version.ref = "ktor" }
 ktor-server-websockets = { module = "io.ktor:ktor-server-websockets", version.ref = "ktor" }
 ktor-server-cio = { module = "io.ktor:ktor-server-cio", version.ref = "ktor" }
 ktor-server-netty = { module = "io.ktor:ktor-server-netty", version.ref = "ktor" }

gradle/wrapper/gradle-wrapper.properties

@@ -1,5 +1,5 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-7.3.3-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-7.4.2-bin.zip
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists

rsocket-transport-ktor/rsocket-transport-ktor-websocket-client/build.gradle.kts

@@ -35,3 +35,5 @@ kotlin {
 }
 
 description = "Ktor WebSocket Client RSocket transport implementation"
+
+evaluationDependsOn(":rsocket-transport-tests")

rsocket-transport-ktor/rsocket-transport-ktor-websocket/src/jsTest/kotlin/io/rsocket/kotlin/transport/ktor/websocket/ClientWebSocketTransportTest.kt → rsocket-transport-ktor/rsocket-transport-ktor-websocket-client/src/jsTest/kotlin/io/rsocket/kotlin/transport/ktor/websocket/client/ClientWebSocketTransportTest.kt

@@ -14,12 +14,11 @@
  * limitations under the License.
  */
 
-package io.rsocket.kotlin.transport.ktor.websocket
+package io.rsocket.kotlin.transport.ktor.websocket.client
 
 import io.ktor.client.*
 import io.ktor.client.engine.js.*
 import io.ktor.client.plugins.websocket.*
-import io.rsocket.kotlin.transport.ktor.websocket.client.*
 import io.rsocket.kotlin.transport.tests.*
 import kotlinx.coroutines.*
 

rsocket-transport-ktor/rsocket-transport-ktor-websocket-server/build.gradle.kts

@@ -19,7 +19,7 @@ plugins {
 }
 
 kotlin {
-    configureJvm {
+    configureCommon {
         main {
             dependencies {
                 api(projects.rsocketCore)
@@ -28,7 +28,25 @@ kotlin {
                 api(libs.ktor.server.websockets)
             }
         }
+        test {
+            dependencies {
+                implementation(projects.rsocketTransportKtor.rsocketTransportKtorWebsocketClient)
+                implementation(libs.ktor.client.cio)
+                implementation(libs.ktor.server.cio)
+            }
+        }
+    }
+    configureJvm {
+        test {
+            dependencies {
+                implementation(libs.ktor.client.okhttp)
+
+                implementation(libs.ktor.server.netty)
+                implementation(libs.ktor.server.jetty)
+            }
+        }
     }
+    configureNative(NativeTargets.Nix)
 }
 
 description = "Ktor WebSocket Server RSocket transport implementation"

rsocket-transport-ktor/rsocket-transport-ktor-websocket-server/src/jvmMain/kotlin/io/rsocket/kotlin/transport/ktor/websocket/server/RSocketSupport.kt → rsocket-transport-ktor/rsocket-transport-ktor-websocket-server/src/commonMain/kotlin/io/rsocket/kotlin/transport/ktor/websocket/server/RSocketSupport.kt

@@ -28,7 +28,7 @@ public class RSocketSupport(
         public var server: RSocketServer = RSocketServer()
     }
 
-    public companion object Feature : ApplicationPlugin<Application, Config, RSocketSupport> {
+    public companion object Feature : BaseApplicationPlugin<Application, Config, RSocketSupport> {
         override val key: AttributeKey<RSocketSupport> = AttributeKey("RSocket")
         override fun install(pipeline: Application, configure: Config.() -> Unit): RSocketSupport {
             pipeline.pluginOrNull(WebSockets)

rsocket-transport-ktor/rsocket-transport-ktor-websocket-server/src/commonTest/kotlin/io/rsocket/kotlin/transport/ktor/websocket/server/CIOWebSocketTransportTest.kt

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2015-2020 the original author or authors.
+ *
+ * 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.
+ */
+
+package io.rsocket.kotlin.transport.ktor.websocket.server
+
+import io.rsocket.kotlin.test.*
+import kotlin.test.*
+import kotlin.time.*
+import kotlin.time.Duration.Companion.minutes
+import io.ktor.client.engine.cio.CIO as ClientCIO
+import io.ktor.server.cio.CIO as ServerCIO
+
+class CIOWebSocketTransportTest : WebSocketTransportTest(ClientCIO, ServerCIO) {
+    //on native we need more time here
+    override val testTimeout: Duration = 5.minutes
+
+    //tests are ignored, because current CIO:native websockets implementation is unstable when working with large frames
+    @Test
+    @IgnoreNative
+    override fun largePayloadFireAndForget10() = super.largePayloadFireAndForget10()
+
+    @Test
+    @IgnoreNative
+    override fun largePayloadMetadataPush10() = super.largePayloadMetadataPush10()
+
+    @Test
+    @IgnoreNative
+    override fun largePayloadRequestChannel200() = super.largePayloadRequestChannel200()
+
+    @Test
+    @IgnoreNative
+    override fun largePayloadRequestResponse100() = super.largePayloadRequestResponse100()
+}

rsocket-transport-ktor/rsocket-transport-ktor-websocket/src/jvmTest/kotlin/io/rsocket/kotlin/transport/ktor/websocket/WebSocketConnectionTest.kt → rsocket-transport-ktor/rsocket-transport-ktor-websocket-server/src/commonTest/kotlin/io/rsocket/kotlin/transport/ktor/websocket/server/WebSocketConnectionTest.kt

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package io.rsocket.kotlin.transport.ktor.websocket
+package io.rsocket.kotlin.transport.ktor.websocket.server
 
 import io.ktor.client.*
 import io.ktor.server.application.*
@@ -25,7 +25,6 @@ import io.rsocket.kotlin.keepalive.*
 import io.rsocket.kotlin.payload.*
 import io.rsocket.kotlin.test.*
 import io.rsocket.kotlin.transport.ktor.websocket.client.*
-import io.rsocket.kotlin.transport.ktor.websocket.server.*
 import io.rsocket.kotlin.transport.tests.*
 import kotlinx.coroutines.*
 import kotlinx.coroutines.flow.*

rsocket-transport-ktor/rsocket-transport-ktor-websocket/src/jvmTest/kotlin/io/rsocket/kotlin/transport/ktor/websocket/WebSocketTransportTest.kt → rsocket-transport-ktor/rsocket-transport-ktor-websocket-server/src/commonTest/kotlin/io/rsocket/kotlin/transport/ktor/websocket/server/WebSocketTransportTest.kt

@@ -14,15 +14,14 @@
  * limitations under the License.
  */
 
-package io.rsocket.kotlin.transport.ktor.websocket
+package io.rsocket.kotlin.transport.ktor.websocket.server
 
 import io.ktor.client.*
 import io.ktor.client.engine.*
 import io.ktor.server.application.*
 import io.ktor.server.engine.*
 import io.ktor.server.routing.*
 import io.rsocket.kotlin.transport.ktor.websocket.client.*
-import io.rsocket.kotlin.transport.ktor.websocket.server.*
 import io.rsocket.kotlin.transport.tests.*
 import kotlinx.coroutines.*
 import io.ktor.client.plugins.websocket.WebSockets as ClientWebSockets
@@ -42,7 +41,7 @@ abstract class WebSocketTransportTest(
     }
 
     override suspend fun before() {
-        (GlobalScope + testJob).embeddedServer(serverEngine, port) {
+        testScope.embeddedServer(serverEngine, port) {
             install(ServerWebSockets)
             install(ServerRSocketSupport) { server = SERVER }
             install(Routing) { rSocket(acceptor = ACCEPTOR) }

rsocket-transport-ktor/rsocket-transport-ktor-websocket/src/jvmTest/kotlin/io/rsocket/kotlin/transport/ktor/websocket/WebSocketTransportTests.kt → rsocket-transport-ktor/rsocket-transport-ktor-websocket-server/src/jvmTest/kotlin/io/rsocket/kotlin/transport/ktor/websocket/server/WebSocketTransportTests.kt

@@ -14,18 +14,14 @@
  * limitations under the License.
  */
 
-package io.rsocket.kotlin.transport.ktor.websocket
+package io.rsocket.kotlin.transport.ktor.websocket.server
 
 import io.ktor.client.engine.okhttp.*
 import io.ktor.server.jetty.*
 import io.ktor.server.netty.*
 import io.ktor.client.engine.cio.CIO as ClientCIO
 import io.ktor.server.cio.CIO as ServerCIO
 
-//cio client and cio server - default for tests
-
-class CIOWebSocketTransportTest : WebSocketTransportTest(ClientCIO, ServerCIO)
-
 class OkHttpClientWebSocketTransportTest : WebSocketTransportTest(OkHttp, ServerCIO)
 
 class NettyServerWebSocketTransportTest : WebSocketTransportTest(ClientCIO, Netty)

rsocket-transport-ktor/rsocket-transport-ktor-websocket/build.gradle.kts

@@ -26,30 +26,10 @@ kotlin {
                 api(libs.ktor.websockets)
             }
         }
-        test {
-            dependencies {
-                implementation(projects.rsocketTransportKtor.rsocketTransportKtorWebsocketClient)
-            }
-        }
-    }
-    configureJvm {
-        test {
-            dependencies {
-                implementation(projects.rsocketTransportKtor.rsocketTransportKtorWebsocketServer)
-
-                implementation(libs.ktor.client.cio)
-                implementation(libs.ktor.client.okhttp)
-
-                implementation(libs.ktor.server.cio)
-                implementation(libs.ktor.server.netty)
-                implementation(libs.ktor.server.jetty)
-            }
-        }
     }
+    configureJvm()
     configureJs()
     configureNative()
 }
 
 description = "Ktor WebSocket RSocket transport implementation"
-
-evaluationDependsOn(":rsocket-transport-tests")

rsocket-transport-nodejs-tcp/src/jsTest/kotlin/io/rsocket/kotlin/transport/nodejs/tcp/TcpTransportTest.kt

@@ -10,7 +10,7 @@ class TcpTransportTest : TransportTest() {
     override suspend fun before() {
         val port = PortProvider.next()
         server = startServer(TcpServerTransport(port, "127.0.0.1", InUseTrackingPool))
-        client = connectClient(TcpClientTransport(port, "127.0.0.1", InUseTrackingPool, testJob))
+        client = connectClient(TcpClientTransport(port, "127.0.0.1", InUseTrackingPool, testContext))
     }
 
     override suspend fun after() {

rsocket-transport-tests/build.gradle.kts

@@ -79,7 +79,7 @@ val startTransportTestServer by tasks.registering(StartTransportTestServer::clas
 }
 
 rootProject.allprojects {
-    if (name == "rsocket-transport-ktor-websocket") {
+    if (name == "rsocket-transport-ktor-websocket-client") {
         val names = setOf(
             "jsLegacyNodeTest",
             "jsIrNodeTest",

rsocket-transport-tests/src/commonMain/kotlin/io/rsocket/kotlin/transport/tests/TransportTest.kt

@@ -33,7 +33,7 @@ import kotlin.time.Duration.Companion.seconds
 abstract class TransportTest : SuspendTest, TestWithLeakCheck {
     override val testTimeout: Duration = 3.minutes
 
-    protected val testJob = Job()
+    private val testJob = SupervisorJob()
     protected val testContext = testJob + TestExceptionHandler
     protected val testScope = CoroutineScope(testContext)
 
@@ -53,21 +53,25 @@ abstract class TransportTest : SuspendTest, TestWithLeakCheck {
     @Test
     fun fireAndForget10() = test {
         (1..10).map { async { client.fireAndForget(payload(it)) } }.awaitAll()
+        delay(1000) //TODO: leak check
     }
 
     @Test
-    fun largePayloadFireAndForget10() = test {
-        (1..10).map { async { client.fireAndForget(requesterLargeMetadata) } }.awaitAll()
+    open fun largePayloadFireAndForget10() = test {
+        (1..10).map { async { client.fireAndForget(requesterLargePayload) } }.awaitAll()
+        delay(1000) //TODO: leak check
     }
 
     @Test
     fun metadataPush10() = test {
         (1..10).map { async { client.metadataPush(packet(requesterData)) } }.awaitAll()
+        delay(1000) //TODO: leak check
     }
 
     @Test
-    fun largePayloadMetadataPush10() = test {
+    open fun largePayloadMetadataPush10() = test {
         (1..10).map { async { client.metadataPush(packet(requesterLargeData)) } }.awaitAll()
+        delay(1000) //TODO: leak check
     }
 
     @Test
@@ -93,12 +97,12 @@ abstract class TransportTest : SuspendTest, TestWithLeakCheck {
     }
 
     @Test
-    fun largePayloadRequestChannel200() = test {
+    open fun largePayloadRequestChannel200() = test {
         val request = flow {
-            repeat(200) { emit(requesterLargeMetadata) }
+            repeat(200) { emit(requesterLargePayload) }
         }
         val list =
-            client.requestChannel(requesterLargeMetadata, request)
+            client.requestChannel(requesterLargePayload, request)
                 .flowOn(PrefetchStrategy(Int.MAX_VALUE, 0))
                 .onEach { it.close() }
                 .toList()
@@ -174,6 +178,7 @@ abstract class TransportTest : SuspendTest, TestWithLeakCheck {
                     assertEquals(requesterMetadata, it.metadata?.readText())
                 }.toList()
         assertEquals(500, list.size)
+        delay(1000) //TODO: leak check
     }
 
     @Test
@@ -192,8 +197,8 @@ abstract class TransportTest : SuspendTest, TestWithLeakCheck {
     }
 
     @Test
-    fun largePayloadRequestResponse100() = test {
-        (1..100).map { async { client.requestResponse(requesterLargeMetadata) } }.awaitAll().onEach { it.close() }
+    open fun largePayloadRequestResponse100() = test {
+        (1..100).map { async { client.requestResponse(requesterLargePayload) } }.awaitAll().onEach { it.close() }
     }
 
     @Test
@@ -230,6 +235,7 @@ abstract class TransportTest : SuspendTest, TestWithLeakCheck {
                 .onEach { checkPayload(it) }
                 .toList()
         assertEquals(500, list.size)
+        delay(1000) //TODO: leak check
     }
 
     companion object {
@@ -249,7 +255,7 @@ abstract class TransportTest : SuspendTest, TestWithLeakCheck {
         const val requesterMetadata: String = "metadata"
 
         val requesterLargeData = "large.text.12345".repeat(2000)
-        val requesterLargeMetadata get() = payload(requesterLargeData, requesterLargeData)
+        val requesterLargePayload get() = payload(requesterLargeData, requesterLargeData)
 
         fun payload(metadataPresent: Int): Payload {
             val metadata = when (metadataPresent % 5) {