0.9.0 - support fs2v3

build.sbt

@@ -3,7 +3,7 @@ name := "xs4s-root"
 ThisBuild / scalaVersion := "2.13.6"
 ThisBuild / crossScalaVersions := Seq("2.12.12", "2.13.6", "3.0.1")
 ThisBuild / organization := "com.scalawilliam"
-ThisBuild / version := "0.8.8"
+ThisBuild / version := "0.9.0"
 ThisBuild / resolvers += Resolver.JCenterRepository
 
 lazy val root = (project in file("."))

fs2v3/src/main/scala/xs4s/fs2compat/Fs2Syntax.scala

@@ -0,0 +1,47 @@
+package xs4s.fs2compat
+
+import _root_.fs2.{Pipe, Stream}
+import cats.effect.{Resource, Sync}
+import xs4s.XmlElementExtractor
+import xs4s.generic.Scanner
+import xs4s.syntax.core._
+
+import javax.xml.stream.XMLEventReader
+import javax.xml.stream.events.XMLEvent
+import scala.language.higherKinds
+
+/**
+  * Utilities to enhance xs4s interaction with FS2
+  */
+trait Fs2Syntax {
+
+  implicit class RichScanner[In, State, Out](scanner: Scanner[In, State, Out]) {
+
+    /** Create an FS2 Pipe from the scanner */
+    def fs2Pipe[F[_]]: Pipe[F, In, Out] =
+      _.scan(scanner.initial)(scanner.scan).map(scanner.collect).unNone
+  }
+
+  implicit class RichFs2StreamObj(obj: Stream.type) {
+
+    /** Create an XMLEvent Stream from an XMLEventReader */
+    def xmlEventStream[F[_]: Sync](
+        xmlEventReader: Resource[F, XMLEventReader],
+        chunkSize: Int = 1): Stream[F, XMLEvent] =
+      Stream
+        .resource(xmlEventReader)
+        .flatMap(
+          reader =>
+            Stream
+              .fromBlockingIterator[F]
+              .apply[XMLEvent](reader.toIterator, chunkSize))
+  }
+
+  implicit class RichXmlElementExtractor[O](
+      xmlElementExtractor: XmlElementExtractor[O]) {
+    def toFs2PipeThrowError[F[_]]: Pipe[F, XMLEvent, O] =
+      xmlElementExtractor.scannerThrowingOnError.fs2Pipe[F]
+    def toFs2PipeIncludeError[F[_]]: Pipe[F, XMLEvent, Either[Throwable, O]] =
+      xmlElementExtractor.scannerEitherOnError.fs2Pipe[F]
+  }
+}

fs2v3/src/main/scala/xs4s/fs2compat/package.scala

@@ -0,0 +1,32 @@
+package xs4s
+
+import cats.effect.{Async, Resource, Sync}
+import fs2._
+import xs4s.syntax.fs2._
+
+import javax.xml.stream.XMLInputFactory
+import javax.xml.stream.events.XMLEvent
+import scala.language.higherKinds
+
+package object fs2compat {
+
+  /**
+    * Turns an FS2 Byte Stream into a stream of XMLEvent.
+    * It turns the source stream into an input stream,
+    * then creates a reader, and then creates a further stream
+    * from the Iterator that was obtained.
+    * */
+  def byteStreamToXmlEventStream[F[_]: Async](
+      xmlInputFactory: XMLInputFactory = defaultXmlInputFactory,
+      chunkSize: Int = 1)(implicit F: Sync[F]): Pipe[F, Byte, XMLEvent] =
+    byteStream =>
+      Stream
+        .resource(io.toInputStreamResource(byteStream))
+        .flatMap(
+          inputStream =>
+            Stream.xmlEventStream(
+              Resource.make(
+                F.delay(xmlInputFactory.createXMLEventReader(inputStream)))(
+                xmlEventReader => F.delay(xmlEventReader.close())),
+              chunkSize))
+}

fs2v3/src/main/scala/xs4s/syntax/fs2.scala

@@ -0,0 +1,3 @@
+package xs4s.syntax
+
+object fs2 extends xs4s.fs2compat.Fs2Syntax {}

fs2v3/src/test/scala/xs4s/fs2compat/Fs2CompatSpec.scala

@@ -0,0 +1,41 @@
+package xs4s.fs2compat
+
+import cats.effect.IO
+import cats.effect.unsafe.implicits.global
+import fs2.Stream
+import org.scalatest.freespec.AnyFreeSpec
+import xs4s.XmlElementExtractor
+import xs4s.syntax.fs2.RichXmlElementExtractor
+
+import scala.xml.Elem
+
+final class Fs2CompatSpec extends AnyFreeSpec {
+  "It works" in {
+    val input =
+      s"""
+         |<items>
+         |<embedded><item>Embedded</item></embedded>
+         |<item>General</item>
+         |<embedded-twice><embedded-once><item>Doubly embedded</item></embedded-once></embedded-twice>
+         |<item><item>Nested</item></item>
+         |</items>
+         |
+       """.stripMargin
+
+    val anchorElementExtractor: XmlElementExtractor[Elem] =
+      XmlElementExtractor.filterElementsByName("item")
+
+    val textStream: Stream[IO, String] = fs2.Stream
+      .apply[IO, String](input)
+      .flatMap(str => fs2.Stream.emits(str.getBytes().toList))
+      .through(byteStreamToXmlEventStream[IO]())
+      .through(anchorElementExtractor.toFs2PipeThrowError)
+      .map(_.text)
+
+    assert(
+      textStream.compile.toList.unsafeRunSync() == List("Embedded",
+                                                        "General",
+                                                        "Doubly embedded",
+                                                        "Nested"))
+  }
+}