monix/monix

Asynchronous, Reactive Programming for Scala and Scala.js

toxicafunk

@toxicafunk

is that what sleuth and dapper and zipkin does?

if so, I don't think there's nothing relevant/specific to monix

Alejandro Gómez

@purrgrammer

hi folks, I'm finding a problem when trying to get a ContextShift[Task] from a Scheduler instance. If i understood correctly, you can do Scheduler#timer and Scheduler#contextShift provided that the right implicits are imported. However, when trying to create a ContextShift from a scheduler instance (https://github.com/47deg/fetch/blob/4328bc61627eaf580624ec825cd63bbe17526659/examples/src/test/scala/MonixExample.scala#L44-L46) I get the following error: value contextShift is not a member of monix.execution.Scheduler

not sure what I'm doing wrong, any input would be appreciated

Piotr Gawryś

@Avasil

It's not in 3.0.0-RC1 but you can try 3.0.0-RC2-d0feeba with cats-effect 1.0.0-RC3

Alejandro Gómez

@purrgrammer

@Avasil oh, my bad then, thanks for pointing me in the right direction

Alexandru Nedelcu

@alexandru

Shit, I really need to publish something fast.

Mateusz Górski

@goral09

That's the price of a success @alexandru :)

Igor Ramazanov

@igor-ramazanov

Hey guys! I'm working on a personal pet-project learning final-tagless and functional programming. Usually I was using Monix through its direct API.
Currently, I'm trying to abstract over the Task, so choosing of the effect monad will be deferred until the end of the world.
I've faced with a problem of being unable to express some functionality using cats-effect:

Task(...).onErrorRestartLoop(MaxRetries) { (ex, retries, retry) =>
        if (retries > 0) {
          retry(retries - 1).delayExecution(1.second)
        } else {
          onError(ex)
          Task.raiseError(ex)
        }
      }

This is a simple retries mechanism with a retries number threshold and delayed execution of retries.

Gabriel Volpe

@gvolpe

@themirrortruth I'm not super familiar with monix Task but more with cats-effect. If you can tell me what is the type of retry I might be able to help you

Gabriel Volpe

@gvolpe

I believe it does something like this:

import cats.MonadError
import cats.effect.Timer
import cats.syntax.all._

import scala.concurrent.duration._

def retry[F[_], A](fa: F[A])(retries: Int)(implicit F: MonadError[F, Throwable], T: Timer[F]): F[A] =
  fa.handleErrorWith {
    case _ if retries > 0 =>
      T.sleep(1.second) >> retry(fa)(retries - 1)
    case e =>
      F.raiseError(e)
  }

Example of use:

// not sure if Task.raiseError exists, maybe Task.eval(throw exception) will do
val task = Task.eval(println("foo")) *> Task.raiseError(new Exception("boom"))
retry(task)(5)

Output:

foo
foo
foo
foo
foo
foo
java.lang.Exception: boom

Alexandru Nedelcu

@alexandru

@gvolpe @themirrortruth in addition to modelling your own loop, Monix also has onErrorRestartLoop, so you can do ...

task.onErrorRestartLoop(retries) { (e, retries, retry) =>
  if (retries > 0)
    Task.sleep(1.second).flatMap(_ => retry(retries - 1))
  else
    Task.raiseError(e)
}

New PR: monix/monix#726

Oleg Pyzhcov

@oleg-py

So it's catnap now :D

Alexandru Nedelcu

@alexandru

catnip was taken :)

Oleg Pyzhcov

@oleg-py

It's so cold and rainy outside, I could use a catnap myself

Alexandru Nedelcu

@alexandru

Yes, laziness wins 😛

Oleg Pyzhcov

@oleg-py

But not by default :smirk:

Alexandru Nedelcu

@alexandru

@themirrortruth for your question, @gvolpe's answer is correct ... onErrorRestartLoop is basically just a loop built with onErrorHandleWith, named handleErrorWith in Cats (which is flatMap for errors)

Gabriel Volpe

@gvolpe

:+1: :smiley:

Igor Ramazanov

@igor-ramazanov

@alexandru @gvolpe Thank you!

SemanticBeeng

@SemanticBeeng

Trying to use cats.data.Nested with Task based apis and getting stuck

      type Dataset = (DataSpecification, DataSetRef)

      val clients = Nested[Task, Either[DataLakeRepositoryError, ?], Seq[Client]](dlRegistry.listClients())

      def clientDatasets(client: String) =
        Nested[Task, Either[DataLakeRepositoryError, ?], Seq[Dataset]](dlRegistry.listDatasets(client))

      val r: Task[_] = (clients map { cls: Seq[Client] ⇒
        cls map { client ⇒
            (clientDatasets( client.id) map { dsds: Seq[Dataset] ⇒
              dsds map { ds ⇒
                ds._2.shortName.pp("\n\n"); ds
              }
            }).value
          }

        }).value

     s"r is ${await(r)}".pp

Gives

r is Right(List(Task.Map$995060245, Task.Map$958851644, Task.Map$119297310, Task.Map$713902690, Task.Map$1792185060, Task.Map$1688986248))

SemanticBeeng

@SemanticBeeng

Instead I'd like a single Task to runAsync for result.

Tried Task.sequence and other things but the result is always something else than desired.

For context about cats.data.Nested see https://typelevel.org/cats/datatypes/nested.html

You can glean from code above but apis are

def listClients(): Task[Either[DataLakeRepositoryError, Seq[Client]]] 
def listDatasets(clientId: String): Task[Either[DataLakeRepositoryError, Seq[(DataSpecification, DataSetRef)]]]

In some variants of this was getting instances of Task.Suspend and did not know what to do with them.

@oleg-py - thoughts please?

Alexandru Nedelcu

@alexandru

@SemanticBeeng what result do you expect?

Task.Suspend is internal, that's not what matters, but rather the type you're expecting.

SemanticBeeng

@SemanticBeeng

A Task that when runAsync-ed gives the entire/concatenated list of shortNames of traversed DatasetRefs for all the Clients

Fabio Labella

@SystemFw

I assume _.id: Client => String does the right thing

SemanticBeeng

@SemanticBeeng

yes. In fact. running the inner part for one Client gives such a result. (Task as per above)...

Fabio Labella

@SystemFw

btw you should have those things return List, and not Seq, or you won't be able to use many cats abstractions

SemanticBeeng

@SemanticBeeng

Ah.

Fabio Labella

@SystemFw

also, this would be easier with ListT (and therefore fs2.Stream, monix.Iterant, monix.Observable)

but the trick there is to imagine what you would do with F[List[Client]] and String => F[List(DataSpec, DataRef)]

so, first .flatMap to get to List[Client]

then traverse(c => listDataSets(c.id))

actually, flatTraverse

I'll write a snippet down

SemanticBeeng

@SemanticBeeng

thanks. parsing...

Fabio Labella

@SystemFw

yeah I'll just write the code

SemanticBeeng

@SemanticBeeng

This "rings" better than mine, the intuition being to not interrupt the flow with separate, disjoint, mapcalls. Right? Feels like Task create intermediate instances that need special action to run.

Where communities thrive

People

Repo info

Activity