8 changed files with 0 additions and 460 deletions
--- a/settings.gradle.kts
+++ b/settings.gradle.kts
@ -52,7 +52,6 @@ dependencyResolutionManagement {
 }
 include(
    ":simulation-kt",
    ":controls-core",
    ":controls-ports-ktor",
    ":controls-serial",
--- a/simulation-kt/build.gradle.kts
+++ b/simulation-kt/build.gradle.kts
@ -1,32 +0,0 @@
 import space.kscience.gradle.Maturity
 plugins {
    id("space.kscience.gradle.mpp")
    `maven-publish`
 }
 description = """
    Core interfaces for building a device server
 """.trimIndent()
 kscience {
    jvm()
    js()
    native()
    wasm()
    useCoroutines()
    useContextReceivers()
    commonMain {
        api(spclibs.kotlinx.datetime)
    }
    jvmTest{
        implementation(spclibs.logback.classic)
    }
 }
 readme{
    maturity = Maturity.EXPERIMENTAL
 }
--- a/simulation-kt/src/commonMain/kotlin/GeneratingTimeline.kt
+++ b/simulation-kt/src/commonMain/kotlin/GeneratingTimeline.kt
@ -1,132 +0,0 @@
 package space.kscience.simulation
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.channels.BufferOverflow
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.flow.*
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
 import kotlinx.datetime.Instant
 import kotlin.time.Duration
 /**
 * @param lookaheadInterval an interval for generated events ahead of the last observed event.
 */
 public class GeneratingTimeline<E : TimelineEvent>(
    private val generationScope: CoroutineScope,
    private val initialEvent: E,
    private val lookaheadInterval: Duration,
    private val generatorChain: suspend (E) -> E
 ) : Timeline<E>, AutoCloseable {
    // push to this channel to trigger event generation
    private val wakeupChannel = Channel<Unit>(onBufferOverflow = BufferOverflow.DROP_OLDEST)
    private suspend fun kickGenerator() {
        wakeupChannel.send(Unit)
    }
    private val mutex = Mutex()
    private val history = ArrayDeque<E>()
    private val lastEvent = MutableSharedFlow<E>(replay = Int.MAX_VALUE)
    private val updateHistoryJob = generationScope.launch {
        lastEvent.onEach {
            mutex.withLock {
                history.add(it)
                //cleanup old events
                val threshold = observedTime ?: return@withLock
                while (history.isNotEmpty() && history.last().time > threshold) {
                    history.removeFirst()
                }
            }
        }
    }
    private val observers: MutableSet<TimelineObserver> = mutableSetOf()
    override val time: Instant
        get() = history.lastOrNull()?.time ?: initialEvent.time
    override val observedTime: Instant?
        get() = observers.minOfNotNullOrNull { it.time }
    override fun flowUnobservedEvents(): Flow<E> = flow {
        history.forEach { e ->
            emit(e)
        }
        emitAll(lastEvent)
    }
    override suspend fun advance(toTime: Instant) {
        observers.forEach {
            it.collect(toTime)
        }
    }
    private var generatorJob: Job = launchGenerator(initialEvent)
    private fun launchGenerator(event: E): Job = generationScope.launch {
        kickGenerator()
        var currentEvent = event
        // for each wakeup generate all events in lookaheadInterval
        for (u in wakeupChannel) {
            while (currentEvent.time < (observedTime ?: event.time) + lookaheadInterval) {
                val nextEvent = generatorChain(currentEvent)
                lastEvent.emit(nextEvent)
                currentEvent = nextEvent
            }
        }
    }
    public suspend fun interrupt(newStart: E) {
        check(newStart.time > (observedTime ?: Instant.DISTANT_FUTURE)) {
            "Can't interrupt generating timeline after observed event"
        }
        mutex.withLock {
            while (history.isNotEmpty() && history.last().time > newStart.time) {
                history.removeLast()
            }
            generatorJob.cancel()
            generatorJob = launchGenerator(newStart)
        }
        kickGenerator()
    }
    override fun close() {
        updateHistoryJob.cancel()
        generatorJob.cancel()
    }
    override suspend fun observe(collector: suspend Flow<E>.() -> Unit): TimelineObserver {
        val observer = object : TimelineObserver {
            override var time: Instant = this@GeneratingTimeline.time
            override suspend fun collect(upTo: Instant) {
                flowUnobservedEvents().takeWhile {
                    it.time <= upTo
                }.onEach {
                    time = it.time
                    kickGenerator()
                }.collector()
            }
            override fun close() {
                observers.remove(this)
                if(observers.isEmpty()){
                    this@GeneratingTimeline.close()
                }
            }
        }
        observers.add(observer)
        return observer
    }
 }
--- a/simulation-kt/src/commonMain/kotlin/MergedTimeline.kt
+++ b/simulation-kt/src/commonMain/kotlin/MergedTimeline.kt
@ -1,51 +0,0 @@
 package space.kscience.simulation
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.merge
 import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.flow.takeWhile
 import kotlinx.datetime.Instant
 public class MergedTimeline<E : TimelineEvent>(
    private val timelines: List<Timeline<E>>
 ) : Timeline<E> {
    override val time: Instant
        get() = timelines.minOfNotNullOrNull { it.time } ?: Instant.DISTANT_PAST
    override val observedTime: Instant?
        get() = timelines.maxOfNotNullOrNull { it.observedTime }
    override fun flowUnobservedEvents(): Flow<E> = timelines.map { flowUnobservedEvents() }.merge()
    override suspend fun advance(toTime: Instant) {
        timelines.forEach { it.advance(toTime) }
    }
 //    override suspend fun interrupt(atTime: Instant) {
 //        timelines.forEach { it.interrupt(atTime) }
 //    }
    private val observers: MutableSet<TimelineObserver> = mutableSetOf()
    override suspend fun observe(collector: suspend Flow<E>.() -> Unit): TimelineObserver {
        val observer = object : TimelineObserver {
            override var time: Instant = this@MergedTimeline.time
            override suspend fun collect(upTo: Instant) = timelines
                .map { flowUnobservedEvents() }
                .merge()
                .takeWhile { it.time <= upTo }.onEach {
                    time = it.time
                }.collector()
            override fun close() {
                observers.remove(this)
            }
        }
        observers.add(observer)
        return observer
    }
 }
--- a/simulation-kt/src/commonMain/kotlin/SharedTimeline.kt
+++ b/simulation-kt/src/commonMain/kotlin/SharedTimeline.kt
@ -1,88 +0,0 @@
 package space.kscience.simulation
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.asFlow
 import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.flow.takeWhile
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
 import kotlinx.datetime.Instant
 /**
 * A manually mutable [Timeline] that could be modified via [emit] method by multiple
 */
 public class SharedTimeline<E : TimelineEvent> : Timeline<E> {
    private val mutex = Mutex()
    private val events = ArrayDeque<E>()
    private val observers: MutableSet<TimelineObserver> = mutableSetOf()
    override val time: Instant
        get() = events.lastOrNull()?.time ?: Instant.DISTANT_PAST
    override val observedTime: Instant?
        get() = observers.minOfNotNullOrNull { it.time }
    override fun flowUnobservedEvents(): Flow<E> = events.asFlow()
    /**
     * Emit new event to the timeline
     */
    public suspend fun emit(event: E): Boolean = mutex.withLock {
        if (event.time < (observedTime ?: Instant.DISTANT_PAST)) {
            error("Can't emit event $event because there are observed events after $observedTime")
        }
        events.add(event)
    }
    override suspend fun advance(toTime: Instant) {
        observers.forEach {
            it.collect(toTime)
        }
    }
    /**
     * Discard all events before [observedTime]
     */
    private suspend fun cleanup(): Unit = mutex.withLock {
        val threshold = observedTime ?: return@withLock
        while (events.isNotEmpty() && events.last().time > threshold) {
            events.removeFirst()
        }
    }
 //    /**
 //     * Discard unconsumed events after [atTime].
 //     */
 //    override suspend fun interrupt(atTime: Instant): Unit = mutex.withLock {
 //        val threshold = observedTime
 //        if (atTime < threshold)
 //            error("Timeline interrupt at time $atTime is not possible because there are observed events before $threshold")
 //        while (events.isNotEmpty() && events.last().time > atTime) {
 //            events.removeLast()
 //        }
 //    }
    override suspend fun observe(collector: suspend Flow<E>.() -> Unit): TimelineObserver {
        val observer = object : TimelineObserver {
            val observerMutex = Mutex()
            override var time: Instant = this@SharedTimeline.time
            override suspend fun collect(upTo: Instant) = observerMutex.withLock {
                flowUnobservedEvents().takeWhile { it.time <= upTo }.onEach {
                    time = it.time
                }.collector()
                cleanup()
            }
            override fun close() {
                observers.remove(this)
            }
        }
        observers.add(observer)
        return observer
    }
 }
--- a/simulation-kt/src/commonMain/kotlin/Timeline.kt
+++ b/simulation-kt/src/commonMain/kotlin/Timeline.kt
@ -1,87 +0,0 @@
 package space.kscience.simulation
 import kotlinx.coroutines.flow.Flow
 import kotlinx.datetime.Instant
 import kotlin.time.Duration
 public interface TimelineEvent {
    public val time: Instant
 }
 public interface TimelineInterval : TimelineEvent {
    public val startTime: Instant
    public val duration: Duration
    override val time: Instant
        get() = startTime + duration
 }
 public data class SimpleTimelineEvent<T>(override val time: Instant, val value: T) : TimelineEvent
 public interface TimelineObserver : AutoCloseable {
    /**
     * The subjective time of this observer
     */
    public val time: Instant
    /**
     * Collect all uncollected events from [time] to [upTo].
     *
     * By default, collects all events.
     */
    public suspend fun collect(upTo: Instant = Instant.DISTANT_FUTURE)
 }
 /**
 * A time-ordered sequence of events of type [E]. There time of events is strictly monotonic, meaning that the time of
 * the next event is greater than the previous event time.
 *
 * Timeline guarantees that all collectors could read all events when they need. Meaning that all unread events are cached.
 *
 * Timeline guarantees that already read events won't change, but unread events could change.
 */
 public interface Timeline<E : TimelineEvent> {
    /**
     * A subjective time of this timeline. The time could advance without events being produced.
     */
    public val time: Instant
    /**
     * The time of the last event that was observed by all observers
     */
    public val observedTime: Instant?
    /**
     * Flow events from [observedTime] to [time].
     *
     * The resulting flow is finite and should not suspend.
     *
     * This method does not affect [observedTime].
     */
    public fun flowUnobservedEvents(): Flow<E>
    /**
     * Attach observer to this [Timeline]. The observer collection is not triggered right away, but only on demand.
     *
     * Each collection shifts [TimelineObserver.time] for this observer.
     * The value of [observedTime] is the least of all observers [TimelineObserver.time].
     */
    public suspend fun observe(
        collector: suspend Flow<E>.() -> Unit
    ): TimelineObserver
    /**
     * Advance simulation time to [toTime]. This method forces all observers to collect all events in the given range.
     *
     * This method suspends until all advancement is done
     */
    public suspend fun advance(toTime: Instant)
 //    /**
 //     * Interrupt generation of this timeline and discard unconsumed events after [atTime].
 //     *
 //     * Throw exception if at least one observer advanced
 //     */
 //    public suspend fun interrupt(atTime: Instant): Unit
 }
--- a/simulation-kt/src/commonMain/kotlin/notNullUtils.kt
+++ b/simulation-kt/src/commonMain/kotlin/notNullUtils.kt
@ -1,35 +0,0 @@
 package space.kscience.simulation
 internal inline fun <T, R : Comparable<R>> Iterable<T>.minOfNotNullOrNull(selector: (T) -> R?): R? {
    val iterator = iterator()
    if (!iterator.hasNext()) return null
    var minValue = selector(iterator.next())
    while (iterator.hasNext()) {
        val v = selector(iterator.next())
        when {
            minValue == null -> minValue = v
            v == null -> {/*do nothing*/}
            minValue > v -> {
                minValue = v
            }
        }
    }
    return minValue
 }
 internal inline fun <T, R : Comparable<R>> Iterable<T>.maxOfNotNullOrNull(selector: (T) -> R?): R? {
    val iterator = iterator()
    if (!iterator.hasNext()) return null
    var maxValue = selector(iterator.next())
    while (iterator.hasNext()) {
        val v = selector(iterator.next())
        when {
            maxValue == null -> maxValue = v
            v == null -> {/*do nothing*/}
            maxValue < v -> {
                maxValue = v
            }
        }
    }
    return maxValue
 }
--- a/simulation-kt/src/commonTest/kotlin/TimelineTests.kt
+++ b/simulation-kt/src/commonTest/kotlin/TimelineTests.kt
@ -1,34 +0,0 @@
 package space.kscience.simulation
 import kotlinx.coroutines.test.runTest
 import kotlinx.datetime.Instant
 import kotlin.test.Test
 import kotlin.time.Duration.Companion.seconds
 class TimelineTests {
    @Test
    fun testGeneration() = runTest(timeout = 5.seconds) {
        val startTime = Instant.parse("2020-01-01T00:00:00.000Z")
        val generation = GeneratingTimeline<SimpleTimelineEvent<DoubleArray>>(
            this,
            initialEvent = SimpleTimelineEvent(startTime, List(10) { it.toDouble() }.toDoubleArray()),
            lookaheadInterval = 1.seconds
        ) { event ->
            val time = event.time + 0.1.seconds
            println("Emit: $time")
            SimpleTimelineEvent(time, event.value.map { it + 1.0 }.toDoubleArray())
        }
        val collector = generation.observe {
            collect {
                println("Consume: ${it.time}")
            }
        }
        collector.collect(startTime + 2.seconds)
        collector.close()
    }
 }