You might need to have
Concurrent on the outside, to manage two slots independently, but that is irrelevant
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
I'll try with Concurrent and Deferred - although I don't mind about cancelability, just concurrency.
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]) = {
^
while I love fs2-io, sometimes I need more :sweat:
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
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
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?
yeah you linked to it
haha
excuse me while I go have more coffee :grin:
And not to the github but to the microsite since it's what I had open at the moment
also I think there is a race between
oldFiber.cancel and buzzer.complete but it's very very likely innocuous
Hi all
Is there a way to integrate cats effect into akka streams in the sink position?
For example:
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]
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
fs2-reactive-streams will get you a stream where multiple libraries form the connecting pieces of it, which might be what you want. Another alternative is to just do an unsafeRunSync inside of the foreach (not my first choice but it should "work"). Personally if it were me I would defer to fs2 as much as possible as it has direct support for cats effect .