import monix.execution.{Ack, Cancelable, Scheduler}
import monix.reactive.Observable
import monix.reactive.observers.Subscriber
import scala.concurrent.Future
class ObservableOps[A](delegate: Observable[A]) {
def concatWithLast(fn: A=>Observable[A]): Observable[A] = {
Observable.unsafeCreate[A] { subscriber =>
var lastOpt: Option[A] = None
var c0 = Cancelable.empty
def subscribeWithLast(): Unit = {
lastOpt match {
case None =>
subscriber.onComplete()
case Some(last) =>
c0 = fn(last).subscribe(subscriber)
}
}
val c1 =
delegate.subscribe(
new Subscriber[A] {
override implicit def scheduler: Scheduler =
subscriber.scheduler
override def onNext(elem: A): Future[Ack] = {
lastOpt = Some(elem)
subscriber.onNext(elem)
}
override def onError(ex: Throwable): Unit = {
subscriber.onError(ex)
}
override def onComplete(): Unit = {
subscribeWithLast()
}
}
)
Cancelable.collection(c0, c1)
}
}
}
Seems like @squadgazzz use case is the same
the source wasn't closed
I can try it again as I fixed a bunch of stuff that "may" have caused that
I mean the source didn't have onComplete eagerly called
which is needed
the use case is reading the tail of a message log
so we want existing messages
then switch to a hot observable that is the tail
so it is in effect
def existingMessages: Observable[Message] = ...
def tailObserver: Observable[Message] = ...existingMessages ++ tailObserver
where we need to get the onComplete on existingMessagess otherwise readers hang around forever
fwiw Monix has made this much much simpler
I plan to create a simple example of what you recommended to show what I ran into
just too many things needed before we release
Seems like @squadgazzz use case is the same
my case is similar, but I know when to stop and this means I know what’s the last element.
val obs1 = Observable
.fromIterable(List(1, 2, 3, 4, 5, 6, 7, 8, 9))
.doOnNext(e => Task(println(e)))
.takeWhileInclusive(_ < 5)
.flatMap {
case 5 => Observable.pure(5) ++ Observable.fromIterable(List(10, 11, 12, 13, 14))
case e => Observable.pure(e)
}In case I don’t know the last element we should properly use publishSelector somehow
I need the existingMessages observable to have onComplete eagerly called
the flatMap means the source (existingMessages in my case) never has onComplete called
so it leaves lots of readers hanging out there
the cool thing for me is I now have an in memory hot path for messages so the latency goes way way down
and the amount of code to do that is almost nil
I just get it for free
so
concatWithLast is a very mild hack for the gains it promises
this is a TON of things listening to a large pulsar cluster
calls a function on a last element and then emits the results concatenated with a current one?
my
concatWithLast is incredibly and admittedly naive
This message was deleted
Can anyone please explain what's going on here?
val obs = Observable.fromIterable(1 to 2)
val result = obs.publishSelector { o =>
o ++ Observable.fromIterable(11 to 12)
}
result.toListL.executeAsync.runToFuture.foreach(l => println(l.mkString("[", ",", "]")))
Thread.sleep(5.seconds.toMillis)result: [1,2,11,12]
val obs = Observable.fromIterable(1 to 2)
val result = obs.publishSelector { o =>
o ++ Observable.fromIterable(11 to 12)
} ++ Observable.fromIterable(101 to 102)
result.toListL.executeAsync.runToFuture.foreach(l => println(l.mkString("[", ",", "]")))
Thread.sleep(5.seconds.toMillis) result: [11,12,101,102]
Where did obs's elements gone?
I smell a subscription timing issue, since
publishSelector works with a PublishSubject underneath, and it might lose initial events if the underlying implementation subscribes too early.
fwiw this does have the behaviour you expect
val obs = Observable.fromIterable(1 to 2)
val result =
obs
.publishSelector { o =>
o ++ Observable.fromIterable(11 to 12)
}
.publishSelector(_ ++ Observable.fromIterable(101 to 102))
result.toListL.executeAsync.runToFuture.foreach(l => println(l.mkString("[", ",", "]")))
Thread.sleep(5000)
but that becomes untenable to anywhere you use the obs you have to use it inside of publishSelector