Fix warnings

coroutines-interop/src/commonMain/kotlin/com/badoo/reaktive/coroutinesinterop/CoroutineContextScheduler.kt

@@ -20,6 +20,7 @@ import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
 import kotlin.coroutines.CoroutineContext
 import kotlin.time.Duration
+import kotlin.time.TimeSource.Monotonic.ValueTimeMark
 
 internal class CoroutineContextScheduler(
     private val context: CoroutineContext,
@@ -102,7 +103,7 @@ internal class CoroutineContextScheduler(
             }
         }
 
-        private suspend fun delayUntilStart(startTime: Duration) {
+        private suspend fun delayUntilStart(startTime: ValueTimeMark) {
             val uptime = clock.uptime
             if (uptime < startTime) {
                 delay(startTime - uptime)
@@ -123,7 +124,7 @@ internal class CoroutineContextScheduler(
         }
 
         private data class Task(
-            val startTime: Duration,
+            val startTime: ValueTimeMark,
             val period: Duration,
             val task: () -> Unit
         )

coroutines-interop/src/commonTest/kotlin/com/badoo/reaktive/coroutinesinterop/CoroutineContextSchedulerTest.kt

@@ -13,6 +13,8 @@ import kotlin.test.assertTrue
 import kotlin.time.Duration
 import kotlin.time.Duration.Companion.milliseconds
 import kotlin.time.Duration.Companion.seconds
+import kotlin.time.TimeSource
+import kotlin.time.TimeSource.Monotonic.ValueTimeMark
 
 @OptIn(ExperimentalCoroutinesApi::class) // UnconfinedTestDispatcher is experimental
 class CoroutineContextSchedulerTest {
@@ -291,7 +293,7 @@ class CoroutineContextSchedulerTest {
     }
 
     private class TestClock : Clock {
-        override var uptime: Duration = Duration.ZERO
+        override var uptime: ValueTimeMark = TimeSource.Monotonic.markNow()
 
         fun advanceBy(duration: Duration) {
             uptime += duration

coroutines-interop/src/commonTest/kotlin/com/badoo/reaktive/coroutinesinterop/SchedulerCoroutineDispatcherTest.kt

@@ -14,7 +14,6 @@ import kotlin.coroutines.EmptyCoroutineContext
 import kotlin.test.Test
 import kotlin.test.assertEquals
 import kotlin.test.assertTrue
-import kotlin.time.Duration
 import kotlin.time.Duration.Companion.milliseconds
 
 class SchedulerCoroutineDispatcherTest {
@@ -44,15 +43,15 @@ class SchedulerCoroutineDispatcherTest {
         val scheduler = TestScheduler(isManualProcessing = false)
         val dispatcher = SchedulerCoroutineDispatcher(scheduler = scheduler)
         val startTime = DefaultClock.uptime
-        val endTime = AtomicReference(Duration.ZERO)
+        val endTime = AtomicReference(startTime)
 
         launch(dispatcher) {
             delay(500.milliseconds)
             endTime.value = DefaultClock.uptime
         }
 
         withContext(Dispatchers.Default) {
-            while (endTime.value == Duration.ZERO) {
+            while (endTime.value == startTime) {
                 yield()
             }
         }

reaktive/src/commonMain/kotlin/com/badoo/reaktive/completable/BlockingAwait.kt

@@ -13,5 +13,5 @@ import com.badoo.reaktive.maybe.blockingGet
  * Please refer to the corresponding RxJava [document](http://reactivex.io/RxJava/javadoc/io/reactivex/Completable.html#blockingAwait--).
  */
 fun Completable.blockingAwait() {
-    asMaybe().blockingGet()
+    asMaybe().blockingGet<Unit>()
 }

reaktive/src/commonMain/kotlin/com/badoo/reaktive/scheduler/TrampolineScheduler.kt

@@ -10,6 +10,7 @@ import com.badoo.reaktive.utils.clock.DefaultClock
 import com.badoo.reaktive.utils.coerceAtLeastZero
 import com.badoo.reaktive.utils.serializer.serializer
 import kotlin.time.Duration
+import kotlin.time.TimeSource.Monotonic.ValueTimeMark
 
 internal class TrampolineScheduler(
     private val clock: Clock = DefaultClock,
@@ -79,19 +80,19 @@ internal class TrampolineScheduler(
                 return false
             }
 
-            val nextStart = if (task.period.isInfinite()) Duration.INFINITE else clock.uptime + task.period
+            val nextStart = if (task.period.isInfinite()) null else clock.uptime + task.period
 
             task.task()
 
-            if (!nextStart.isInfinite()) {
+            if (nextStart != null) {
                 submit(task.copy(startTime = nextStart))
             }
 
             return true
         }
 
         private data class Task(
-            val startTime: Duration,
+            val startTime: ValueTimeMark,
             val period: Duration,
             val task: () -> Unit
         ) : Comparable<Task> {

reaktive/src/darwinCommonMain/kotlin/com/badoo/reaktive/scheduler/MainScheduler.kt

@@ -10,7 +10,7 @@ import platform.darwin.DISPATCH_TIME_NOW
 import platform.darwin.dispatch_after
 import platform.darwin.dispatch_get_main_queue
 import platform.darwin.dispatch_time
-import kotlin.native.concurrent.AtomicReference
+import kotlin.concurrent.AtomicReference
 import kotlin.time.Duration
 
 internal class MainScheduler : Scheduler {

reaktive/src/nativeCommonMain/kotlin/com/badoo/reaktive/disposable/Various.kt

@@ -1,11 +1,9 @@
 package com.badoo.reaktive.disposable
 
-import kotlin.native.concurrent.AtomicInt
+import kotlin.concurrent.AtomicInt
 
-@Suppress("FunctionName")
 actual inline fun Disposable(crossinline onDispose: () -> Unit): Disposable =
     object : Disposable {
-        @Suppress("ObjectPropertyName") // Backing property
         private var _isDisposed = AtomicInt(0)
         override val isDisposed: Boolean get() = _isDisposed.value != 0
 
@@ -16,7 +14,6 @@ actual inline fun Disposable(crossinline onDispose: () -> Unit): Disposable =
         }
     }
 
-@Suppress("FunctionName")
 actual fun Disposable(): Disposable = SimpleDisposable()
 
 private class SimpleDisposable : Disposable {

reaktive/src/nativeCommonMain/kotlin/com/badoo/reaktive/looperthread/FixedLooperThreadStrategy.kt

@@ -1,6 +1,6 @@
 package com.badoo.reaktive.looperthread
 
-import kotlin.native.concurrent.AtomicInt
+import kotlin.concurrent.AtomicInt
 
 internal class FixedLooperThreadStrategy(threadCount: Int) : LooperThreadStrategy {
 

reaktive/src/nativeCommonMain/kotlin/com/badoo/reaktive/looperthread/LooperThread.kt

@@ -1,10 +1,12 @@
 package com.badoo.reaktive.looperthread
 
 import com.badoo.reaktive.utils.DelayQueue
+import kotlin.native.concurrent.ObsoleteWorkersApi
 import kotlin.native.concurrent.TransferMode
 import kotlin.native.concurrent.Worker
-import kotlin.time.Duration
+import kotlin.time.TimeSource.Monotonic.ValueTimeMark
 
+@OptIn(ObsoleteWorkersApi::class) // There is no replacement yet
 internal class LooperThread {
 
     private val queue = DelayQueue<Message>()
@@ -14,7 +16,7 @@ internal class LooperThread {
         worker.execute(TransferMode.SAFE, { this }) { it.loop() }
     }
 
-    fun schedule(token: Any, startTime: Duration, task: () -> Unit) {
+    fun schedule(token: Any, startTime: ValueTimeMark, task: () -> Unit) {
         queue.offerAt(Message(token, task), startTime)
     }
 

reaktive/src/nativeCommonMain/kotlin/com/badoo/reaktive/scheduler/SchedulerImpl.kt

@@ -7,8 +7,9 @@ import com.badoo.reaktive.looperthread.LooperThreadStrategy
 import com.badoo.reaktive.utils.clock.Clock
 import com.badoo.reaktive.utils.clock.DefaultClock
 import com.badoo.reaktive.utils.coerceAtLeastZero
-import kotlin.native.concurrent.AtomicInt
+import kotlin.concurrent.AtomicInt
 import kotlin.time.Duration
+import kotlin.time.TimeSource.Monotonic.ValueTimeMark
 
 internal class SchedulerImpl(
     private val looperThreadStrategy: LooperThreadStrategy,
@@ -63,7 +64,7 @@ internal class SchedulerImpl(
             submit(startTime = getStartTime(delay), task = t)
         }
 
-        private fun submit(startTime: Duration, task: () -> Unit) {
+        private fun submit(startTime: ValueTimeMark, task: () -> Unit) {
             if (!isDisposed) {
                 looperThread.schedule(token = this, startTime = startTime, task = task)
             }
@@ -73,7 +74,7 @@ internal class SchedulerImpl(
             looperThread.cancel(this)
         }
 
-        private fun getStartTime(delay: Duration): Duration =
+        private fun getStartTime(delay: Duration): ValueTimeMark =
             clock.uptime + delay.coerceAtLeastZero()
     }
 }

reaktive/src/nativeCommonMain/kotlin/com/badoo/reaktive/utils/DelayQueue.kt

@@ -5,8 +5,9 @@ import com.badoo.reaktive.utils.clock.DefaultClock
 import com.badoo.reaktive.utils.lock.ConditionLock
 import com.badoo.reaktive.utils.lock.synchronized
 import com.badoo.reaktive.utils.queue.PriorityQueue
-import kotlin.native.concurrent.AtomicLong
+import kotlin.concurrent.AtomicLong
 import kotlin.time.Duration
+import kotlin.time.TimeSource.Monotonic.ValueTimeMark
 
 internal class DelayQueue<T : Any>(
     private val clock: Clock = DefaultClock,
@@ -66,7 +67,7 @@ internal class DelayQueue<T : Any>(
         offerAt(value, clock.uptime + timeout)
     }
 
-    fun offerAt(value: T, time: Duration) {
+    fun offerAt(value: T, time: ValueTimeMark) {
         lock.synchronized {
             val queue = queue ?: return
             queue.offer(Holder(value, time))
@@ -90,12 +91,12 @@ internal class DelayQueue<T : Any>(
 
     private data class Holder<out T>(
         val value: T,
-        val endTime: Duration,
+        val endTime: ValueTimeMark,
     ) {
         val sequenceNumber = sequencer.addAndGet(1L)
 
         private companion object {
-            private val sequencer = AtomicLong()
+            private val sequencer = AtomicLong(0L)
         }
     }
 

reaktive/src/nativeCommonMain/kotlin/com/badoo/reaktive/utils/ExpirationPool.kt

@@ -1,5 +1,6 @@
 package com.badoo.reaktive.utils
 
+import kotlin.native.concurrent.ObsoleteWorkersApi
 import kotlin.native.concurrent.TransferMode
 import kotlin.native.concurrent.Worker
 import kotlin.time.Duration
@@ -9,6 +10,7 @@ import kotlin.time.Duration
  * DelayQueue should be destroyed, but all readers and writers must be cancelled first.
  * See LooperThread for sample implementation.
  */
+@OptIn(ObsoleteWorkersApi::class) // There is no replacement yet
 internal class ExpirationPool<T : Any>(
     private val onItemExpired: (T) -> Unit
 ) {

reaktive/src/nativeCommonTest/kotlin/com/badoo/reaktive/looperthread/LooperThreadTest.kt

@@ -10,6 +10,7 @@ import kotlin.test.Test
 import kotlin.test.assertFalse
 import kotlin.test.assertTrue
 import kotlin.time.Duration.Companion.milliseconds
+import kotlin.time.TimeSource
 
 class LooperThreadTest {
 
@@ -18,9 +19,9 @@ class LooperThreadTest {
         val isExecuted = AtomicBoolean()
         val lock = ConditionLock()
         val thread = LooperThread()
-        val startTime = getTimeMillis() + 200L
+        val startTime = TimeSource.Monotonic.markNow() + 200.milliseconds
 
-        thread.schedule(Unit, startTime.milliseconds) {
+        thread.schedule(Unit, startTime) {
             lock.synchronized {
                 isExecuted.value = true
                 lock.signal()
@@ -31,7 +32,7 @@ class LooperThreadTest {
             lock.waitForOrFail(predicate = isExecuted::value)
         }
 
-        assertTrue(getTimeMillis() >= startTime)
+        assertTrue(startTime.hasPassedNow())
     }
 
     @Test
@@ -40,7 +41,7 @@ class LooperThreadTest {
         val lock = ConditionLock()
         val thread = LooperThread()
 
-        thread.schedule(Unit, getTimeMillis().milliseconds) {
+        thread.schedule(Unit, TimeSource.Monotonic.markNow()) {
             lock.synchronized {
                 isExecuted.value = true
                 lock.signal()
@@ -54,7 +55,7 @@ class LooperThreadTest {
         thread.destroy()
 
         isExecuted.value = false
-        thread.schedule(Unit, getTimeMillis().milliseconds) {
+        thread.schedule(Unit, TimeSource.Monotonic.markNow()) {
             isExecuted.value = true
         }
 

utils-internal/src/commonMain/kotlin/com/badoo/reaktive/utils/clock/Clock.kt

@@ -1,10 +1,10 @@
 package com.badoo.reaktive.utils.clock
 
 import com.badoo.reaktive.utils.InternalReaktiveApi
-import kotlin.time.Duration
+import kotlin.time.TimeSource.Monotonic.ValueTimeMark
 
 @InternalReaktiveApi
 interface Clock {
 
-    val uptime: Duration
+    val uptime: ValueTimeMark
 }

utils-internal/src/commonMain/kotlin/com/badoo/reaktive/utils/clock/DefaultClock.kt

@@ -1,6 +1,11 @@
 package com.badoo.reaktive.utils.clock
 
 import com.badoo.reaktive.utils.InternalReaktiveApi
+import kotlin.time.TimeSource
+import kotlin.time.TimeSource.Monotonic.ValueTimeMark
 
 @InternalReaktiveApi
-expect object DefaultClock : Clock
+object DefaultClock : Clock {
+
+    override val uptime: ValueTimeMark get() = TimeSource.Monotonic.markNow()
+}

utils-internal/src/commonTest/kotlin/com/badoo/reaktive/utils/clock/DefaultClockTest.kt

@@ -4,7 +4,6 @@ import kotlin.test.Test
 import kotlin.test.assertTrue
 import kotlin.time.Duration
 import kotlin.time.Duration.Companion.milliseconds
-import kotlin.time.ExperimentalTime
 import kotlin.time.TimeSource
 
 class DefaultClockTest {
@@ -15,11 +14,9 @@ class DefaultClockTest {
         busySleep(1.milliseconds)
         val time2 = DefaultClock.uptime
 
-        assertTrue(time1.isPositive())
         assertTrue(time2 - time1 >= 1.milliseconds)
     }
 
-    @OptIn(ExperimentalTime::class)
     private fun busySleep(duration: Duration) {
         val end = TimeSource.Monotonic.markNow()
         while (end.elapsedNow() < duration) {

utils-internal/src/nativeCommonMain/kotlin/com/badoo/reaktive/utils/atomic/AtomicBoolean.kt

@@ -5,7 +5,7 @@ import com.badoo.reaktive.utils.InternalReaktiveApi
 @InternalReaktiveApi
 actual class AtomicBoolean actual constructor(initialValue: Boolean) {
 
-    private val delegate = kotlin.native.concurrent.AtomicInt(initialValue.intValue)
+    private val delegate = kotlin.concurrent.AtomicInt(initialValue.intValue)
 
     actual var value: Boolean
         get() = delegate.value != 0

utils-internal/src/nativeCommonMain/kotlin/com/badoo/reaktive/utils/atomic/AtomicInt.kt

@@ -5,7 +5,7 @@ import com.badoo.reaktive.utils.InternalReaktiveApi
 @InternalReaktiveApi
 actual class AtomicInt actual constructor(initialValue: Int) {
 
-    private val delegate = kotlin.native.concurrent.AtomicInt(initialValue)
+    private val delegate = kotlin.concurrent.AtomicInt(initialValue)
 
     actual var value: Int
         get() = delegate.value