How to use concurrent class of io.kotest.matchers.concurrent package

1package arrow2import arrow.core.Either3import arrow.core.identity4import arrow.core.right5import arrow.fx.coroutines.ExitCase6import arrow.fx.coroutines.guaranteeCase7import arrow.fx.coroutines.never8import io.kotest.assertions.fail9import io.kotest.core.spec.style.StringSpec10import io.kotest.matchers.collections.shouldBeIn11import io.kotest.matchers.nulls.shouldNotBeNull12import io.kotest.matchers.shouldBe13import io.kotest.matchers.types.shouldBeTypeOf14import io.kotest.property.Arb15import io.kotest.property.arbitrary.int16import io.kotest.property.arbitrary.string17import io.kotest.property.checkAll18import kotlin.time.Duration19import kotlin.time.ExperimentalTime20import kotlin.time.seconds21import kotlinx.coroutines.CompletableDeferred22import kotlinx.coroutines.CoroutineScope23import kotlinx.coroutines.Deferred24import kotlinx.coroutines.Job25import kotlinx.coroutines.async26import kotlinx.coroutines.awaitAll27import kotlinx.coroutines.coroutineScope28import kotlinx.coroutines.flow.flow29import kotlinx.coroutines.launch30import kotlinx.coroutines.suspendCancellableCoroutine31import kotlinx.coroutines.withTimeout32@OptIn(ExperimentalTime::class)33class StructuredConcurrencySpec :34  StringSpec({35    "async - suspendCancellableCoroutine.invokeOnCancellation is called with Shifted Continuation" {36      val started = CompletableDeferred<Unit>()37      val cancelled = CompletableDeferred<Throwable?>()38      cont<String, Nothing> {39          coroutineScope {40            val never = async {41              suspendCancellableCoroutine<Nothing> { cont ->42                cont.invokeOnCancellation { cause ->43                  require(cancelled.complete(cause)) { "cancelled latch was completed twice" }44                }45                require(started.complete(Unit))46              }47            }48            async<Int> {49                started.await()50                shift("hello")51              }52              .await()53            never.await()54          }55        }56        .runCont() shouldBe "hello"57      withTimeout(Duration.seconds(2)) {58        cancelled.await().shouldNotBeNull().message shouldBe "Shifted Continuation"59      }60    }61    "Computation blocks run on parent context" {62      val parentCtx = currentContext()63      cont<Nothing, Unit> { currentContext() shouldBe parentCtx }.runCont()64    }65    "Concurrent shift - async await" {66      checkAll(Arb.int(), Arb.int()) { a, b ->67        cont<Int, String> {68            coroutineScope {69              val fa = async<String> { shift(a) }70              val fb = async<String> { shift(b) }71              fa.await() + fb.await()72            }73          }74          .runCont() shouldBeIn listOf(a, b)75      }76    }77    "Concurrent shift - async await exit results" {78      checkAll(Arb.int()) { a ->79        val scopeExit = CompletableDeferred<ExitCase>()80        val fbExit = CompletableDeferred<ExitCase>()81        val startLatches = (0..11).map { CompletableDeferred<Unit>() }82        val nestedExits = (0..10).map { CompletableDeferred<ExitCase>() }83        fun CoroutineScope.asyncTask(84          start: CompletableDeferred<Unit>,85          exit: CompletableDeferred<ExitCase>86        ): Deferred<Unit> = async {87          guaranteeCase({88            start.complete(Unit)89            never<Unit>()90          }) { case -> require(exit.complete(case)) }91        }92        cont<Int, String> {93            guaranteeCase({94              coroutineScope {95                val fa =96                  async<Unit> {97                    startLatches.drop(1).zip(nestedExits) { start, promise ->98                      asyncTask(start, promise)99                    }100                    startLatches.awaitAll()101                    shift(a)102                  }103                val fb = asyncTask(startLatches.first(), fbExit)104                fa.await()105                fb.await()106              }107            }) { case -> require(scopeExit.complete(case)) }108            fail("Should never come here")109          }110          .runCont() shouldBe a111        withTimeout(2.seconds) {112          scopeExit.await().shouldBeTypeOf<ExitCase.Cancelled>()113          fbExit.await().shouldBeTypeOf<ExitCase.Cancelled>()114          nestedExits.awaitAll().forEach { it.shouldBeTypeOf<ExitCase.Cancelled>() }115        }116      }117    }118    "Concurrent shift - async" {119      checkAll(Arb.int(), Arb.int()) { a, b ->120        cont<Int, String> {121            coroutineScope {122              val fa = async<Nothing> { shift(a) }123              val fb = async<Nothing> { shift(b) }124              "I will be overwritten by shift - coroutineScope waits until all async are finished"125            }126          }127          .fold({ fail("Async is never awaited, and thus ignored.") }, ::identity) shouldBe128          "I will be overwritten by shift - coroutineScope waits until all async are finished"129      }130    }131    "Concurrent shift - async exit results" {132      checkAll(Arb.int(), Arb.string()) { a, str ->133        val exitScope = CompletableDeferred<ExitCase>()134        val startLatches = (0..10).map { CompletableDeferred<Unit>() }135        val nestedExits = (0..10).map { CompletableDeferred<ExitCase>() }136        fun CoroutineScope.asyncTask(137          start: CompletableDeferred<Unit>,138          exit: CompletableDeferred<ExitCase>139        ): Deferred<Unit> = async {140          guaranteeCase({141            start.complete(Unit)142            never<Unit>()143          }) { case -> require(exit.complete(case)) }144        }145        cont<Int, String> {146            guaranteeCase({147              coroutineScope {148                val fa =149                  async<Unit> {150                    startLatches.zip(nestedExits) { start, promise -> asyncTask(start, promise) }151                    startLatches.awaitAll()152                    shift(a)153                  }154                str155              }156            }) { case -> require(exitScope.complete(case)) }157          }158          .runCont() shouldBe str159        withTimeout(2.seconds) {160          nestedExits.awaitAll().forEach { it.shouldBeTypeOf<ExitCase.Cancelled>() }161        }162      }163    }164    "Concurrent shift - launch" {165      checkAll(Arb.int(), Arb.int()) { a, b ->166        cont<Int, String> {167            coroutineScope {168              launch { shift(a) }169              launch { shift(b) }170              "shift does not escape `launch`"171            }172          }173          .runCont() shouldBe "shift does not escape `launch`"174      }175    }176    "Concurrent shift - launch exit results" {177      checkAll(Arb.int(), Arb.string()) { a, str ->178        val scopeExit = CompletableDeferred<ExitCase>()179        val startLatches = (0..10).map { CompletableDeferred<Unit>() }180        val nestedExits = (0..10).map { CompletableDeferred<ExitCase>() }181        fun CoroutineScope.launchTask(182          start: CompletableDeferred<Unit>,183          exit: CompletableDeferred<ExitCase>184        ): Job = launch {185          guaranteeCase({186            start.complete(Unit)187            never<Unit>()188          }) { case -> require(exit.complete(case)) }189        }190        cont<Int, String> {191            guaranteeCase({192              coroutineScope {193                val fa = launch {194                  startLatches.zip(nestedExits) { start, promise -> launchTask(start, promise) }195                  startLatches.awaitAll()196                  shift(a)197                }198                str199              }200            }) { case -> require(scopeExit.complete(case)) }201          }202          .runCont() shouldBe str203        withTimeout(2.seconds) {204          scopeExit.await().shouldBeTypeOf<ExitCase.Completed>()205          nestedExits.awaitAll().forEach { it.shouldBeTypeOf<ExitCase.Cancelled>() }206        }207      }208    }209    // `shift` escapes `cont` block, and gets rethrown inside `coroutineScope`.210    // Effectively awaiting/executing DSL code, outside of the DSL...211    "async funky scenario #1 - Extract `shift` from `cont` through `async`" {212      checkAll(Arb.int(), Arb.int()) { a, b ->213        runCatching {214            coroutineScope {215              val shiftedAsync =216                cont<Int, Deferred<String>> {217                    val fa = async<Int> { shift(a) }218                    async { shift(b) }219                  }220                  .fold({ fail("shift was never awaited, so it never took effect") }, ::identity)221              shiftedAsync.await()222            }223          }224          .exceptionOrNull()225          ?.message shouldBe "Shifted Continuation"226      }227    }228  })...

...9import io.ktor.server.testing.handleRequest10import io.ktor.server.testing.withTestApplication11import kotlinx.coroutines.*12import java.util.*13import java.util.concurrent.ConcurrentHashMap14import kotlin.random.Random15class ApplicationTest : StringSpec({16    "/" {17        withTestApplication({ module() }) {18            handleRequest(HttpMethod.Get, "/").apply {19                response.status() shouldBe HttpStatusCode.OK20                response.content shouldBe "root"21            }22        }23    }24    "ws/echo" {25        withTestApplication({ module() }) {26            handleWebSocketConversation("/ws/echo") { incoming, outgoing ->27                val textMessages = listOf("111", "222")...

...75    }76    @Nested77    inner class Async {78        @Test79        fun `add entities concurrently`() = runTest {80            clearEngine()81            val arc = engine.getArchetype(GearyType(ulongArrayOf(componentId<String>() or HOLDS_DATA)))82            concurrentOperation(10000) {83                arc.addEntityWithData(engine.newEntity().getRecord(), arrayOf("Test"))84            }.awaitAll()85            arc.ids.size shouldBe 1000086            arc.ids.shouldBeUnique()87        }88    }89    // The two tests below are pretty beefy and more like benchmarks so they're disabled by default90//    @Test91    fun `set and remove concurrency`() = runTest {92        println(measureTime {93            concurrentOperation(100) {94                val entity = entity()95                repeat(1000) { id ->96                    launch {97//                        entity.withLock {98                        entity.setRelation(id.toULong(), "String")99                        println("Locked for ${entity.id}: $id, size ${engine.archetypeCount}")100//                        }101                    }102                }103            }.awaitAll()104        })105//        entity.getComponents().shouldBeEmpty()106    }107    //    @Test108//    fun `mutliple locks`() {109//        val a = entity()110////        val b = entity()111//        concurrentOperation(10000) {112//            engine.withLock(setOf(a/*, b*/)) {113//                println("Locking")114//                delay(100)115//            }116//        }117//    }118    //    @Test119    fun `concurrent archetype creation`() = runTest {120        clearEngine()121        val iters = 10000122        println(measureTime {123            for (i in 0 until iters) {124//            concurrentOperation(iters) { i ->125                engine.getArchetype(GearyType((0uL..i.toULong()).toList()))126                println("Creating arc $i, total: ${engine.archetypeCount}")127//            }.awaitAll()128            }129        })130        engine.archetypeCount shouldBe iters + 1131    }132}...

...16import org.junit.jupiter.api.Timeout17import java.time.Duration18import java.time.Duration.ofMillis19import java.util.*20import java.util.concurrent.ConcurrentLinkedQueue21import java.util.concurrent.TimeUnit.SECONDS22import java.util.concurrent.atomic.AtomicBoolean23class ProducerActorTest : KafkaSuite {24    override val kafkaCluster = createDockerKafkaCluster()25    @Test26    fun `metrics collection is accurate`() {27        val testTopic = UUID.randomUUID().toString()28        val generatedMessageCount = 2029        val testMessages = generateStringRecords(testTopic, generatedMessageCount)30        val metricsQueue = ConcurrentLinkedQueue<Metrics<String, String>>()31        val producer = ProducerActor(32                id = "firstProducer",33                kafkaProducer = kafkaCluster.createStringProducer(),34                records = testMessages,35                rate = 0L,36                metricsQueue = metricsQueue,...

...7import io.kotest.data.headers8import io.kotest.data.row9import io.kotest.data.table10import io.kotest.inspectors.forAll11import io.kotest.matchers.concurrent.shouldCompleteWithin12import io.kotest.matchers.shouldBe13import kotlinx.coroutines.GlobalScope14import kotlinx.coroutines.coroutineScope15import kotlinx.coroutines.runBlocking16import java.util.concurrent.TimeUnit17import amber.collections.SyncMutableList as SyncList18class `synchronized list concurrent write and read - integration` : AnnotationSpec() {19    private suspend fun testWith(it: Int, b: SyncMode) {20        val factor = it21        val list = SyncList<String>(synchronized = Synchronized(b))22        coroutineScope<Unit> {23            this.joinAllJobs {24                repeat(factor) {25                    GlobalScope.launch {26                        repeat(factor) {27                            list.add("test")28                        }29                    }30                }31            }32        }...

...3import io.kotest.matchers.booleans.shouldBeFalse4import io.kotest.matchers.collections.shouldBeEmpty5import io.kotest.matchers.collections.shouldContain6import io.kotest.matchers.collections.shouldContainInOrder7import java.util.concurrent.Semaphore8import kotlin.concurrent.thread9import kotlin.time.DurationUnit10import kotlin.time.ExperimentalTime11import kotlin.time.toDuration12private class TestActionForActions13@ExperimentalTime14internal class ActionsTests : BehaviorSpec({15    given("A empty actions") {16        val actions = Actions<TestActionForActions>()17        `when`("I try to read it") {18            then("It should be empty") {19                actions.shouldBeEmpty()20            }21        }22        `when`("I add one action") {...

1package io.perfometer.statistics2import io.kotest.matchers.nulls.shouldNotBeNull3import io.kotest.matchers.shouldBe4import io.perfometer.http.HttpMethod5import io.perfometer.http.HttpStatus6import java.time.Instant7import kotlin.test.Test8@Suppress("FunctionName")9class ConcurrentQueueStatisticsCollectorSpecification {10    @Test11    fun `gather should add statistics to the summary`() {12        val scenarioStatistics = ConcurrentQueueStatisticsCollector()13        scenarioStatistics.start(Instant.ofEpochSecond(0))14        scenarioStatistics.gather(StatisticsFixture.singleGetRequestStatistics())15        val postStatistics = RequestStatistics("POST /", HttpMethod.POST, "/", Instant.ofEpochSecond(3), Instant.ofEpochSecond(4), HttpStatus(201))16        scenarioStatistics.gather(postStatistics)17        val scenarioSummary = scenarioStatistics.finish(Instant.ofEpochSecond(5))18        scenarioSummary.totalSummary.shouldNotBeNull()19                .requestCount shouldBe 220        scenarioSummary.summaries.size shouldBe 221    }22    @Test(IllegalStateException::class)23    fun `gather should not allow adding new statistics when the scenario was not started`() {24        val scenarioStatistics = ConcurrentQueueStatisticsCollector()25        scenarioStatistics.gather(StatisticsFixture.singleGetRequestStatistics())26    }27    @Test(IllegalStateException::class)28    fun `gather should not allow adding new statistics when the scenario has ended`() {29        val scenarioStatistics = ConcurrentQueueStatisticsCollector()30        scenarioStatistics.start(Instant.ofEpochSecond(0))31        scenarioStatistics.gather(StatisticsFixture.singleGetRequestStatistics())32        scenarioStatistics.finish(Instant.ofEpochSecond(5))33        scenarioStatistics.gather(StatisticsFixture.singleGetRequestStatistics())34    }35    @Test(IllegalArgumentException::class)36    fun `gather should not allow finish with time before start time`() {37        val scenarioStatistics = ConcurrentQueueStatisticsCollector()38        scenarioStatistics.start(Instant.ofEpochSecond(5))39        scenarioStatistics.gather(40            StatisticsFixture.singleGetRequestStatistics(41                Instant.ofEpochSecond(6),42                Instant.ofEpochSecond(7),43            )44        )45        scenarioStatistics.finish(Instant.ofEpochSecond(1))46    }47}...

...7import locutus.net.messages.Message.Testing.BarMessage8import locutus.net.messages.Message.Testing.FooMessage9import locutus.net.messages.MessageRouter.*10import java.net.*11import java.util.concurrent.*12import kotlin.time.*13@ExperimentalTime14@Order(0)15class MessageRouterSpec : FunSpec({16    context("Given a MessageRouter and an extractor for FooMessage") {17        val messageRouter = MessageRouter()18        val fooExtractor = Extractor<FooMessage, Int>("fooExtractor") { message.v }19        val fooReceived = ConcurrentLinkedQueue<SenderMessage<FooMessage>>()20        context("Create listener for FooMessage(1) that will cancel after initial message is received") {21            println("A")22            messageRouter.listen(fooExtractor, 1, NEVER) { sender, msg ->23                fooReceived += SenderMessage(sender, msg)24            }25            println("B")...

1    import io.kotest.matchers.concurrent.shouldBeRunning2    import io.kotest.matchers.concurrent.shouldCompleteWithin3    import io.kotest.matchers.concurrent.shouldNotCompleteWithin4    import io.kotest.matchers.concurrent.shouldNotBeRunning5    import io.kotest.matchers.concurrent.shouldNotThrowAnyException6    import io.kotest.matchers.concurrent.shouldThrowAnyException7    import io.kotest.matchers.concurrent.shouldThrowException8    import io.kotest.matchers.concurrent.shouldThrowExceptionOfType9    import io.kotest.matchers.concurrent.shouldThrowInstanceOf10    import kotlinx.coroutines.delay11    import kotlinx.coroutines.runBlocking12    import org.junit.jupiter.api.Test13    import java.io.IOException14    import java.time.Duration15    class ConcurrentMatchersTest {16        fun `should be running`() {17            val thread = Thread { Thread.sleep(1000) }18            thread.start()19            thread.shouldBeRunning()20        }21        fun `should not be running`() {22            val thread = Thread { Thread.sleep(1000) }23            thread.shouldNotBeRunning()24        }25        fun `should complete within`() {26            val thread = Thread { Thread.sleep(1000) }27            thread.start()28            thread.shouldCompleteWithin(Duration.ofSeconds(2))29        }30        fun `should not complete within`() {31            val thread = Thread { Thread.sleep(1000) }32            thread.start()33            thread.shouldNotCompleteWithin(Duration.ofMillis(500))34        }35        fun `should throw any exception`() {36            val thread = Thread { throw IOException() }37            thread.start()38            thread.shouldThrowAnyException()39        }40        fun `should not throw any exception`() {41            val thread = Thread { Thread.sleep(1000) }42            thread.start()43            thread.shouldNotThrowAnyException()44        }45        fun `should throw exception`() {46            val thread = Thread { throw IOException() }47            thread.start()48            thread.shouldThrowException(IOException::class)49        }50        fun `should throw exception of type`() {51            val thread = Thread { throw IOException() }52            thread.start()53            thread.shouldThrowExceptionOfType(IOException::class)54        }55        fun `should throw instance of`() {

1import io.kotest.matchers.concurrent.shouldBeCompleted2import kotlinx.coroutines.delay3import kotlinx.coroutines.runBlocking4import org.junit.jupiter.api.Test5class ExampleTest {6    fun shouldReturnCompleted() {7        runBlocking {8        }9    }10}11import io.kotest.matchers.string.shouldContain12import org.junit.jupiter.api.Test13class ExampleTest {14    fun shouldReturnContains() {15    }16}17import io.kotest.matchers.throwable.shouldHaveCause18import org.junit.jupiter.api.Test19class ExampleTest {20    fun shouldReturnCause() {21        val exception = RuntimeException(RuntimeException("Hello World"))22    }23}24import io.kotest.matchers.types.shouldBeTypeOf25import org.junit.jupiter.api.Test26class ExampleTest {27    fun shouldReturnType() {28    }29}30import io.kotest.matchers.collections.shouldContain31import org.junit.jupiter.api.Test32class ExampleTest {33    fun shouldReturnCollection() {34        val list = listOf("Hello", "World")35    }36}37import io.kotest.matchers.doubles.shouldBeGreaterThan38import org.junit.jupiter.api.Test39class ExampleTest {40    fun shouldReturnDouble() {41    }42}43import io.kotest.matchers.floats.shouldBeGreaterThan44import org.junit.jupiter.api.Test45class ExampleTest {

1val future = async { 1 + 1 }2future should completeWithin(1.second)3val future = async { 1 + 1 }4future should completeWithin(1.second) and return 25val future = async { 1 + 1 }6future should completeWithin(1.second) and return 27val future = async { 1 + 1 }8future should completeWithin(1.second) and return 29val future = async { 1 + 1 }10future should completeWithin(1.second) and return 211val future = async { 1 + 1 }12future should completeWithin(1.second) and return 213val future = async { 1 + 1 }14future should completeWithin(1.second) and return 215val future = async { 1 + 1 }16future should completeWithin(1.second) and return 217val future = async { 1 + 1 }18future should completeWithin(1.second) and return 2

1class ConcurrentTest : FunSpec({2test("concurrent test") {3}4})5class ConcurrentTest : DescribeSpec({6describe("concurrent test") {7}8})9class ConcurrentTest : BehaviorSpec({10Given("concurrent test") {11}12})13class ConcurrentTest : FeatureSpec({14feature("concurrent test") {15}16})17class ConcurrentTest : FreeSpec({18"concurrent test" {19}20})21class ConcurrentTest : ExpectSpec({22expect("concurrent test") {23}24})25class ConcurrentTest : WordSpec({26"concurrent test" should {27}28})29class ConcurrentTest : ShouldSpec({30"concurrent test" {31}32})33class ConcurrentTest : StringSpec({34"concurrent test" {35}36})