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
@squadgazzz can you please open an issue? I'll take a look.
@fizzy33 yeah, this works if I know about previous
publishSelector call.
Hello guys, have you ever experienced migrating custom akka flow stages to monix, is it possible to do it without scheduler evidence? In the example below is it possible to emit batches from OutputStream to something observable-like
val buf = new OutputStream {
override def write(b: Int): Unit = {
/* previously bytes were emitted to the next stage of the flow */
}
}
Hello, I'm using monix circuit breaker. All in all it works great, but I'd like to let some of the exceptions happen and still be thrown without counting towards failures. Does anyone have any hints on how to achieve that?
Currently it looks similar to:
circuitBreaker
.protect { Task... }
.toAsync(Async[M], eff)
.recoverWith {
case t: ErrIDontWantToCount => Effect[M].raiseError(t)
case NonFatal(ex) if (n - 1) > 0 =>
retry(n - 1)
case t => Effect[M].raiseError(FailedLastRetry(t))
}and ErrIDontWantToCount is handled correctly, but counts towards failures
hang on I need to read the docs for timeout :-)
ah I see
nm
tx