functional-streams-for-scala/fs2

Compositional, streaming I/O library for Scala

mn98

@mn98

@tulth thanks, I'll take a look!

corentin

@corenti13711539_twitter

I'd like to add a rate limiting mechanism in a fs2 Stream. What's the idiomatic way to do that with fs2 v2.4?

Domas Poliakas

@domaspoliakas

.metered could be what you’re looking for

Lucas Satabin

@satabin

@corenti13711539_twitter depending on your needs, you might also have a look at upperbound if metered is not working for you https://github.com/SystemFw/upperbound

corentin

@corenti13711539_twitter

thanks! :thumbsup: What would be the major differences between these two approaches?

Lucas Satabin

@satabin

I would personally start with metered and friends from fs2 core, and if it is not working for my use case, then look at an extra dependency like upperbound

corentin

@corenti13711539_twitter

Sounds reasonable since ATM we'd only need basic rate limiting :thumbsup:

Lucas Satabin

@satabin

in the core library there is also .debounce which drops elements, depending on how you want your rate limiting to behave

if you want to create a stream that execute something at a given rate, you can also have a look at the constructor functions like Stream.fixedRate or Stream.fixedDelay

corentin

@corenti13711539_twitter

I'm processing a Kafka based database update stream with fs2-kafka, so we'd need to just limit in intake rate, but not drop anything.

Fabio Labella

@SystemFw

if you only need to limit one Stream, metered should be fine

corentin

@corenti13711539_twitter

hmm. It's a combined stream created as Stream(kafkaStream1, kafkaStream2, ...).parJoin(x).
If I invoke metered I guess rate limiting is applied globally across all inner streams, right?

corentin

@corenti13711539_twitter

Also, the documentation on parJoin states "maxOpen - Maximum number of open inner streams at any time". Does that really mean what it reads and not the number of streams that can be processed in parallel? If it's the former, then I guess I should typically just set it to the total number of inner streams. Just out of curiosity, what happens if maxOpen is smaller than the number of inner streams?

Gavin Bisesi

@Daenyth

It depends whether the inner streams terminate or not

if smaller and the inner streams never terminate, the "excess" inner streams never execute at all

if they do terminate, then it does what you'd expect - run N to completion at any one time. Once one terminates, start the next

corentin

@corenti13711539_twitter

:thumbsup: ok, so since in my case the inner streams stream data from different topics using parJoinUnbounded is probably the right combinator to use.

Gavin Bisesi

@Daenyth

yes - it's also not really unbounded there. The number of topics is bounded by your app structure and config

Sergey Torgashov

@satorg

Hi guys. I has been working on some quite sophisticated nested streams processing and found out that one of the way to approach the problem could be an evalFold method in Stream (which would look like the regular fold but would take (O2, O) => F[O2] function). But there's no such method in Stream although there're many other eval* counterparts for effectless methods.

I wonder, is it due to some conceptual concerns regarding such method or does evalFold just still await its hero to get implemented?

Gavin Bisesi

@Daenyth

Isn't there an evalScan?

yes. evalFold should be easy then

conceptually fold is just scan.last

even if you might implement it lower than that for efficiency

Sergey Torgashov

@satorg

Yes, indeed, thanks for the hint. I just was confused that there're a plenty of eval* methods but no evalFold nor evalFold (or evalReduce).

Sergey Torgashov

@satorg

To me it should be as easy as the following:

    def evalFold[F2[x] >: F[x], O2](z: O2)(f: (O2, O) => F2[O2]): Stream[F2, O2] = {

      def go(in: Stream[F2, O])(zz: O2): Pull[F2, INothing, O2] = {
        in.pull.uncons1.flatMap {
          case None => Pull.pure(zz)
          case Some((hd, tl)) =>
            Pull.eval(f(zz, hd)).flatMap { go(tl) }
        }
      }

      go(this)(z).flatMap(Pull.output1).stream
    }

I borrowed it from implementations of evalScan and regular fold, actually.

Rafi Baker

@BakerRafi_twitter

Is there a combinator to reduce a Stream to singleton Stream of Unit?

Fabio Labella

@SystemFw

@satorg it's usual problem of knowing where to stop

like, do we add eval versions of everything?

@BakerRafi_twitter why do you need it to be a singleton?

anyway you can do .drain ++ Stream(()) , or void.foldMonoid

Rafi Baker

@BakerRafi_twitter

Thanks. I have db insert that emits a Stream of ids which I don’t care to continue with next step. But having this in a for comp is making the downstream steps repeat for every id emitted.

Fabio Labella

@SystemFw

right

I would recommend against for in general

what you want to express there is ++ really

Gavin Bisesi

@Daenyth

yeah, for is often an antipattern with streams

Generally only useful for parts that could be done in Resource/F anyway

Ayoub

@I_am_ayoub_twitter

Hi, I am combining finch with fs2 and I have a Stream[IO, Array[Byte]] which I am trying to write using writeAll. But The compiler complains about diverging implicit expansion for type cats.effect.Sync[F] starting with method catsReaderWriteStateTSync in object Sync on the writeAll call.

Am I missing an import or something ?

import fs2.Stream
import fs2.io

import java.nio.file._
import cats.effect.{ContextShift, IO}
import concurrent.ExecutionContext.Implicits.global
import cats.effect.Blocker
import cats.effect.IO

object Utils {
  def writeToFile(p: Path, stream: Stream[IO, Array[Byte]]) = {
    implicit val contextShift: ContextShift[IO] = IO.contextShift(global)
    Stream.resource(Blocker[IO]).flatMap { blocker =>
      stream
        .through(
          io.file.writeAll(p, blocker, Seq(StandardOpenOption.CREATE, StandardOpenOption.WRITE))
        )
        .compile
        .drain
    }
  }
}

Michael Pilquist

@mpilquist

Just a limitation of inference I believe -- try io.file.writeAll[IO](p, ...)

Ayoub

@I_am_ayoub_twitter

hmm now I get

type mismatch; found   : fs2.Pipe[cats.effect.IO,Byte,Unit]
                           (which expands to)  fs2.Stream[cats.effect.IO,Byte] => fs2.Stream[cats.effect.IO,Unit]
               required: fs2.Stream[cats.effect.IO,Array[Byte]] => fs2.Stream[?,?]

Michael Pilquist

@mpilquist

ok that’s good, that’s a legitimate error

Lucas Satabin

@satabin

the writeAll pipe works on a stream of bytes, you will need to "flatten" the arrays. for instance before the through you could do .flatMap(Stream.emits(_)) or something similar

Ayoub

@I_am_ayoub_twitter

is writeAll limited to Stream[IO, Byte] ?

Michael Pilquist

@mpilquist

you’ll need to convert that stream from Stream[IO, Array[Byte]] to Stream[IO, Byte] first via stream.flatMap(arr => Stream.chunk(Chunk.bytes(arr)))

Ayoub

@I_am_ayoub_twitter

I worked, thank you ! :)

Daniel Capo Sobral

@dcsobral

I’m confused about the “output” part of a Pull. I see it used with uncons.flatMap, but I’m unsure of how its mechanics work in other cases. In particular, I have a List[Pull[F,Byte,Unit]], and calling .sequence on it yields Pull[F,Byte,List[Unit]]. Is that outputting all individual pulls in order? Can I turn it into a Pull[F, Byte, Unit]?

Ryan Zeigler

@rzeigler

Is anyone aware of some kind of pipe like Pipe[F, Byte, SomeKindOfHash] ?

Gavin Bisesi

@Daenyth

yeah

ish

gzip CRC check, kind of

not terribly hard to write using mapAccumulate

Where communities thrive

People

Repo info

Activity