make reactive state variables lazy

colibri/src/main/scala/colibri/Subject.scala

@@ -13,6 +13,10 @@ final class ReplayLatestSubject[A] extends Observer[A] with Observable.MaybeValu
 
   @inline def now(): Option[A] = current
 
+  def unsafeResetState(): Unit = {
+    current = None
+  }
+
   def unsafeOnNext(value: A): Unit = {
     current = Some(value)
     state.unsafeOnNext(value)
@@ -37,6 +41,10 @@ final class ReplayAllSubject[A] extends Observer[A] with Observable[A] {
 
   @inline def now(): Seq[A] = current.toSeq
 
+  def unsafeResetState(): Unit = {
+    current.clear()
+  }
+
   def unsafeOnNext(value: A): Unit = {
     current += value
     state.unsafeOnNext(value)

reactive/src/main/scala-2/colibri/reactive/OwnerPlatform.scala

@@ -2,26 +2,14 @@ package colibri.reactive
 
 import colibri.{Observable, Cancelable}
 
-trait OwnerPlatform {
-  @annotation.compileTimeOnly(
-    "No implicit Owner is available here! Wrap inside `Owned { <code> }`, or provide an implicit `Owner`, or `import Owner.unsafeImplicits._` (dangerous).",
-  )
-  implicit object compileTimeMock extends Owner {
-    def unsafeSubscribe(): Cancelable                   = ???
-    def unsafeOwn(subscription: () => Cancelable): Unit = ???
-    def cancelable: Cancelable                          = ???
-  }
-}
-
 trait LiveOwnerPlatform {
   @annotation.compileTimeOnly(
     "No implicit LiveOwner is available here! Wrap inside `Rx { <code> }`, or provide an implicit `LiveOwner`.",
   )
   implicit object compileTimeMock extends LiveOwner {
-    def unsafeSubscribe(): Cancelable                   = ???
-    def unsafeOwn(subscription: () => Cancelable): Unit = ???
-    def cancelable: Cancelable                          = ???
-    def unsafeLive[A](rx: Rx[A]): A                     = ???
-    def liveObservable: Observable[Any]                 = ???
+    def cancelable: Cancelable          = ???
+    def unsafeNow[A](rx: Rx[A]): A      = ???
+    def unsafeLive[A](rx: Rx[A]): A     = ???
+    def liveObservable: Observable[Any] = ???
   }
 }

reactive/src/main/scala-2/colibri/reactive/RxPlatform.scala

@@ -3,5 +3,5 @@ package colibri.reactive
 import colibri.reactive.internal.MacroUtils
 
 trait RxPlatform {
-  def apply[R](f: R)(implicit owner: Owner): Rx[R] = macro MacroUtils.rxImpl[R]
+  def apply[R](f: R): Rx[R] = macro MacroUtils.rxImpl[R]
 }

reactive/src/main/scala-2/colibri/reactive/internal/MacroUtils.scala

@@ -1,30 +1,23 @@
 package colibri.reactive.internal
 
-import colibri.reactive.{Rx, Owner, LiveOwner}
-import colibri.effect.SyncEmbed
-import colibri.SubscriptionOwner
+import colibri.reactive.{Rx, NowOwner, LiveOwner}
 import scala.reflect.macros._
 
 // Inspired by scala.rx
 object MacroUtils {
 
-  private val ownerName     = "colibriOwner"
-  private val liveOwnerName = "colibriLiveOwner"
-
-  def injectOwner[T](c: blackbox.Context)(src: c.Tree, newOwner: c.universe.TermName, exceptOwner: c.Type): c.Tree = {
+  def injectOwner[T](c: blackbox.Context)(src: c.Tree, newOwner: c.universe.TermName, replaces: Set[c.Type]): c.Tree = {
     import c.universe._
 
-    val implicitOwnerAtCaller     = c.inferImplicitValue(typeOf[Owner], silent = false)
-    val implicitLiveOwnerAtCaller = c.inferImplicitValue(typeOf[LiveOwner], silent = false)
+    val implicitOwnerAtCallers = replaces.map(c.inferImplicitValue(_, silent = false))
 
     object transformer extends c.universe.Transformer {
       override def transform(tree: c.Tree): c.Tree = {
         val shouldReplaceOwner = tree != null &&
           tree.isTerm &&
-          (tree.tpe =:= implicitOwnerAtCaller.tpe || tree.tpe =:= implicitLiveOwnerAtCaller.tpe) &&
-          tree.tpe <:< typeOf[Owner] &&
-          !(tree.tpe =:= typeOf[Nothing]) &&
-          !(tree.tpe <:< exceptOwner)
+          implicitOwnerAtCallers.exists(_.tpe =:= tree.tpe) &&
+          tree.tpe <:< typeOf[NowOwner] &&
+          !(tree.tpe =:= typeOf[Nothing])
 
         if (shouldReplaceOwner) q"$newOwner"
         else super.transform(tree)
@@ -33,37 +26,23 @@ object MacroUtils {
     transformer.transform(src)
   }
 
-  def ownedImpl[R](
-      c: blackbox.Context,
-  )(f: c.Expr[R])(subscriptionOwner: c.Expr[SubscriptionOwner[R]], syncEmbed: c.Expr[SyncEmbed[R]]): c.Expr[R] = {
+  def rxImpl[R](c: blackbox.Context)(f: c.Expr[R]): c.Expr[Rx[R]] = {
     import c.universe._
 
-    val newOwner = c.freshName(TermName(ownerName))
+    val newOwner = c.freshName(TermName("colibriLiveOwner"))
 
-    val newTree = c.untypecheck(injectOwner(c)(f.tree, newOwner, typeOf[LiveOwner]))
-
-    val tree = q"""
-    _root_.colibri.reactive.Owned.function { ($newOwner: _root_.colibri.reactive.Owner) =>
-      $newTree
-    }($subscriptionOwner, $syncEmbed)
-    """
-
-    // println(tree)
-
-    c.Expr(tree)
-  }
-
-  def rxImpl[R](c: blackbox.Context)(f: c.Expr[R])(owner: c.Expr[Owner]): c.Expr[Rx[R]] = {
-    import c.universe._
-
-    val newOwner = c.freshName(TermName(liveOwnerName))
-
-    val newTree = c.untypecheck(injectOwner(c)(f.tree, newOwner, typeOf[Nothing]))
+    val newTree = c.untypecheck(
+      injectOwner(c)(
+        f.tree,
+        newOwner,
+        replaces = Set(typeOf[LiveOwner], typeOf[NowOwner]),
+      ),
+    )
 
     val tree = q"""
     _root_.colibri.reactive.Rx.function { ($newOwner: _root_.colibri.reactive.LiveOwner) =>
       $newTree
-    }($owner)
+    }
     """
 
     // println(tree)

reactive/src/main/scala-3/colibri/reactive/OwnerPlatform.scala

@@ -1,5 +1,3 @@
 package colibri.reactive
 
-trait OwnerPlatform
-
 trait LiveOwnerPlatform

reactive/src/main/scala-3/colibri/reactive/RxPlatform.scala

@@ -1,5 +1,5 @@
 package colibri.reactive
 
 trait RxPlatform {
-  def apply[R](f: LiveOwner ?=> R)(implicit owner: Owner): Rx[R] = Rx.function(implicit owner => f)
+  def apply[R](f: LiveOwner ?=> R): Rx[R]   = Rx.function(implicit owner => f)
 }

reactive/src/main/scala/colibri/reactive/Owner.scala

@@ -2,68 +2,46 @@ package colibri.reactive
 
 import colibri._
 
-@annotation.implicitNotFound(
-  "No implicit Owner is available here! Wrap inside `Owned { <code> }`, or provide an implicit `Owner`, or `import Owner.unsafeImplicits._` (dangerous).",
-)
-trait Owner {
-  def cancelable: Cancelable
-  def unsafeSubscribe(): Cancelable
-  def unsafeOwn(subscription: () => Cancelable): Unit
+trait NowOwner {
+  def unsafeNow[A](rx: Rx[A]): A
 }
-object Owner extends OwnerPlatform {
-  def unsafeHotRef(): Owner = new Owner {
-    val refCountBuilder = Cancelable.refCountBuilder()
-    var initialRef      = refCountBuilder.ref()
-
-    def cancelable: Cancelable = refCountBuilder
-
-    def unsafeSubscribe(): Cancelable = if (initialRef == null) {
-      refCountBuilder.ref()
-    } else {
-      val result = initialRef
-      initialRef = null
-      result
+object NowOwner {
+  implicit object global extends NowOwner {
+    def unsafeNow[A](rx: Rx[A]): A = {
+      val cancelable = rx.unsafeSubscribe()
+      try(rx.nowIfSubscribed()) finally(cancelable.unsafeCancel())
     }
-
-    def unsafeOwn(subscription: () => Cancelable): Unit = refCountBuilder.unsafeAdd(subscription)
-  }
-
-  object unsafeGlobal extends Owner {
-    def cancelable: Cancelable                          = Cancelable.empty
-    def unsafeSubscribe(): Cancelable                   = Cancelable.empty
-    def unsafeOwn(subscription: () => Cancelable): Unit = {
-      subscription()
-      ()
-    }
-  }
-
-  object unsafeImplicits {
-    implicit def unsafeGlobalOwner: Owner = unsafeGlobal
   }
 }
 
 @annotation.implicitNotFound(
   "No implicit LiveOwner is available here! Wrap inside `Rx { <code> }`, or provide an implicit `LiveOwner`.",
 )
-trait LiveOwner  extends Owner             {
+trait LiveOwner extends NowOwner {
+  def cancelable: Cancelable
+
   def liveObservable: Observable[Any]
+
   def unsafeLive[A](rx: Rx[A]): A
 }
 object LiveOwner extends LiveOwnerPlatform {
-  def unsafeHotRef()(implicit parentOwner: Owner): LiveOwner = new LiveOwner {
-    val owner: Owner = Owner.unsafeHotRef()
-    parentOwner.unsafeOwn(() => owner.unsafeSubscribe())
+  def unsafeHotRef(): LiveOwner = new LiveOwner {
+    private val ref = Cancelable.builder()
 
-    val liveObservableArray             = new scala.scalajs.js.Array[Observable[Any]]()
-    val liveObservable: Observable[Any] = Observable.mergeIterable(liveObservableArray)
+    private val subject = Subject.publish[Any]()
 
-    def unsafeLive[A](rx: Rx[A]): A = {
-      liveObservableArray.push(rx.observable)
-      rx.now()
+    val cancelable: Cancelable = ref
+
+    val liveObservable: Observable[Any] = subject
+
+    def unsafeNow[A](rx: Rx[A]): A = {
+      ref.unsafeAdd(() => rx.observable.unsafeSubscribe())
+      rx.nowIfSubscribed()
     }
 
-    def unsafeSubscribe(): Cancelable                   = owner.unsafeSubscribe()
-    def unsafeOwn(subscription: () => Cancelable): Unit = owner.unsafeOwn(subscription)
-    def cancelable: Cancelable                          = owner.cancelable
+    def unsafeLive[A](rx: Rx[A]): A = {
+      ref.unsafeAdd(() => rx.observable.via(subject).unsafeSubscribe())
+      rx.nowIfSubscribed()
+    }
   }
 }