Address PR #30 review: fix cross-platform tests, remove redundant APIs

- Fix commonTest for all platforms: drop JVM-only imports, use encodeToByteArray/ByteString comparison
- Remove Manager.openSuspend() entirely
- Remove Socket.emitSuspend(event, *args) without ack; use socket.emit() in tests/examples
- RESERVED_EVENTS back to private; drop server-side events (disconnecting, newListener, removeListener)
- Document Flow vs listener API in Emitter.flow() KDoc; flowWithReplay simplified
- README and examples updated

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: zhimbura

README.md

@@ -86,9 +86,9 @@ socket.flow("echoBack").collect { args ->
 // Open socket using suspend function
 socket.openSuspend()
 
-// Emit event using suspend function
+// Emit event using regular emit; suspend is only needed for ack
 val bin = UnsafeByteStringOperations.wrapUnsafe(byteArrayOf(0x1, 0x3, 0x1, 0x4))
-socket.emitSuspend("echo", 1, "2", bin, GMTDate())
+socket.emit("echo", 1, "2", bin, GMTDate())
 
 // Emit with suspend acknowledgement callback (trailing lambda syntax)
 socket.emitSuspend("echoWithAck", 42, "test") { args ->
@@ -104,9 +104,7 @@ Most of the APIs are the same as socket.io-client-java, here are some difference
 ### Available Extension Functions
 
 - `IO.socketSuspend(uri, opt)` - Create socket using coroutines
-- `Manager.openSuspend()` - Open manager using coroutines
 - `Socket.openSuspend()` - Open socket using coroutines
-- `Socket.emitSuspend(event, *args)` - Emit event using coroutines
 - `Socket.emitSuspend(event, *args, ack: suspend (Array<out Any>) -> Unit)` - Emit with suspend acknowledgement callback (trailing lambda)
 - `Manager.stateFlow` - StateFlow for tracking manager state
 - `Socket.connectedFlow` - StateFlow for tracking connection state

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

@@ -58,9 +58,9 @@ class CoroutinesExample {
         // Wait a bit for connection
         kotlinx.coroutines.delay(1000)
 
-        // Emit event using suspend function
+        // Emit event using regular API; suspend is only needed for ack
         val bin = UnsafeByteStringOperations.wrapUnsafe(byteArrayOf(0x1, 0x3, 0x1, 0x4))
-        socket.emitSuspend("echo", 1, "2", bin, GMTDate())
+        socket.emit("echo", 1, "2", bin, GMTDate())
 
         // Emit with suspend acknowledgement callback (trailing lambda syntax)
         socket.emitSuspend("echoWithAck", 42, "test") { args ->
@@ -98,7 +98,7 @@ class CoroutinesExample {
         // Example 1: Handle errors in suspend functions with try-catch
         try {
             socket.openSuspend()
-            socket.emitSuspend("someEvent", "data")
+            socket.emit("someEvent", "data")
         } catch (e: Exception) {
             println("Error occurred: ${e.message}")
             // Handle error appropriately
@@ -263,7 +263,7 @@ class CoroutinesExample {
             delay(1000)
 
             // Use socket for some operations
-            socket.emitSuspend("someEvent", "data")
+            socket.emit("someEvent", "data")
 
             // Step 2: Close old socket properly
             println("Closing old socket...")
@@ -293,7 +293,7 @@ class CoroutinesExample {
             delay(1000)
 
             // Use new socket
-            socket.emitSuspend("someEvent", "new_data")
+            socket.emit("someEvent", "new_data")
 
             delay(1000)
 
@@ -337,7 +337,7 @@ class CoroutinesExample {
             try {
                 socket.openSuspend()
                 delay(1000)
-                socket.emitSuspend("test", "data")
+                socket.emit("test", "data")
                 delay(1000)
             } catch (e: Exception) {
                 println("Error: ${e.message}")

kmp-socketio/src/commonMain/kotlin/com/piasy/kmp/socketio/socketio/CoroutinesExtensions.kt

@@ -1,8 +1,6 @@
 package com.piasy.kmp.socketio.socketio
 
 import com.piasy.kmp.socketio.emitter.Emitter
-import com.piasy.kmp.socketio.engineio.State
-import com.piasy.kmp.socketio.engineio.WorkThread
 import com.piasy.kmp.xlog.Logging
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
@@ -11,32 +9,6 @@ import kotlinx.coroutines.suspendCancellableCoroutine
 import kotlin.coroutines.resume
 import kotlin.coroutines.resumeWithException
 
-/**
- * Extension function to open Manager using coroutines.
- * Converts the callback-based open() method to a suspend function.
- * 
- * @throws Exception if connection fails or times out.
- */
-@WorkThread
-suspend fun Manager.openSuspend() {
-    suspendCancellableCoroutine<Unit> { continuation ->
-        var callbackInvoked = false
-        open { error ->
-            if (!callbackInvoked) {
-                callbackInvoked = true
-                if (error.isEmpty()) {
-                    continuation.resume(Unit)
-                } else {
-                    continuation.resumeWithException(Exception(error))
-                }
-            }
-        }
-        continuation.invokeOnCancellation {
-            // Cleanup if cancelled
-        }
-    }
-}
-
 /**
  * Extension function to open Socket using coroutines.
  * Converts the callback-based open() method to a suspend function.
@@ -71,32 +43,6 @@ suspend fun Socket.openSuspend() {
     }
 }
 
-/**
- * Extension function to emit event using coroutines.
- * Supports both regular Ack and suspend lambda for acknowledgements.
- * 
- * @param event event name
- * @param args only accepts String/Boolean/Number/JsonElement/ByteString
- */
-suspend fun Socket.emitSuspend(event: String, vararg args: Any) {
-    // Check reserved events manually (since RESERVED_EVENTS is private)
-    val reservedEvents = setOf(
-        Socket.EVENT_CONNECT,
-        Socket.EVENT_CONNECT_ERROR,
-        Socket.EVENT_DISCONNECT,
-        "disconnecting",
-        "newListener",
-        "removeListener",
-    )
-    if (reservedEvents.contains(event)) {
-        Logging.error(Socket.TAG, "emit reserved event: $event")
-        return
-    }
-    
-    // Regular emit without ack - just call the existing method
-    emit(event, *args)
-}
-
 /**
  * Extension function to emit event with suspend acknowledgement callback.
  * 
@@ -105,28 +51,14 @@ suspend fun Socket.emitSuspend(event: String, vararg args: Any) {
  * @param ack suspend lambda that will be called when acknowledgement is received (trailing lambda)
  */
 suspend fun Socket.emitSuspend(event: String, vararg args: Any, ack: suspend (Array<out Any>) -> Unit) {
-    // Check reserved events manually (since RESERVED_EVENTS is private)
-    val reservedEvents = setOf(
-        Socket.EVENT_CONNECT,
-        Socket.EVENT_CONNECT_ERROR,
-        Socket.EVENT_DISCONNECT,
-        "disconnecting",
-        "newListener",
-        "removeListener",
-    )
-    if (reservedEvents.contains(event)) {
-        Logging.error(Socket.TAG, "emit reserved event: $event")
-        return
-    }
-    
     // Use suspendCancellableCoroutine to wait for ack
     suspendCancellableCoroutine<Unit> { continuation ->
         val ackWrapper = object : Ack {
-            override fun call(vararg ackArgs: Any) {
+            override fun call(vararg args: Any) {
                 // Use CoroutineScope since socket's scope is private
                 CoroutineScope(Dispatchers.Default).launch {
                     try {
-                        ack(ackArgs)
+                        ack(args)
                         continuation.resume(Unit)
                     } catch (e: Exception) {
                         continuation.resumeWithException(e)

kmp-socketio/src/commonMain/kotlin/com/piasy/kmp/socketio/socketio/Socket.kt

@@ -466,15 +466,10 @@ class Socket(
 
         const val EVENT_MESSAGE = "message"
         const val EVENT_ERROR = Manager.EVENT_ERROR
-
         private val RESERVED_EVENTS = setOf(
             EVENT_CONNECT,
             EVENT_CONNECT_ERROR,
             EVENT_DISCONNECT,
-            // used on the server-side
-            "disconnecting",
-            "newListener",
-            "removeListener",
         )
     }
 }

kmp-socketio/src/commonMain/kotlin/com/piasy/kmp/socketio/socketio/StateFlowExtensions.kt

@@ -3,12 +3,14 @@ package com.piasy.kmp.socketio.socketio
 import com.piasy.kmp.socketio.emitter.Emitter
 import com.piasy.kmp.socketio.engineio.State
 import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.channels.awaitClose
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.asSharedFlow
 import kotlinx.coroutines.flow.asStateFlow
+import kotlinx.coroutines.flow.callbackFlow
 
 /**
  * Extension property to get StateFlow for Manager state.
@@ -79,32 +81,33 @@ val Socket.socketIdFlow: StateFlow<String>
     }
 
 /**
- * Extension function to get Flow for Emitter events.
- * Allows using coroutines to listen to events.
- * 
+ * Extension function to get a cold [Flow] for [Emitter] events.
+ *
+ * Differences from the original listener API:
+ * - integrates with coroutines (cancellation, operators, structured concurrency);
+ * - automatically subscribes/unsubscribes the underlying listener with the Flow lifecycle.
+ *
  * @param event event name
- * @return Flow that emits event arguments
+ * @return cold [Flow] that emits event arguments while it is collected
  */
-fun Emitter.flow(event: String): Flow<Array<out Any>> {
-    val sharedFlow = MutableSharedFlow<Array<out Any>>(
-        replay = 0,
-        extraBufferCapacity = Channel.UNLIMITED
-    )
-    
-    val listener = object : Emitter.Listener {
-        override fun call(vararg args: Any) {
-            sharedFlow.tryEmit(args)
+fun Emitter.flow(event: String): Flow<Array<out Any>> =
+    callbackFlow {
+        val listener = object : Emitter.Listener {
+            override fun call(vararg args: Any) {
+                trySend(args).isSuccess
+            }
+        }
+
+        on(event, listener)
+
+        awaitClose {
+            off(event, listener)
         }
     }
-    
-    on(event, listener)
-    
-    return sharedFlow.asSharedFlow()
-}
 
 /**
  * Extension function to get Flow for Emitter events with replay buffer.
- * 
+ *
  * @param event event name
  * @param replay number of events to replay to new subscribers
  * @return Flow that emits event arguments
@@ -114,14 +117,14 @@ fun Emitter.flowWithReplay(event: String, replay: Int = 1): Flow<Array<out Any>>
         replay = replay,
         extraBufferCapacity = Channel.UNLIMITED
     )
-    
+
     val listener = object : Emitter.Listener {
         override fun call(vararg args: Any) {
             sharedFlow.tryEmit(args)
         }
     }
-    
+
     on(event, listener)
-    
+
     return sharedFlow.asSharedFlow()
 }