typelevel/cats-effect

Most discussion is on Typelevel Discord: https://discord.gg/bSQBZA3Ced

Oleg Pyzhcov

@oleg-py

Deferred sounds like a good approach, and if you want to be able to concurrently cancel the acquisition somehow, you'd need Concurrent.cancelable builder

The Async doesn't let you "run many of them" out of the box - if your underlying API is thread-blocking, you can't get around with thread management, but you should be able to only use Sync[F] + ContextShift[F] + Blocker

You might need to have Concurrent on the outside, to manage two slots independently, but that is irrelevant

Oleg Pyzhcov

@oleg-py

Async[F] only means "I can lift anything that is calling a callback once, that also doesn't care much about cancelability, into F".

So, it depends on the API you're wrapping - and if you're making your own, Concurrent is probably better since you get cancelable version of Deferred when you have it

vimalaguti

@vimalaguti

Thanks!

I'll try with Concurrent and Deferred - although I don't mind about cancelability, just concurrency.

vimalaguti

@vimalaguti

Another question: in the Ref page, they use Ref.of[IO, Int]. Is IO replaceable with Concurrent or it requires a similar class like SyncIO or Eval?

Ryan Peters

@sloshy

IO is replaceable with any type that implements Sync, so you can write:

def giveMeRef[F[_]: Sync] = Ref.of[F, Int](1)

Gabriel Volpe

@gvolpe

Concurrent is a typeclass. Ref.of takes any data type that has an instance of Sync, so SyncIO is included.

Ryan Peters

@sloshy

Ah right sync (edited my previous comment)

Gabriel Volpe

@gvolpe

:+1: :)

Ryan Peters

@sloshy

I need to slow down I'm answering questions faster than my brain works this morning

vimalaguti

@vimalaguti

thanks!

Rolando Manrique

@rolandomanrique

Hi, I'm trying to use tuple parMapN syntax but I can't make it work with parameterized effect, this is a very simple representation of what I'm trying to do:

scala> :paste
// Entering paste mode (ctrl-D to finish)

  import cats.effect._
  import cats.implicits._
  import cats.effect.implicits._
  import scala.concurrent.ExecutionContext
  val cs = IO.contextShift(ExecutionContext.global)

  def parMapIO(implicit cs: ContextShift[IO]) = {
    val ioA = IO("A")
    val ioB = IO(10)
    val ioC = IO(false)
    (ioA, ioB, ioC).parMapN { (a, b, c) => println(s"done: $a $b $c") }
  }

scala> parMapIO(cs).unsafeRunSync
done: A 10 false

scala> :paste
// Entering paste mode (ctrl-D to finish)


  def parMapF[F[_]](implicit F: ConcurrentEffect[F], cs: ContextShift[F]) = {
    val ioA = F.delay("A")
    val ioB = F.delay(10)
    val ioC = F.delay(false)
    (ioA, ioB, ioC).parMapN { (a, b, c) => println(s"$a > $b > $c") }
  }

// Exiting paste mode, now interpreting.

<pastie>:25: error: could not find implicit value for parameter p: cats.NonEmptyParallel[F,F]
    (ioA, ioB, ioC).parMapN { (a, b, c) => println(s"$a > $b > $c") }
                            ^
<pastie>:21: warning: parameter value F in method parMapF is never used
def parMapF[F[_]](implicit F: ConcurrentEffect[F], cs: ContextShift[F]) = {
                           ^
<pastie>:21: warning: parameter value cs in method parMapF is never used
def parMapF[F[_]](implicit F: ConcurrentEffect[F], cs: ContextShift[F]) = {
                                                   ^

Oleg Pyzhcov

@oleg-py

@rolandomanrique you need Parallel constraint on your F to use .par methods

Loránd Szakács

@lorandszakacs

did someone write a cats-effect wrapper over better-files? :smile:

while I love fs2-io, sometimes I need more :sweat:

Piotr Gawryś

@Avasil

haha, the same topic just came up in few different places

Fabio Labella

@SystemFw

@lorandszakacs functional-streams-for-scala/fs2#1533

Piotr Gawryś

@Avasil

it would be nice to have common files library on top of cats-effect and then specialized modules for streaming

Wojtek Pituła

@Krever

totally agree with Piotr as much as it matters :D

Piotr Gawryś

@Avasil

Having it in fs2 is not that bad considering it's easy to have this dependency because of doobie/http4s but it can be harder to discover/use for Monix/ZIO users. Also many utilities are not related to streaming at all. I would be happy to contribute to something like it and bring fresh contributors but I would count on someone else (TM) to kickstart it :D

Fabio Labella

@SystemFw

the same could be said for monix-catnap though

Piotr Gawryś

@Avasil

monix-catnap doesn't have a dependency on Task / Observable / Iterant but that's still a good point

Loránd Szakács

@lorandszakacs

@SystemFw nice, thanks :D

Fabio Labella

@SystemFw

monix-catnap doesn't have a dependency on Task...

ah, that's true

Daniel Spiewak

@djspiewak

I’ve often thought about doing something like a cats-effect-jvm and cats-effect-js library

Maybe even merging them together into a cats-effect-platform

Like, there isn’t much point in having a file read or write function without streaming effects, but other things like moving, copying, listing, etc, not to mention more basic stuff like println and exit hooks, not to mention browser things, all could be nicely wrapped

Ryan Peters

@sloshy

I'm trying to build a concurrent variable like structure that resets itself after nobody is using it for some period of time. So multiple subscribers can "acquire" it, and when all have released I set off a timer that, when finished, resets the variable to an initial value provided nobody subscribes during that window. I have the timer part handled with https://christopherdavenport.github.io/agitation/, and I'm thinking semaphore would be good for figuring out when the number of users that have acquired the variable is 0. Semaphore does not have an "unbounded" variant, and I'm wondering whether or not such a thing would even be useful, or if there is a simpler way of expressing what it is I'm going for (that may or may not need semaphore). Maybe Long.MaxValue as the number of permits is a fine way of getting what I need. What are your thoughts?

Fabio Labella

@SystemFw

and I'm thinking semaphore would be good for figuring out when the number of users that have acquired the variable is 0

if you don't need any semantic blocking, just a Ref should be enough

Ryan Peters

@sloshy

I'll try doing it with just Ref then, I think that should work fine. Thanks for the suggestion @SystemFw

Gavin Bisesi

@Daenyth

@sloshy you might also consider chris d's agitation, it has things like resettable timeouts

Ryan Peters

@sloshy

Way ahead of you @Daenyth :D

Gavin Bisesi

@Daenyth

oh wow

yeah you linked to it

haha

excuse me while I go have more coffee :grin:

Ryan Peters

@sloshy

TBF it was a hyperlink, easy to skim right over

And not to the github but to the microsite since it's what I had open at the moment

Fabio Labella

@SystemFw

agitate seems leaky rn (if you cancel agitate your reset is still happening)

also I think there is a race between oldFiber.cancel and buzzer.complete but it's very very likely innocuous

bifunctor

@bifunctor

Hi all
Is there a way to integrate cats effect into akka streams in the sink position?
For example:

    implicit val system = ActorSystem("QuickStart")
    implicit val materializer = ActorMaterializer()
    implicit val executor = system.getDispatcher

    val ticker = Source.tick(0.second, 1.second, "H")
      .runWith(Sink.foreach(println))

the type ticker of the ticker is Future[Done]

Ryan Peters

@sloshy

@bifunctor Are you asking about running IO inside Sink.foreach or are you looking at making the type of ticker be IO[Done]?

bifunctor

@bifunctor

I am looking the run inside the foreach

I think, I have to create Graphstage

but there is a good example here https://github.com/krasserm/streamz/blob/master/streamz-converter/src/main/scala/streamz/converter/Converter.scala

Daniel Spiewak

@djspiewak

@bifunctor I think that fs2-reactive-streams is one answer to this. at least it seems like it should be. I don't know much about akka streams.

Where communities thrive

People

Repo info

Activity