Add binary message support (#7)

* passing more Java version's tests

* support binary message and ack

* add ws binary test

* fix test

* fix broadcast case

Author: Piasy

README.md

@@ -48,7 +48,8 @@ IO.socket("http://localhost:3000", IO.Options()) { socket ->
     socket.on(Socket.EVENT_CONNECT) { args ->
         println("on connect ${args.joinToString()}")
 
-        socket.emit("echo", 1, "2", GMTDate())
+        val bin = UnsafeByteStringOperations.wrapUnsafe(byteArrayOf(0x1, 0x3, 0x1, 0x4))
+        socket.emit("echo", 1, "2", bin, GMTDate())
     }.on("echoBack") { args ->
         println("on echoBack ${args.joinToString()}")
     }
@@ -57,6 +58,29 @@ IO.socket("http://localhost:3000", IO.Options()) { socket ->
 }
 ```
 
+Most of the APIs are the same as socket.io-client-java, here are some differences:
+
+- Create socket is asynchronous, to make it's easier to guarantee thread safety.
+- Binary messages can't be nested, because `emit` only accepts String/Boolean/Number/JsonElement/ByteString, other types will be converted to String using `toString()`, so there is no way to put ByteString in JsonElement.
+
+### Set logging callback
+
+```kotlin
+Logger.setLogger(object : LoggerInterface {
+    override fun debug(tag: String, log: String) {
+        TODO("Not yet implemented")
+    }
+
+    override fun info(tag: String, log: String) {
+        TODO("Not yet implemented")
+    }
+
+    override fun error(tag: String, log: String) {
+        TODO("Not yet implemented")
+    }
+})
+```
+
 ## Example
 
 ### Android

build.gradle.kts

@@ -20,6 +20,7 @@ plugins {
     alias(libs.plugins.versionUpdate)
 }
 
+// ./gradlew versionCatalogUpdate
 versionCatalogUpdate {
     sortByKey = false
     keep {

example/shared/src/commonMain/kotlin/com/piasy/kmp/socketio/example/Greeting.kt

@@ -1,16 +1,37 @@
 package com.piasy.kmp.socketio.example
 
+import com.piasy.kmp.socketio.logging.Logger
+import com.piasy.kmp.socketio.logging.LoggerInterface
 import com.piasy.kmp.socketio.socketio.IO
 import com.piasy.kmp.socketio.socketio.Socket
 import io.ktor.util.date.GMTDate
+import kotlinx.io.bytestring.ByteString
+import kotlinx.io.bytestring.unsafe.UnsafeByteStringApi
+import kotlinx.io.bytestring.unsafe.UnsafeByteStringOperations
 
 class Greeting {
+    @OptIn(UnsafeByteStringApi::class)
     fun greet() {
+        Logger.setLogger(object : LoggerInterface {
+            override fun debug(tag: String, log: String) {
+                TODO("Not yet implemented")
+            }
+
+            override fun info(tag: String, log: String) {
+                TODO("Not yet implemented")
+            }
+
+            override fun error(tag: String, log: String) {
+                TODO("Not yet implemented")
+            }
+        })
+
         IO.socket("http://172.16.11.25:3000", IO.Options()) { socket ->
             socket.on(Socket.EVENT_CONNECT) { args ->
                 println("on connect ${args.joinToString()}")
 
-                socket.emit("echo", 1, "2", GMTDate())
+                val bin = UnsafeByteStringOperations.wrapUnsafe(byteArrayOf(0x1, 0x3, 0x1, 0x4))
+                socket.emit("echo", 1, "2", bin, GMTDate())
             }.on("echoBack") { args ->
                 println("on echoBack ${args.joinToString()}")
             }

gradle/libs.versions.toml

@@ -4,22 +4,21 @@ iosDeploymentTarget = "14.0"
 compileSdk = "35"
 minSdk = "21"
 targetSdk = "35"
-agp = "8.7.2"
+agp = "8.8.0"
 kotlin = "2.1.0"
-mockk = "1.13.13"
+mockk = "1.13.16"
 ktor = "3.0.3"
 coroutine = "1.10.1"
 compose = "1.7.5"
 
 [libraries]
 junit = "junit:junit:4.13.2"
-hamcrest = "org.hamcrest:hamcrest-library:1.3"
+hamcrest = "org.hamcrest:hamcrest-library:3.0"
 kotlin-test = { module = "org.jetbrains.kotlin:kotlin-test", version.ref = "kotlin" }
 mockk = { module = "io.mockk:mockk", version.ref = "mockk" }
 kotlinx-coroutines-test = { module = "org.jetbrains.kotlinx:kotlinx-coroutines-test", version.ref = "coroutine" }
-json = "org.json:json:20241224"
-
-socketioParser = "org.hildan.socketio:socketio-kotlin:1.6.0"
+json = "org.json:json:20250107"
+socketioParser = "org.hildan.socketio:socketio-kotlin:2.0.0"
 kotlinx-coroutines-core = { module = "org.jetbrains.kotlinx:kotlinx-coroutines-core", version.ref = "coroutine" }
 ktor-client-core = { module = "io.ktor:ktor-client-core", version.ref = "ktor" }
 ktor-client-websockets = { module = "io.ktor:ktor-client-websockets", version.ref = "ktor" }

gradle/wrapper/gradle-wrapper.properties

@@ -1,6 +1,6 @@
 #Wed Nov 16 22:27:43 CST 2022
 distributionBase=GRADLE_USER_HOME
-distributionUrl=https\://services.gradle.org/distributions/gradle-8.9-all.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-8.12-all.zip
 distributionPath=wrapper/dists
 zipStorePath=wrapper/dists
 zipStoreBase=GRADLE_USER_HOME

kmp-socketio/src/commonMain/kotlin/com/piasy/kmp/socketio/engineio/EngineSocket.kt

@@ -19,6 +19,7 @@ class EngineSocket(
     @JvmField internal val opt: Options,
     private val scope: CoroutineScope,
     private val factory: TransportFactory = DefaultTransportFactory,
+    private val rawMessage: Boolean = false,
 ) : Emitter() {
     open class Options : Transport.Options() {
         /**
@@ -42,7 +43,7 @@ class EngineSocket(
 
     internal var disablePingTimeout = false // to help unit test
     private var state = State.INIT
-    private var id = ""
+    internal var id = ""
     private var upgrades = emptyList<String>()
     private var pingInterval = 0
     private var pingTimeout = 0
@@ -134,7 +135,12 @@ class EngineSocket(
      */
     @WorkThread
     fun send(packet: EngineIOPacket<*>) {
-        sendPacket(packet)
+        sendPackets(listOf(packet))
+    }
+
+    @WorkThread
+    fun send(packets: List<EngineIOPacket<*>>) {
+        sendPackets(packets)
     }
 
     /**
@@ -212,7 +218,7 @@ class EngineSocket(
         opts.timestampParam = options?.timestampParam ?: opt.timestampParam
         opts.extraHeaders = opt.extraHeaders
 
-        val transport = factory.create(name, opts, scope)
+        val transport = factory.create(name, opts, scope, rawMessage)
         emit(EVENT_TRANSPORT, transport)
         return transport
     }
@@ -283,15 +289,15 @@ class EngineSocket(
     }
 
     @WorkThread
-    private fun sendPacket(packet: EngineIOPacket<*>) {
-        Logger.debug(TAG, "sendPacket: state $state, pkt $packet")
+    private fun sendPackets(packets: List<EngineIOPacket<*>>) {
+        Logger.debug(TAG, "sendPackets: state $state, $packets")
         if (state != State.OPENING && state != State.OPEN) {
-            Logger.error(TAG, "sendPacket at wrong state: $state")
+            Logger.error(TAG, "sendPackets at wrong state: $state")
             return
         }
 
-        emit(EVENT_PACKET_CREATE, packet)
-        writeBuffer.addLast(packet)
+        emit(EVENT_PACKET_CREATE, packets.size)
+        writeBuffer.addAll(packets)
         flush()
     }
 
@@ -310,14 +316,14 @@ class EngineSocket(
             is EngineIOPacket.Open -> onHandshake(packet)
             is EngineIOPacket.Ping -> {
                 emit(EVENT_PING)
-                sendPacket(EngineIOPacket.Pong(null))
+                sendPackets(listOf(EngineIOPacket.Pong(null)))
             }
 
+            is EngineIOPacket.BinaryData -> emit(EVENT_DATA, packet.payload)
             is EngineIOPacket.Message<*> -> {
                 val data = packet.payload
                 if (data != null) {
                     emit(EVENT_DATA, data)
-                    emit(EVENT_MESSAGE, data)
                 }
             }
 
@@ -357,7 +363,7 @@ class EngineSocket(
         emit(EVENT_OPEN)
         flush()
 
-        if (opt.upgrade && transport?.name == PollingXHR.NAME) {
+        if (opt.upgrade && transport?.name == PollingXHR.NAME && upgrades.isNotEmpty()) {
             Logger.info(TAG, "starting upgrade probes")
             for (upgrade in upgrades) {
                 probe(upgrade)
@@ -591,7 +597,7 @@ class EngineSocket(
         /**
          * Called when data is received from the server.
          */
-        const val EVENT_MESSAGE = "message"
+        const val EVENT_DATA = "data"
 
         /**
          * Called when an error occurs.
@@ -618,7 +624,6 @@ class EngineSocket(
         const val EVENT_PACKET = "packet"
         const val EVENT_PACKET_CREATE = "packetCreate"
         const val EVENT_HEARTBEAT = "heartbeat"
-        const val EVENT_DATA = "data"
         const val EVENT_PING = "ping"
 
         /**

kmp-socketio/src/commonMain/kotlin/com/piasy/kmp/socketio/engineio/Transport.kt

@@ -5,14 +5,14 @@ import com.piasy.kmp.socketio.logging.Logger
 import com.piasy.kmp.socketio.parseqs.ParseQS
 import io.ktor.util.date.*
 import kotlinx.coroutines.CoroutineScope
-import org.hildan.socketio.EngineIO
 import org.hildan.socketio.EngineIOPacket
 import kotlin.jvm.JvmField
 
 abstract class Transport(
     internal val opt: Options,
     protected val scope: CoroutineScope,
     val name: String,
+    protected val rawMessage: Boolean,
 ) : Emitter() {
     open class Options {
         // @JvmField for jvm test code
@@ -60,7 +60,7 @@ abstract class Transport(
         if (state == State.OPEN) {
             doSend(packets)
         } else {
-            throw RuntimeException("Transport not open")
+            onError("Transport not open")
         }
     }
 
@@ -88,21 +88,6 @@ abstract class Transport(
         }
     }
 
-    @WorkThread
-    protected fun onWsData(data: String) {
-        logD("onData: `$data`")
-        if (stringMessagePayloadForTesting) {
-            onPacket(EngineIO.decodeWsFrame(data, deserializePayload = { it }))
-        } else {
-            onPacket(EngineIO.decodeSocketIO(data))
-        }
-    }
-
-    @WorkThread
-    protected fun onWsData(data: ByteArray) {
-        // TODO: binary
-    }
-
     @WorkThread
     protected fun onPacket(packet: EngineIOPacket<*>) {
         logD("onPacket $packet")
@@ -184,7 +169,6 @@ abstract class Transport(
         const val EVENT_REQUEST_HEADERS: String = "requestHeaders"
         const val EVENT_RESPONSE_HEADERS: String = "responseHeaders"
 
-        internal var stringMessagePayloadForTesting = false
         private const val TAG = "Transport"
     }
 }

kmp-socketio/src/commonMain/kotlin/com/piasy/kmp/socketio/engineio/transports/PollingXHR.kt

@@ -11,17 +11,15 @@ import io.ktor.util.*
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.launch
-import org.hildan.socketio.EngineIO
-import org.hildan.socketio.EngineIOPacket
-import org.hildan.socketio.SocketIO
-import org.hildan.socketio.SocketIOPacket
+import org.hildan.socketio.*
 
 open class PollingXHR(
     opt: Options,
     scope: CoroutineScope,
     private val ioScope: CoroutineScope = CoroutineScope(Dispatchers.Default),
     private val factory: HttpClientFactory = DefaultHttpClientFactory,
-) : Transport(opt, scope, NAME) {
+    rawMessage: Boolean,
+) : Transport(opt, scope, NAME, rawMessage) {
     private var polling = false
 
     @WorkThread
@@ -138,10 +136,17 @@ open class PollingXHR(
     @WorkThread
     private fun onPollComplete(data: String) {
         logD("onPollComplete: state $state, `$data`")
-        val packets = if (stringMessagePayloadForTesting) {
-            EngineIO.decodeHttpBatch(data, deserializeTextPayload = { it })
-        } else {
-            EngineIO.decodeHttpBatch(data, SocketIO::decode)
+        val packets = try {
+            if (rawMessage) {
+                EngineIO.decodeHttpBatch(data, deserializePayload = { it })
+            } else {
+                EngineIO.decodeHttpBatch(data, SocketIO::decode)
+            }
+        } catch (e: InvalidSocketIOPacketException) {
+            val log = "onPollComplete decode error: ${e.message}"
+            logE(log)
+            onError(log)
+            return
         }
         for (pkt in packets) {
             if ((state == State.OPENING || state == State.CLOSING) && pkt is EngineIOPacket.Open) {
@@ -171,17 +176,10 @@ open class PollingXHR(
     @WorkThread
     override fun doSend(packets: List<EngineIOPacket<*>>) {
         writable = false
-        @Suppress("UNCHECKED_CAST")
-        val data = if (stringMessagePayloadForTesting) {
-            EngineIO.encodeHttpBatch(
-                packets as List<EngineIOPacket<String>>,
-                serializePayload = { it }
-            )
+        val data = if (rawMessage) {
+            EngineIO.encodeHttpBatch(packets, serializePayload = { it.toString() })
         } else {
-            EngineIO.encodeHttpBatch(
-                packets as List<EngineIOPacket<SocketIOPacket>>,
-                SocketIO::encode
-            )
+            EngineIO.encodeHttpBatch(packets, serializePayload = { SocketIO.encode(it as SocketIOPacket) })
         }
 
         val method = HttpMethod.Post