arrow-fx-coroutines/arrow.fx.coroutines/executor

executor

suspend fun ResourceScope.executor(timeout: Duration = Duration.INFINITE, closingDispatcher: CoroutineDispatcher = Dispatchers.IO, create: suspend () -> ExecutorService): ExecutorCoroutineDispatcher(source)

Creates a single threaded CoroutineContext as a Resource. Upon release an orderly shutdown of the ExecutorService takes place in which previously submitted tasks are executed, but no new tasks will be accepted.

import arrow.fx.coroutines.executor
import arrow.fx.coroutines.resourceScope
import arrow.fx.coroutines.parTraverse
import java.util.concurrent.Executors
import java.util.concurrent.atomic.AtomicInteger
import kotlin.math.max

suspend fun main(): Unit {
  resourceScope {
    val pool = executor {
      val ctr = AtomicInteger(0)
      val size = max(2, Runtime.getRuntime().availableProcessors())
      Executors.newFixedThreadPool(size) { r ->
        Thread(r, "computation-${ctr.getAndIncrement()}")
          .apply { isDaemon = true }
      }
    }

    listOf(1, 2, 3, 4, 5).parTraverse(pool) { i ->
      println("#$i running on ${Thread.currentThread().name}")
    }
  }
}

fun executor(timeout: Duration = Duration.INFINITE, closingDispatcher: CoroutineDispatcher = Dispatchers.IO, create: suspend () -> ExecutorService): Resource<ExecutorCoroutineDispatcher>(source)