ConcurrentQueue?
@invis87
There are at least two ways, depending on whether you want to be pure or not, I will show you both.
Impure version using Hot Observable:
implicit val s = Scheduler.global
val stream: Observable[Long] =
Observable
.intervalAtFixedRate(100.millis)
.doOnNext(l => Task(println(s"Processing $l")))
// BehaviorSubject will always emit latest value as first element
val behaviorObservable: ConnectableObservable[Long] = stream.behavior(initialValue = 0L)
val impure = for {
// starts emitting elements to all subscribers, at this point `stream` will start executing
_ <- Task(behaviorObservable.connect())
lastElement: Task[Long] = behaviorObservable.firstL
_ <- lastElement.map(el => println(s"lastElement: $el"))
_ <- lastElement.map(el => println(s"lastElement: $el")).delayExecution(500.millis)
_ <- lastElement.map(el => println(s"lastElement: $el")).delayExecution(1.second)
} yield ()
impure.runSyncUnsafe()Observable + Ref: val pure = for {
lastElement <- Ref[Task].of(0L)
// will start running in the background
_ <- stream.doOnNext(l => lastElement.set(l)).completedL.forkAndForget
_ <- lastElement.get.map(el => println(s"lastElement: $el"))
_ <- lastElement.get.map(el => println(s"lastElement: $el")).delayExecution(500.millis)
_ <- lastElement.get.map(el => println(s"lastElement: $el")).delayExecution(1.second)
} yield ()
pure.runSyncUnsafe()
var currentValue: Long = 0
val stream: Observable[Long] =
Observable
.intervalAtFixedRate(300.millis)
.doOnNext(l => println(s"processing: $l"))
.doOnNext(v => currentValue = v)
stream.subscribe()currentValue should be volatile
It's bad if you want pure FP but you can't have a pure mutable object without any effect like Task. In your case I think this approach is fine.
Yes, it needs to be either volatile or Atomic. Maybe try something like
def currentToken: AuthToken = currentValue
@volatile private var currentValue: Long = 0or
def currentToken: AuthToken = currentValue.get
private val currentValue: AtomicLong = AtomicLong(0L)so it can't be mutated from outside.
Var should be available for all Scala versions
3.0.0-eddc79d-SNAPSHOT
Hi, I just read through https://blog.softwaremill.com/thread-shifting-in-cats-effect-and-zio-9c184708067b and was wondering how this works in monix.
Suppose I run
t1 *> t2 *> t3and I want t1 and t3 to run on the same scheduler, but t2 is wrapping some library call that uses its own thread pool which is not in my control,
how can I achieve this? I found executeOn, but from reading the documentation I don't understand if it would shift back after the task is run.
@danielkarch
It will always come back like in ZIO:
implicit val s1 = Scheduler.singleThread("s1") // default scheduler
val s2 = Scheduler.singleThread("s2")
val s3 = Scheduler.singleThread("s3")
def task(i: Int) = Task(println(s"$i: Executing on ${Thread.currentThread().getName}"))
val t =
for {
_ <- task(1).executeOn(s2) // executes on s2
_ <- task(2) // executes on s1
_ <- task(3).executeOn(s3) // executes on s3
_ <- task(4) // executes on s1
// executes both tasks on s2
_ <- (task(5) >> Task.shift >> task(6)).executeOn(s2)
} yield ()
t.runSyncUnsafe()Task.async , fromFuture worked like in Cats-Effect until quite recently, on master it always stays on the default Scheduler
Task.shift in the example?
Thank you. So I get this behaviour with
3.0.0-RC3?
For executeOn yes, I think it works like this even in 2.x but I'm not sure. For Task builders it is not released yet
What is the purpose of the
Task.shiftin the example?
shift without argument shifts to the default thread pool, not the current one. cats.effect.IO with evalOn would execute 5 on s2, then shift back to s1 and execute 6 on s1. Task.executeOn overrides the default for the duration of a Task so it is safe in this regard. I think it will also happen for IO in 3.0 but it needs a major change in the internals
Hi All,
I just learned about TaskLocal.isolate and started using it to isolate task from incoming http requests to make sure that MDC and other tracing utilities are set properly. It works well for that use case.
However I found a strange behavior when using Task.fromFuture. My goal was to emulate a 3rd party library, blocking and clearing the MDC.
test("isolate contexts from Future" ) {
implicit val scheduler: Scheduler = monix.execution.Scheduler.Implicits.traced
implicit val opts: Task.Options = Task.defaultOptions.enableLocalContextPropagation
val s2 = Scheduler.singleThread("s2")
val local: Local[Int] = Local[Int](-1)
val test = (i: Int) => {
for {
_ <- Task(local.update(i))
_ <- TaskLocal.isolate(Task.fromFuture(Future {
Thread.sleep(10)
local.clear()
}(s2)))
r <- Task(assert(local.get == i))
} yield r
}
Task.gather(Range(0, 5).map(i => TaskLocal.isolate(test(i)))).runSyncUnsafeOpt()
}The code above fails, but only with the addition of Thread.sleep.
I could make the test pass by shifting back just after Task.fromFuture, or using Task.executeOn.
Am I missing something?
I tested with the latest on master, as I thought that this fix could help: monix/monix#966
val local: Local[Int] = Local[Int](-1)
val test = (i: Int) => {
for {
_ <- Task(local.update(i))
_ <- TaskLocal.isolate(Task.deferFuture(Future {
Thread.sleep(10)
local.clear()
}) <* Task.shift)
r <- Task(assert(local.get == i))
} yield r
}
Task.gather(Range(0, 5).map(i => TaskLocal.isolate(test(i)))).runSyncUnsafeOpt()@oleg-py I think it is just yet another reason to revisit restoreLocals in 3.1.0 :P monix/monix#986
@botrunner_gitlab I think I minimized failing case a bit and I think my fix should work with your original example. Are you able to verify from my PR?
Resource[Coeval, Whatever] to Resource[Task, Whatever]. I've found .mapK and applied it by defining a def coevalToTask[A](coeval: Coeval[A]): Task[A] = coeval.to[Task], but this feels like I'm just repeating the signature of .to[Task]. I'm pretty new to the FunctionK-concept so maybe I'm missing some imports?
unfold?