search

nrinaudo / kantan.csv

CSV handling library for Scala
http://nrinaudo.github.io/kantan.csv/
Apache License 2.0
346 stars 36 forks source link

FS2 streams integration #207

Open vasiliybondarenko opened 4 years ago

vasiliybondarenko commented 4 years ago

It would be nice to have integration with FS2 streams or at least some low level API to help implementing this.

nrinaudo commented 4 years ago

I was under the impression that the iterator-like CsvReader was basically all you needed and plugging that into a stream was essentially a one liner. Was that incorrect?

soujiro32167 commented 3 years ago

I have a similar use case, where I get an fs2 stream (or ZIO stream for that matter) and want to transform it into CSV. The iterator in CsvReader does not encode effects, so I cannot use it. For example: 

val readFromDB[A]: Stream[IO, A] = ???

object csv {
  import fs2.{Pipe, Stream} 
  import kantan.csv.HeaderEncoder
  import kantan.csv._
  import kantan.csv.ops._

  def streamCsv[F[_], A: HeaderEncoder]: Pipe[F, A, String] =  s =>
      implicitly[HeaderEncoder[A]].header.fold[Stream[F, String]](Stream.empty)(h => Stream.emit(h.mkString(",") + "\r\n")) ++
      s.chunks
        .map { chunk =>
          chunk.toList.asCsv(rfc)
        }
}

val webServerResponse[A: HeaderEncoder]: Stream[IO, Byte] = readFromDB.through(streamCsv)