rx (feature): Add rx.delay

airframe-rx/src/main/scala/wvlet/airframe/rx/Rx.scala

@@ -151,6 +151,14 @@ trait Rx[+A] extends RxOps[A] {
 
   def withName(name: String): Rx[A] = NamedOp(this, name)
 
+  /**
+    * Emit the new value after the given time interval.
+    * @param interval
+    * @param unit
+    * @return
+    */
+  def delay(interval: Long, unit: TimeUnit): Rx[A] = DelayOp(this, interval, unit)
+
   /**
     * Applies `f` to the input value and return the result.
     * @param f
@@ -1031,7 +1039,9 @@ object Rx extends LogSupport {
   case class TimerOp(interval: Long, unit: TimeUnit) extends Rx[Long] {
     override def parents: Seq[RxOps[_]] = Seq.empty
   }
-
+  case class DelayOp[A](input: RxOps[A], interval: Long, unit: TimeUnit) extends UnaryRx[A, A] {
+    override def parents: Seq[RxOps[_]] = Seq(input)
+  }
   case class TakeOp[A](input: RxOps[A], n: Long) extends Rx[A] {
     override def parents: Seq[RxOps[_]] = Seq(input)
   }

airframe-rx/src/main/scala/wvlet/airframe/rx/RxRunner.scala

@@ -342,6 +342,18 @@ class RxRunner(
               effect(OnError(e))
           }
         }
+      case DelayOp(in, interval, unit) =>
+        val delayMillis = TimeUnit.MILLISECONDS.convert(interval, unit).max(1)
+        run(in) {
+          case OnNext(v) =>
+            // Delay the emission of the value
+            compat.scheduleOnce(delayMillis) {
+              effect(OnNext(v.asInstanceOf[A]))
+            }
+            RxResult.Continue
+          case other =>
+            effect(other)
+        }
       case ThrottleFirstOp(in, interval, unit) =>
         var lastUpdateTimeNanos = -interval
         run(in) {