XML streaming hangs when using `fs2.data.xml.xpath.filter.raw`

[mrdziuban]

I might be misunderstanding how to use fs2.data.xml.xpath.filter.raw, but my first attempt led to my code hanging forever.

Using a simple XML file, test.xml:

<?xml version='1.0' encoding='UTF-8'?>
<data>
  <test>text</test>
</data>

and running this code with scala-cli /path/to/test.scala:

//> using scala 3.6.4
//> using dep co.fs2::fs2-core:3.11.0
//> using dep co.fs2::fs2-io:3.11.0
//> using dep org.gnieh::fs2-data-xml:1.11.2

import cats.effect.{IO, IOApp}

object Test extends IOApp.Simple {
  def run: IO[Unit] =
    fs2.data.xml.xpath.XPathParser[IO]("/data/test").flatMap(xpath =>
      fs2.io.file.Files[IO].readAll(fs2.io.file.Path("/path/to/test.xml"))
        .through(fs2.text.utf8.decode)
        .through(fs2.data.xml.events[IO, String]())
        .through(fs2.data.xml.xpath.filter.raw(xpath))
        // This `debug` call prints `Stream(..)`
        .debug()
        // Nothing is ever printed by this `debug` call
        .evalMap(_.debug().compile.drain)
        .compile
        .drain
    )
}

The first .debug() call on the main Stream prints Stream(..), but the .debug().compile.drain calls on the inner Stream don't seem to run, and the code never terminates.

[mrdziuban]

Interestingly if I change evalMap to parEvalMapUnbounded then it works, as does parEvalMap with any maxConcurrent argument greater than 1 (since fs2 calls evalMap if you pass 1)

[satabin]

Hi. Thanks for reporting. I will try it locally and let you know.

[satabin]

Hey I think I understand what the problem is.
With evalMap when the outer stream is pulled, on first matching element an inner stream is created. And this inner stream is compiled and drained, which will only return once the inner stream is closed. The inner stream is closed, when the match region is over. This means that the inner stream is waiting for more elements, which are pulled from the outer stream.

However, the outer stream is not pulled until the inner stream is finished, since evalMap will not pull until the effect has returned. So you have a stream that hangs.

For this raw operator to work properly, inner stream(s) must be consumed parallelly to the outer stream, which is not the case with evalMap. Depending on your use case you can use a higher level filter operator. For instance if you are only interested in the first match, you can use first:

    fs2.data.xml.xpath.XPathParser[IO]("/data/test").flatMap(xpath =>
      fs2.io.file.Files[IO].readAll(fs2.io.file.Path("/path/to/test.xml"))
        .through(fs2.text.utf8.decode)
        .through(fs2.data.xml.events[IO, String]())
        .through(fs2.data.xml.xpath.filter.first(xpath))
        // This `debug` call prints the element composing the first match
        .debug()
        .compile
        .drain

The raw operator is tricky to use properly, but the general guidelines are:

• make sure to consume all emitted inner streams
• make consumption happen in parallel

This is why I added the other combinators, to make it a bit easier.

• filter.first(xpath) is a Pipe returns the flat stream of the first match
• filter.collect(xpath, collector) aggregates each inner stream with the provided collector (could be for instance filter.collect(xpath, List) to aggregate each match in a list and emit those lists)
• filter.dom[Node](xpath) build the DOM for each match for type Node that has a DocumentBuilder in scope.

All these operators are safer to use and should always be preferred. I will rework the documentation to not make the raw operator the first thing mentioned, and add more warning in its usage.

Maybe an operator is missing here. What is your use case?

[mrdziuban]

I see, thanks for debugging and for the explanation! I don't have a solid reason to use raw, I was just testing out the library and hit this.

Maybe to avoid the footgun raw could be rewritten to not even give users the ability to consume the inner streams sequentially? e.g.

def raw[A](
  path: XPath,
  maxConcurrent: Int,
  process: Stream[F, XmlEvent] => F[A],
  maxMatch: Int = Int.MaxValue,
  maxNest: Int = Int.MaxValue
)(implicit F: Concurrent[F]): Pipe[F, XmlEvent, A] =
  new XmlQueryPipe(compileXPath(path)).raw(maxMatch, maxNest)(_).parEvalMap(maxConcurrent)(process)

Though I suppose there's still the possibility that a user passes 1 for maxConcurrent...

[satabin]

What you are proposing here is more or less filter.collect. I have the raw one because it does not make assumption on how you will consume the result, and offers full flexibility offered by fs2. But it comes at a cost of being not so safe. I think I will follow the convention of adding unsafe to the name to mark it for what it is.

Thanks a lot for reporting this issue, I need to work on the documentation and usability.

For the v2 plan, I plan on improving UX of the various APIs, to have a more natural higer level default API, easier to use, and safer, this is definitely something to add to this API rework.