I've written one function with Flow collector which is as shown below,

private fun callSocket(
        eventEmmit: String,
        eventOn: String,
        request: JSONObject
    ): Flow<SocketCallback<JSONObject>> =
        flow {
            try {
                if (socket.connected()) {
                    var response = JSONObject()
                    Log.e("EMIT", JSONObject(Gson().toJson(request)).toString())

                    socket.on(
                        eventOn
                    ) { args ->
                        response = args[0] as JSONObject
                        Log.e("ON", response.toString())
                        **this.emit(SocketCallback.OnSuccess(response))**
                    }.emit(
                        eventEmmit,
                        request
                    )
                    emit(SocketCallback.OnSuccess(response))
                } else {
                    Log.e("SOCKET_ERROR", "Socket connection failed")
                    emit(SocketCallback.OnError("Socket connection failed"))
                }
            } catch (e: SocketException) {
                emit(SocketCallback.OnError(e.toString()))
            }
        }.flowOn(Dispatchers.IO)

But when I write this.emit(SocketCallback.OnSuccess(response))(enclosed in ** in code) in on method it shows me the error "Suspension functions can be called only within coroutine body".

Any solution for this?

Thanks in advance.

CodePudding user response：

You are trying to emit events to flow outside of coroutineScope. socket.on() function probably has signature:

fun on(ev: String, block: (args: String) -> Unit) {

}

in that case, inside lambda block: (args: String) -> Unit) you are outside of scope and you can not invoke suspending functions.

You have only 2 solutions:

1. Every time new event approach - create new coroutine with coroutine builder launch:

socket.on(
    eventOn
) { args ->
    response = args[0] as JSONObject
    Log.e("ON", response.toString())
    launch {
        emit(SocketCallback.OnSuccess(response))
    }
}.emit(
    eventEmmit,
    request
)

2. Use callbackFlow to avoid creation of new coroitines on every event. Please check especially this post.

CodePudding user response：

Here's how I solved it by using the CallBackFlow

private fun callOnSocket(
        eventOn: String
    ): Flow<SocketCallback<JSONObject>> =
        callbackFlow<SocketCallback<JSONObject>> {
            try {
                if (socket.connected()) {
                    Log.e("ON", "Started")
                    var response = JSONObject()
                    socket.on(
                        eventOn
                    ) {
                        response = it[0] as JSONObject
                        Log.e("ON", response.toString())
                        trySend(SocketCallback.OnSuccess(response))
                    }
                } else {
                    Log.e("SOCKET_ERROR", "Socket connection failed")
                    trySend(SocketCallback.OnError("Socket connection failed"))
                }
            } catch (e: SocketException) {
                trySend(SocketCallback.OnError("Socket connection failed"))
            }
            awaitClose { cancel() }
        }.flowOn(Dispatchers.IO)