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

executor

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

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.parMap
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).parMap(pool) { i ->
      println("#$i running on ${Thread.currentThread().name}")
    }
  }
}

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