monix/monix

Asynchronous, Reactive Programming for Scala and Scala.js

matfournier

@matfournier

But how would I detect that I've hit the limit? That this has happened?

Piotr Gawryś

@Avasil

The simplest way is probably doOnEarlyStop. Something like doOnComplete or guarantee should work as well

Paweł Gontarz

@Garnek20

hello guys, can anyone tell me why Task is represented as: (A => Unit) => Unit ?
It is depicted in documentation that type representation of Future is the same. How do we see in it laziness as a core difference between both?
And also aside, cats IO is () => (A=>Unit) => Unit ?
Trying to understand it :)
Cheers

Fabio Labella

@SystemFw

@Garnek20 neither of the two is implemented exactly like that. One of the constructors (the one for async) has a shape similar to that. I'm not really aware (contributors please correct me if wrong) of a detailed explanation of Monix internals specifically. I touch on some aspects in my talk on Fibers (they are for an abstract IO, so most things should be relevant), so I would recommend starting from there

Second talk on systemfw.org/talks

if you delve into the implementation of both IO/Task and Future, you will see the difference boils down to: when receiving a block of code, IO/Task store it in suspended form in a data structure, and it does not get evaluated until interpretation, whereas Future sends it to a thread pool almost straight away (in the implementation of transform), and that's the bit that breaks Referential Transparency for Future. Unfortunately in both implementations you need to jump through some hoops (various methods) to get to the exact place

Paweł Gontarz

@Garnek20

@SystemFw exactly Async constructor is the closest one I found too. Thanks for your answer!
What is odd to me is that we project type A to Unit but IO or Task construction of the effect will return A anyway.

I'll take a look at your slides! And about cancellation of the effects, what are the strategies? Interrupting threads or removing tasks from thread pool queue (if interruption not supported)? Do I get it correctly?

Oleg Pyzhcov

@oleg-py

@Garnek20 cats-effect compatible types usually stop running next steps after some condition (e.g. after an async boundary or every several flatMaps).

But we also require effect types to provide a cancelable builder that can execute arbitrary user-supplied logic for cancellation. It can be used, for instance, to make a call with Java thread interruption, or lift CompletableFuture from Java 8 with its cancel logic.

Fabio Labella

@SystemFw

@Garnek20 I'd recommend to watch the talk, I go into this. The slides are really not meant to be consumed standalone

interruption is supported in the way Oleg says :)

Paweł Gontarz

@Garnek20

@SystemFw I indeed went this way! Could you also tell me wether monix Task applies the same strategy for abstracting JVM threads into sort of Fibers to achieve non-blocking execution?

Fabio Labella

@SystemFw

it does

Paweł Gontarz

@Garnek20

@SystemFw thanks!

Piotr Gawryś

@Avasil

Hello everyone, I've been working on bifunctor implementation for Monix Task lately (sorry for ignoring issues on monix/monix :'D I needed to focus) and I'm ready to share my progress: https://github.com/monix/monix-bio

https://twitter.com/p_gawrys/status/1204507365485293570

TapanVaishnav

@TapanVaishnav

:+1:

Kasper Kondzielski

@ghostbuster91

Hi guys, how do I restart monix observable with backoff?
I found that there is

  final def onErrorRestart(maxRetries: Long): Observable[A] = {
    require(maxRetries >= 0, "maxRetries should be positive")
    new OnErrorRetryCountedObservable(self, maxRetries)
  }

I think that it will be sufficient to add .delay on selft but unfortunately OnErrorRetryCountedObservable is package private. Is there any other way to do that?

Alexey Kotlyarov

@koterpillar

Hello everyone, couldn't Google an answer for this, so: is there something built-in that will update a Ref[A] for me whenever my Observable[A] changes?

Piotr Gawryś

@Avasil

@ghostbuster91 I usually write simple recursive function, e.g. https://monix.io/docs/3x/reactive/observable.html#retrying-with-delay

@koterpillar What do you mean by changes? If it's for all elements then you can do observable.doOnNext(el => ref.update(...))

Kasper Kondzielski

@ghostbuster91

@Avasil awesome, thanks!

Alexey Kotlyarov

@koterpillar

@Avasil yes, all elements. Looks workable, thank you!

Kasper Kondzielski

@ghostbuster91

After having a deeper look on the example I think that this is not something what I want.
This backoff will grow indefinitely with each error. What I would like is to reset the backoff to the initialDelay after emitting single item.

Piotr Gawryś

@Avasil

A bit ugly but should work:

  def retryBackoffNew[A](source: Observable[A],
                         maxRetries: Int, firstDelay: FiniteDuration): Observable[A] = {
    @volatile var resetDelay = false

    def loop(source: Observable[A], retriesLeft: Int, currentDelay: FiniteDuration): Observable[A] =
      source.onErrorHandleWith {
        case ex: Exception =>
          if (retriesLeft > 0) {
            val newDelay = if (resetDelay) {
              resetDelay = false
              firstDelay
            } else {
              currentDelay
            }

            loop(source, retriesLeft - 1, newDelay * 100)
              .delayExecution(newDelay)
          }
          else
            Observable.raiseError(ex)
      }

    loop(source.doOnStart(_ => Task {resetDelay = true}), maxRetries, firstDelay)

Perhaps there is a better way to write this. Other option is custom operator like OnErrorRetryCountedObservable

Kasper Kondzielski

@ghostbuster91

Thanks, yes I also saw this operator, but it seems not so trivial to write another one :)

Luka Jacobowitz

@LukaJCB

@Avasil Couldn’t you just use cats-retry?

Piotr Gawryś

@Avasil

Yeah, custom operators require some understanding of internals and the protocol is dirty :D I'm doing a talk on this in March so I hope to make it easier

I havent used cats-retry so I don't know if it supports Observable (it doesnt have Sync)

Luka Jacobowitz

@LukaJCB

cats-retry only needs MonadError :)

Kasper Kondzielski

@ghostbuster91

Yeah, custom operators require some understanding of internals and the protocol is dirty :D I'm doing a talk on this in March so I hope to make it easier

You mean on scalar-conf?

Vasily Shiyan

@xicmiah

@Avasil I'm curiuous about the original name for BIO from monix-bio. I mean I get what it's referencing, but would you care to give more context?

Luka Jacobowitz

@LukaJCB

@xicmiah I think it’s my fault, I called it BIO out of lack of creativity

Viktor Lund

@vilu

What would be the idiomatic way of handling errors without recovery?

If a task succeeds I want to asynchronously do something (send off some analytics) and if it fails I want to asynchronously do something else

Luka Jacobowitz

@LukaJCB

@vilu flatMap for the happy path and handleErrorWith on the exception path

Viktor Lund

@vilu

for {
    ...
    response <- client
        .doSomething(request).onError(recordFailedRequest)
      _              = recordSuccesfulRequest(response)    
    ...
} yield response

Ok, I was wondering if there was something that did not require to keep the types.

Then I wasn't too far off.

Piotr Gawryś

@Avasil

@LukaJCB @xicmiah Yep, it's not set in stone yet though, I have an issue for a discussion: monix/monix-bio#6
I lack creativity as well but I didn't want it to block me :D

@vilu There is also redeemWith that does both. If you don't want it to change the type, you can use guaranteeCase, it will also work for cancelation

Oh wait, guaranteeCase won't have access to the value. There's also continual https://github.com/typelevel/cats-effect/blob/master/core/shared/src/main/scala/cats/effect/Concurrent.scala#L392 which should be available via cats-effect

Piotr Gawryś

@Avasil

Going back to BIO - the technique is often called Bifunctor IO since it was announced (e.g. if you google it there will be articles from Mayb 2018 when John and Alex were arguing about it :D)

I wish it could have been called Task because that's what I think about it when I hear Monix but I feel like it is more important to have a simple migration and keep it an alias to BIO[Throwable, A]

xplosunn

@xplosunn

The monix-bio repository mentiones talking to @Avasil here on gitter, so two questions:

Should monix-bio have it's own gitter channel or maybe add the link to this one there?
Should I mention on an issue that I'm giving it a go or is it alright if I just attempt it with no compromise and open up a PR if I finish it?

Hi, btw =)

Piotr Gawryś

@Avasil

Hello @xplosunn !

Right, I think I didnt provide the link, I will update it tomorrow morning and consider separate channel

I guess it is fine but it is usually a good idea to post there to minimize the chances that someone else is also working on it

It's fine if it turns out that you wont finish it

Piotr Gawryś

@Avasil

I created a separate channel: https://gitter.im/monix/monix-bio

Where communities thrive

People

Repo info

Activity