import kotlinx.coroutines.experimental.channels.*
import kotlinx.coroutines.experimental.delay
import kotlinx.coroutines.experimental.launch
import kotlinx.coroutines.experimental.runBlocking

fun main(args: Array<String>) = runBlocking<Unit> {
    // Create the listener pool
    val listener = ConflatedBroadcastChannel<Int>()

    // Create the actor
    val actor = counterActor(listener)

    // Observe the counter
    val printJob = launch {
        listener.consumeEach {
            println("The counter is $it")
        }
    }

    // Send some events
    launch{
        repeat(5) {
            actor.send(Event.Increment)
            delay(250)
        }
        repeat(5) {
            actor.send(Event.Decrement)
            delay(250)
        }
        actor.close()
    }

    printJob.join()
}

fun counterActor(listener: SendChannel<Int>) = actor<Event> {
    var counter = 0
    listener.offer(counter)
    for (msg in channel) {
        counter = when (msg) {
            is Event.Increment -> counter + 1
            is Event.Decrement -> counter - 1
        }.also {
            listener.offer(it)
        }
    }
    listener.close()
}

sealed class Event {
    object Increment: Event()
    object Decrement: Event()
}