remove the filtered line from the file using scala fs2 file streaming

Question

how to remove the filtered lines from the current streaming file using fs2 and get the count of filtered lines as the return type?

ex: If the old.txt contains strings separated by a newline (\n):

 john
 sam
 chen
 yval
 ....

and val myList = List("chen","yval").

def converter[F[_]](implicit F: Sync[F]): F[Unit] =
  io.file.readAll[F](Paths.get("testdata/old.txt"), 4096)
    .through(text.utf8Decode)
    .through(text.lines)
    .filter(s => myList.contains(s))//remove this from the old file and write to new file
    .intersperse("\n")
    .through(text.utf8Encode)
    .through(io.file.writeAll(Paths.get("testdata/new.txt")))
    .compile.drain

// at the end of the universe...
val u: Unit = converter[IO].unsafeRunSync()

Diego E. Alonso-Blas · Accepted Answer · 2018-07-12T23:36:59.730

You can use the observe method of the Stream class.

You are looking for a function def converter[F[_]: Sync]: F[Int], which produces a computation F[Int] whose result (of type Int) is the number of filtered lines, and whose effect is to write those lines to the output file. To follow a plumbing analogy, you want to feed your filtered stream to two outputs, one for the result, and one for the effect. You can do this using the function observe, defined as

def observe(sink: Sink[F, O])(implicit F: Effect[F], ec: ExecutionContext): Stream[F, O]

A Sink[F,O] is an alias for a function Stream[F, O] => Stream[F, Unit]. In your case, the sink is the part of your code that writes your filtered stream into the output file:

def writeToFile[F[_]: Sync]: Sink[F, String] = 
  _.intersperse("\n")
  .through(text.utf8Encode)
  .through(io.file.writeAll(Paths.get("testdata/new.txt")))

The other output is to reduce, or rather fold,

  def converter[F[_]](implicit F: Effect[F], ec: ExecutionContext): F[Int] = 
    io.file.readAll[F](Paths.get("testdata/old.txt"), 4096)
      .through(text.utf8Decode)
      .through(text.lines)
      .filter(s => myList.contains(s))
      .observe(writeToFile[F])
      .compile
      .fold[Int](0)( (c, _) => c+1)
}

Note: for this solution you need to restrict the type-class of F to be Effect, and you need to use an ExecutionContext. The fold is defined in the ToEffect class.

remove the filtered line from the file using scala fs2 file streaming

1 Answers1