You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
83 lines
2.7 KiB
83 lines
2.7 KiB
import org.scalatest.funsuite.AnyFunSuite
|
|
import monix.eval.Task
|
|
import monix.reactive.subjects.ConcurrentSubject
|
|
import monix.eval.Coeval
|
|
import com.typesafe.scalalogging.LazyLogging
|
|
import monix.execution.Scheduler.global
|
|
import nova.monadic_sfx.util.reactive.store.MonixProSubject
|
|
import scala.concurrent.duration._
|
|
import scala.concurrent.Await
|
|
import monix.reactive.Observable
|
|
|
|
class ProSubjectTest extends AnyFunSuite with LazyLogging {
|
|
test("task1") {
|
|
implicit val s = global
|
|
val x = Await.result(task1.runToFuture, 10.seconds)
|
|
assert(x == 1)
|
|
}
|
|
|
|
test("task2") {
|
|
implicit val s = global
|
|
val x = Await.result(task2.runToFuture, 10.seconds)
|
|
}
|
|
|
|
def task1 =
|
|
Task
|
|
.deferAction(implicit s =>
|
|
Task {
|
|
val sub = ConcurrentSubject.publish[Int]
|
|
// val obs = sub.scan0(0)(_ + _).behavior(2).refCount
|
|
val obs =
|
|
sub
|
|
.scanEval0F(Task.pure(0))((a, b) => Task(a + b))
|
|
.behavior(2)
|
|
.refCount
|
|
.doOnNextF(i => Coeval(println(s"Emitted1: $i")))
|
|
// type MonixProSubject[-I, +O] = Observable[O] with Observer[I]
|
|
// def from[I, O](
|
|
// observer: Observer[I],
|
|
// observable: Observable[O]
|
|
// ): MonixProSubject[I, O] =
|
|
// new Observable[O] with Observer[I] {
|
|
// override def onNext(elem: I): Future[Ack] = observer.onNext(elem)
|
|
// override def onError(ex: Throwable): Unit = observer.onError(ex)
|
|
// override def onComplete(): Unit = observer.onComplete()
|
|
// override def unsafeSubscribeFn(
|
|
// subscriber: Subscriber[O]
|
|
// ): Cancelable =
|
|
// observable.unsafeSubscribeFn(subscriber)
|
|
// }
|
|
MonixProSubject.from(sub, obs)
|
|
}.flatMap {
|
|
case ps =>
|
|
Task {
|
|
ps
|
|
.doOnNextF(i => Coeval(println(s"Emitted item 1: $i")))
|
|
.subscribe()
|
|
|
|
} >> Task {
|
|
ps
|
|
.doOnNextF(i => Coeval(println(s"Emitted item 2: $i")))
|
|
.subscribe()
|
|
} >> Task {
|
|
(0 to 5).foreach { i => ps.onNext(i) }
|
|
} >> Task(1)
|
|
}
|
|
)
|
|
|
|
val task2 = Task.deferAction(implicit s =>
|
|
Task(
|
|
Observable(1, 2, 3, 4, 5)
|
|
.scan0(0)(_ + _)
|
|
.behavior(2)
|
|
.refCount
|
|
// .doOnNextF(i => Coeval(println(s"Emitted2: $i")))
|
|
.delayExecution(10.millis)
|
|
)
|
|
.flatMap(res =>
|
|
Task(res.doOnNextF(i => Coeval(println(s"1: $i"))).subscribe()).start >>
|
|
Task(res.doOnNextF(i => Coeval(println(s"2: $i"))).subscribe()).start
|
|
)
|
|
)
|
|
|
|
}
|