How could I implement the message queue programmin...

# multiplatform

[Piasy]

How could I implement the message queue programming pattern in mpp?

“message queue programming pattern” means: an object, let’s call it Counter, provides some operations, e.g. increase and decrease, these operations could be called from main thread or other worker threads, but Counter will schedule these calls into its own single threaded message queue, then change its state, e.g. the count, it may also output its state through some callback, and the callback should be invoked on main thread.

On Android, we could using Handler or Executor to implement this pattern easily, I’m thinking if it’s possible to do it on Kotlin/Native. I want to develop a Kotlin multiplatform app and want to use this pattern in it.

I tried worker, but everything submitted to worker must be frozen, so I can’t update Counter ’s state.

I also tried coroutine , but on Kotlin/Native, coroutine only works for main thread, although there is a preview version supporting multi-thread coroutine, but when I try to launch a coroutine on the dispatcher created by newSingleThreadContext("test") , it fails with “This dispatcher can be used only from a single thread test, but now in MainThread”.

[olonho]

in Native every Worker has its own message queue, so https://github.com/JetBrains/kotlin-native/blob/45953f03a505ed2a54185fa5407529ef48466578/runtime/src/main/kotlin/kotlin/native/concurrent/Worker.kt#L104 will help you here

[Piasy]

Thanks for the response, but as I said, when using Worker , the captured objects are frozen, then I can't change it's state inside the lambda posted to Worker .

[Piasy]

Actually what I want to achieve is serializing the calls to Counter from multiple threads, then I don't need to consider concurrency issue any more.

[olonho]

you can update the counter state if it has concurrency safe API, for example if it is an AtomicInt, even being frozen increment operation is still available.

[Piasy]

Oh I'll try that, thanks!

[Piasy]

Hi @olonho, I did some experiments: I encapsulate all states (although it's just count now) of Counter into a immutable data class called CounterState , and define a class TaskQueue which holds a CounterState :

expect class TaskQueue<T>(state: T) {
    fun post(task: () -> Unit)

    fun getState(): T

    fun setState(newState: T)
}

Android actual class is:

actual class TaskQueue<T> actual constructor(private var state: T) {
  private val executor = Executors.newSingleThreadExecutor()

  actual fun post(task: () -> Unit) {
    executor.execute(task)
  }

  actual fun getState(): T = state

  actual fun setState(newState: T) {
    state = newState
  }
}

iOS actual class is:

actual class TaskQueue<T> actual constructor(state: T) {
  private val worker = Worker.start()
  private val state = AtomicReference(state.freeze())

  actual fun post(task: () -> Unit) {
    worker.executeAfter(0, task.freeze())
  }

  actual fun getState(): T = state.value

  actual fun setState(newState: T) {
    newState.freeze()
    while (!state.compareAndSet(state.value, newState)) {
    }
  }
}

and the increment fun of Counter is (in common module):

fun increment() {
    <http://taskQueue.post|taskQueue.post> {
      taskQueue.setState(CounterState(taskQueue.getState().count + 1))
    }
  }

It works as expected. How does it look like to you?