fix of #160

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ## 0.12.13
 
+* `Coroutines`:
+  * Add opportunity to use markers in actors (solution of [#160](https://github.com/InsanusMokrassar/MicroUtils/issues/160))
 * `Koin`:
   * Module inited :)
 

coroutines/src/commonMain/kotlin/dev/inmo/micro_utils/coroutines/Actor.kt

@@ -1,28 +1,27 @@
 package dev.inmo.micro_utils.coroutines
 
-import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.Channel
-import kotlinx.coroutines.launch
+import kotlinx.coroutines.flow.consumeAsFlow
 
 fun <T> CoroutineScope.actor(
     channelCapacity: Int = Channel.UNLIMITED,
+    markerFactory: suspend (T) -> Any? = { null },
     block: suspend (T) -> Unit
 ): Channel<T> {
     val channel = Channel<T>(channelCapacity)
-    launch {
-        for (data in channel) {
-            block(data)
-        }
-    }
+    channel.consumeAsFlow().subscribeAsync(this, markerFactory, block)
     return channel
 }
 
 inline fun <T> CoroutineScope.safeActor(
     channelCapacity: Int = Channel.UNLIMITED,
     noinline onException: ExceptionHandler<Unit> = defaultSafelyExceptionHandler,
+    noinline markerFactory: suspend (T) -> Any? = { null },
     crossinline block: suspend (T) -> Unit
 ): Channel<T> = actor(
-    channelCapacity
+    channelCapacity,
+    markerFactory
 ) {
     safely(onException) {
         block(it)