Nodejs TCP server and client transport

.github/workflows/gradle-main.yml

@@ -86,6 +86,15 @@ jobs:
           distributions-cache-enabled: false
           dependencies-cache-enabled: false
           configuration-cache-enabled: false
+      - name: Test rsocket-transport-nodejs module
+        if: (matrix.target == 'jsIrNode' || matrix.target == 'jsLegacyNode') && (success() || failure())
+        timeout-minutes: 15
+        uses: gradle/gradle-build-action@v1
+        with:
+          arguments: rsocket-transport-nodejs-tcp:${{ matrix.target }}Test --scan --info
+          distributions-cache-enabled: false
+          dependencies-cache-enabled: false
+          configuration-cache-enabled: false
 
       - name: Publish Test Report
         if: always()

.github/workflows/gradle-pr.yml

@@ -81,6 +81,15 @@ jobs:
           distributions-cache-enabled: false
           dependencies-cache-enabled: false
           configuration-cache-enabled: false
+      - name: Test rsocket-transport-nodejs module
+        if: (matrix.target == 'jsIrNode' || matrix.target == 'jsLegacyNode') && (success() || failure())
+        timeout-minutes: 15
+        uses: gradle/gradle-build-action@v1
+        with:
+          arguments: rsocket-transport-nodejs-tcp:${{ matrix.target }}Test --scan --info
+          distributions-cache-enabled: false
+          dependencies-cache-enabled: false
+          configuration-cache-enabled: false
 
       - name: Publish Test Report
         if: always()

.github/workflows/run-tests.yml

@@ -80,6 +80,15 @@ jobs:
           distributions-cache-enabled: false
           dependencies-cache-enabled: false
           configuration-cache-enabled: false
+      - name: Test rsocket-transport-nodejs module
+        if: (matrix.target == 'jsIrNode' || matrix.target == 'jsLegacyNode') && (success() || failure())
+        timeout-minutes: 15
+        uses: gradle/gradle-build-action@v1
+        with:
+          arguments: rsocket-transport-nodejs-tcp:${{ matrix.target }}Test --scan --info
+          distributions-cache-enabled: false
+          dependencies-cache-enabled: false
+          configuration-cache-enabled: false
 
       - name: Publish Test Report
         if: always()

build.gradle.kts

@@ -65,10 +65,12 @@ subprojects {
                 project.name != "rsocket-transport-ktor" &&
                         project.name != "rsocket-transport-ktor-client"
 
+            val jsOnly = project.name == "rsocket-transport-nodejs-tcp"
+            val nodejsOnly = project.name == "rsocket-transport-nodejs-tcp"
 
             if (!isAutoConfigurable) return@configure
 
-            jvm {
+            if (!jsOnly) jvm {
                 testRuns.all {
                     executionTask.configure {
                         // ActiveProcessorCount is used here, to make sure local setup is similar as on CI
@@ -90,7 +92,7 @@ subprojects {
                         }
                     }
                 }
-                browser {
+                if (!nodejsOnly) browser {
                     testTask {
                         useKarma {
                             useConfigDirectory(rootDir.resolve("js").resolve("karma.config.d"))
@@ -100,6 +102,8 @@ subprojects {
                 }
             }
 
+            if (jsOnly) return@configure
+
             //native targets configuration
             val linuxTargets = listOf(linuxX64())
             val mingwTargets = if (supportMingw) listOf(mingwX64()) else emptyList()
@@ -317,7 +321,7 @@ subprojects {
         }
 
         tasks.matching { it.name == "generatePomFileForKotlinMultiplatformPublication" }.configureEach {
-            dependsOn(tasks["generatePomFileForJvmPublication"])
+            tasks.findByName("generatePomFileForJvmPublication")?.let { dependsOn(it) }
         }
     }
 }

gradle.properties

@@ -23,7 +23,6 @@ kotlinxCoroutinesVersion=1.5.2-native-mt
 kotlinxAtomicfuVersion=0.17.0
 kotlinxSerializationVersion=1.3.1
 kotlinxBenchmarkVersion=0.4.0
-kotlinxNodejsVersion=0.0.7
 rsocketJavaVersion=1.1.1
 turbineVersion=0.7.0
 artifactoryVersion=4.25.1

examples/nodejs-tcp-transport/build.gradle.kts → rsocket-transport-nodejs-tcp/build.gradle.kts

@@ -16,22 +16,18 @@
 
 plugins {
     kotlin("multiplatform")
-}
+    id("kotlinx-atomicfu")
 
-val kotlinxNodejsVersion: String by rootProject
+    signing
+    `maven-publish`
+    id("com.jfrog.artifactory")
+}
 
 kotlin {
-    js(IR) {
-        nodejs {
-            binaries.executable()
-        }
-    }
-
     sourceSets {
         val jsMain by getting {
             dependencies {
-                implementation(project(":rsocket-core"))
-                implementation("org.jetbrains.kotlinx:kotlinx-nodejs:$kotlinxNodejsVersion")
+                api(project(":rsocket-core"))
             }
         }
     }

rsocket-transport-nodejs-tcp/src/jsMain/kotlin/io/rsocket/kotlin/transport/nodejs/tcp/FrameWithLengthAssembler.kt

@@ -0,0 +1,44 @@
+package io.rsocket.kotlin.transport.nodejs.tcp
+
+import io.ktor.utils.io.core.*
+import io.rsocket.kotlin.frame.io.*
+
+internal fun ByteReadPacket.withLength(): ByteReadPacket = buildPacket {
+    @Suppress("INVISIBLE_MEMBER") writeLength([email protected]())
+    writePacket(this@withLength)
+}
+
+internal class FrameWithLengthAssembler(private val onFrame: (frame: ByteReadPacket) -> Unit) {
+    private var expectedFrameLength = 0 //TODO atomic for native
+    private val packetBuilder: BytePacketBuilder = BytePacketBuilder()
+    inline fun write(write: BytePacketBuilder.() -> Unit) {
+        packetBuilder.write()
+        loop()
+    }
+
+    private fun loop() {
+        while (true) when {
+            expectedFrameLength == 0 && packetBuilder.size < 3 -> return // no length
+            expectedFrameLength == 0                           -> withTemp { // has length
+                expectedFrameLength = @Suppress("INVISIBLE_MEMBER") it.readLength()
+                if (it.remaining >= expectedFrameLength) build(it) // if has length and frame
+            }
+            packetBuilder.size < expectedFrameLength           -> return // not enough bytes to read frame
+            else                                               -> withTemp { build(it) } // enough bytes to read frame
+        }
+    }
+
+    private fun build(from: ByteReadPacket) {
+        val frame = buildPacket {
+            writePacket(from, expectedFrameLength)
+        }
+        expectedFrameLength = 0
+        onFrame(frame)
+    }
+
+    private inline fun withTemp(block: (tempPacket: ByteReadPacket) -> Unit) {
+        val tempPacket = packetBuilder.build()
+        block(tempPacket)
+        packetBuilder.writePacket(tempPacket)
+    }
+}

rsocket-transport-nodejs-tcp/src/jsMain/kotlin/io/rsocket/kotlin/transport/nodejs/tcp/TcpClientTransport.kt

@@ -0,0 +1,25 @@
+package io.rsocket.kotlin.transport.nodejs.tcp
+
+import io.ktor.utils.io.core.internal.*
+import io.ktor.utils.io.pool.*
+import io.rsocket.kotlin.*
+import io.rsocket.kotlin.transport.*
+import io.rsocket.kotlin.transport.nodejs.tcp.internal.*
+import kotlinx.coroutines.*
+import kotlin.coroutines.*
+
+public class TcpClientTransport(
+    private val port: Int,
+    private val hostname: String,
+    private val pool: ObjectPool<ChunkBuffer> = ChunkBuffer.Pool,
+    coroutineContext: CoroutineContext = EmptyCoroutineContext
+) : ClientTransport {
+
+    override val coroutineContext: CoroutineContext = coroutineContext + SupervisorJob(coroutineContext[Job])
+
+    @TransportApi
+    override suspend fun connect(): Connection {
+        val socket = connect(port, hostname)
+        return TcpConnection(coroutineContext, pool, socket)
+    }
+}