Where do you call Local.isolate after this change if using Akka-Http?
I think the whole point is that we don't call Local.isolate since we don't even want isolation
Is this theoretically a problem?
This is the future -> task -> future case (which is sequential)
Hello!
I was wondering if it does make sense creating a Consumer that returns an Observable on success?
That would allow to chaining it with another consumer or working with the result value of each processed event, as it is done in akka streams with .mapMaterializedValue()
In which in that case the materialised value return a Task[Observable[Task[Response]].
This would be the consumer:
Consumer.create[In, Observable[Task[Out]]] { (scheduler, _, callback) =>
new Observer.Sync[In] {
val observable: ConcurrentSubject[Task[Out], Task[Out]] =
ConcurrentSubject[Task[Out]](MulticastStrategy.replay)(scheduler)
def onNext(request: In): Ack = {
val response: Task[Out] = Task.fromFuture(request.execute(query))
observable.onNext(observable)
monix.execution.Ack.Continue
}
def onComplete(): Unit = {
callback.onSuccess(subject)
}
def onError(ex: Throwable): Unit = {
callback.onError(ex)
}
}
}
Comments are on the PR
@Avasil Is there anything else you want to add for monix/monix#1146 or do you want to go ahead and merge it?
BTW all of the tests are now passing in monix-opentracing with the latest monix snapshot that has the changes in the PR, only one test is failing but its completely unrelated to monix
This is
FutureScalaConcurrentAutoFinishSpec
I would also look at https://github.com/opentracing-contrib/scala-concurrent
There is also
io.opentracing.util.ThreadLocalScopeManager which is what monix-opentracing is based off (this is OpenTracing's default ScopeManager which just uses a ThreadLocal)
Let me know if you need any help!
Hello everyone. Sometimes I get this message when I shut down the application. https://codeshare.io/5PVgRd
Looks like some task is trying to execute after the Scheduler is already down, isn't it? Here's a sample code. Any advice on how to avoid this kind of message?
Looks like some task is trying to execute after the Scheduler is already down, isn't it? Here's a sample code. Any advice on how to avoid this kind of message?
import io.netty.channel.{Channel, ChannelHandlerContext, ChannelInboundHandlerAdapter}
import monix.execution.Scheduler
import monix.reactive.subjects.ConcurrentSubject
// scheduler here is just a monix.execution.schedulers.fixedPool
class MyObserver(someConstant: Int, implicit val scheduler: Scheduler) extends ChannelInboundHandlerAdapter {
private val scores = ConcurrentSubject.publish[(Channel, BigInt)]
override def channelRead(ctx: ChannelHandlerContext, msg: AnyRef): Unit = msg match {
case sc: BigInt => scores.onNext((ctx.channel(), sc))
case _ => super.channelRead(ctx, msg)
}
def shutdown(): Unit = {
scores.onComplete()
}
}
@AlexITC Yes, you can track: monix/monix#1131
I will see if I have permissions to update minitest myself if @alexandru is away
I will see if I have permissions to update minitest myself if @alexandru is away
We are also facing the same problem you are
Should expect a release soon
I also have to do some research to see how OpenTracing's
Scope works because its slightly confusing now
Sharing this small program here in case anyone is interested:
override def run(args: List[String]): Task[ExitCode] = {
val writeToFile = (str: String) => Task.sleep(FiniteDuration(5, TimeUnit.SECONDS)) >> Task {
import java.io._
val pw = new PrintWriter(new File("hello.txt"))
pw.write(str + "\n")
pw.close
}
val program = for {
userInput <- Task(StdIn.readLine())
fiber <- writeToFile(userInput).start
_ <-
if (userInput.startsWith("ignore"))
fiber.cancel >> Task(println(s"Ignoring user input: $userInput"))
else
Task(println(s"Slowly writing to file...: $userInput"))
} yield ExitCode.Success
program >> run(args)
}
@Avasil as you said, the use case is also covered by
Observable.toListL, the one think is that it might be handy for people whose are familiarised with Akka Streams api, in which for materialising streams Sink.seq is commonly used. This is my opinion, on the other hand if you think it is going to bring confusion we should just not add it.
And I run the following code
val conf = KafkaConsumerConfig(myConf, includeDefaults=true)
val foo = myConf.getString("auto.commit.enable")
println(foo)
println(conf.enableAutoCommit)
myConf being the above configuration. I get this output
true
false
The defaults is overriding the conf that I passed in.
And for context I'm running these versions
val monixVersion = "3.1.0"
val monixKafkaVersion = "1.0.0-RC5"
val kafkaVersion = "2.4.0"