### Compile and Run JVM Example

Source: https://github.com/gnieh/fs2-data/blob/main/site/cookbooks/jq.md

Use this command to compile the JVM version of the example using sbt-assembly and then run the resulting fat jar with a sample JSON file and a JQ-like query.

```shell
$ sbt exampleJqJVM/assembly
$ java -jar examples/jqlike/.jvm/target/scala-2.13/jq-like.jar -q '.[] | { "full_name": .name, "language": .language }' -f site/cookbooks/data/json/sample.json
```

--------------------------------

### Compile and Run Scala.js Example

Source: https://github.com/gnieh/fs2-data/blob/main/site/cookbooks/jq.md

Use this command to compile the Scala.js version of the example using sbt and then run the resulting JavaScript file with Node.js, a sample JSON file, and a JQ-like query.

```shell
$ sbt exampleJqJS/fastLinkJS
$ node examples/jqlike/.js/target/scala-2.13/jq-like-fastopt/main.js -q '.[] | { "full_name": .name, "language": .language }' -f site/cookbooks/data/json/sample.json
```

--------------------------------

### Compile and Run Scala Native Example

Source: https://github.com/gnieh/fs2-data/blob/main/site/cookbooks/jq.md

Use this command to compile the Scala Native version of the example using sbt and then run the resulting native executable with a sample JSON file and a JQ-like query.

```shell
$ sbt exampleJqNative/nativeLink
$ examples/jqlike/.native/target/scala-2.13/jq-like-out -q '.[] | { "full_name": .name, "language": .language }' -f site/cookbooks/data/json/sample.json
```

--------------------------------

### Initialize CSV Stream

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/csv/generic.md

Sets up an input CSV string and converts it into an fs2 Stream for processing. This is a common starting point for CSV operations.

```scala
import fs2._
import fs2.data.csv._

val input = """i,s,j
1,test,2
,other,-3
""".stripMargin

val stream = Stream.emit(input).covary[Fallible]
```

--------------------------------

### Define Input Stream for JSON Parsing

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/json/libraries.md

Sets up a sample JSON input stream for use in subsequent examples. This stream contains two JSON objects.

```scala
import fs2.{Fallible, Stream}
import fs2.data.json._
import fs2.data.json.jsonpath._
import fs2.data.json.jsonpath.literals._

def input[F[_]] = Stream.emit("""{
  "field1": 0,
  "field2": "test",
  "field3": [1, 2, 3]
  }
  {
  "field1": 2,
  "field3": []
}""").covary[F]

val stream = input[Fallible].through(tokens)

val sel = jsonpath"$$.field3[*]"
```

--------------------------------

### Define XML Input Stream

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/xml/xpath.md

Example of creating an XML input stream using the xml interpolator. This sets up the data for subsequent XPath operations.

```scala
import cats.syntax.all._

import fs2._
import fs2.data.xml._

val stream = xml"""
<root>
  <a attr="value">
    <b>text</b>
    <a>
      <c><![CDATA[some other text]]></c>
    </a>
  </a>
</root>
"""
```

--------------------------------

### Run JSON Lines Example with Scala CLI (JVM)

Source: https://github.com/gnieh/fs2-data/blob/main/site/cookbooks/jsonlines.md

Command to execute the Scala CLI script for handling JSON Lines data on the JVM platform. It specifies the script path and the 'read' command with an input file.

```shell
scala-cli site/cookbooks/scripts/jsonlines.scala -- read site/cookbooks/data/jsonl/nested.jsonl
```

--------------------------------

### Example Test Case Verification with csv-spectrum

Source: https://github.com/gnieh/fs2-data/blob/main/csv/shared/src/test/resources/csv-spectrum/readme.md

Demonstrates how to use the loaded test data within a test suite. It logs the name of the test being run and asserts the equality of a parsed CSV to its JSON representation.

```javascript
vsr spectrum = require('csv-spectrum')
spectrum(function(err, data) {
  console.log('testing ' + data[0].name)
  t.equal(csv2json(data[0].csv), JSON.parse(data[0].json))
})
```

--------------------------------

### Run JSON Lines Example with Scala CLI (Native)

Source: https://github.com/gnieh/fs2-data/blob/main/site/cookbooks/jsonlines.md

Command to execute the Scala CLI script for handling JSON Lines data on the native platform. It utilizes the --native flag to target native compilation and includes the 'read' command with an input file.

```shell
scala-cli --native site/cookbooks/scripts/jsonlines.scala -- read site/cookbooks/data/jsonl/nested.jsonl
```

--------------------------------

### Run JSON Lines Example with Scala CLI (JavaScript)

Source: https://github.com/gnieh/fs2-data/blob/main/site/cookbooks/jsonlines.md

Command to execute the Scala CLI script for handling JSON Lines data on the JavaScript platform. It uses the --js flag to specify the target platform and includes the 'read' command with an input file.

```shell
scala-cli --js site/cookbooks/scripts/jsonlines.scala -- read site/cookbooks/data/jsonl/nested.jsonl
```

--------------------------------

### Read Bytes from a File

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/index.md

Illustrates how to read raw bytes from a file into an `fs2.Stream[IO, Byte]`. This is a common starting point for processing file data, especially textual formats that require decoding.

```scala
import cats.effect._

import fs2._
import fs2.io.file.{Files, Flags, Path}

Files[IO]
  .readAll(Path("/some/path/to/a/file.data"), 1024, Flags.Read)
  // perform your decoding, parsing, and transformation here
  .compile
  .drain
```

--------------------------------

### Consume and Save XML Matches with XPath

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/xml/xpath.md

Use `filter.consumer` to handle each XML match as a stream without emitting values. This example saves each match to a file, incrementing a counter.

```scala
import fs2.io.file.{Files, Path}

def saveXml(counter: Ref[IO, Int], events: Stream[IO, XmlEvent]): Stream[IO, Nothing] = 
  Stream.eval(counter.getAndUpdate(_ + 1)).flatMap { index =>
    events
      .through(render.raw())
      .through(Files[IO].writeUtf8(Path(s"match-$index.xml")))
  }

val program = 
  for {
    counter <- Ref[IO].of(0)
    _ <- stream
      .lift[IO]
      .through(filter.consume(path, saveXml(counter, _)))
      .compile
      .drain
  } yield ()

program.unsafeRunSync()

Files[IO].readUtf8(Path("match-0.xml")).compile.string.unsafeRunSync()
Files[IO].readUtf8(Path("match-1.xml")).compile.string.unsafeRunSync()
```

--------------------------------

### Wrap JSON Lines Stream as Top-Level Array

Source: https://github.com/gnieh/fs2-data/blob/main/site/cookbooks/jsonlines.md

Reads a JSON Lines file, processes it using `readJsonLines`, tokenizes the JSON, wraps the resulting stream into a single top-level JSON array, and then pretty-prints it. This example demonstrates reading a file and producing a formatted JSON array string.

```scala
val array = 
  Files[IO]
    .readAll(Path("site/cookbooks/data/jsonl/nested.jsonl"))
    .through(readJsonLines)
    .through(fs2.data.json.ast.tokenize)
    .through(fs2.data.json.wrap.asTopLevelArray)
    .through(fs2.data.json.render.prettyPrint(width = 35))

array
  .compile
  .string
  .unsafeRunSync()
```

--------------------------------

### Implement RowEncoder for Shapeless HList

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/csv/index.md

Define a `RowEncoder` for a shapeless `HList` to encode data into CSV rows positionally. This example encodes an `Option[Int]`, a `String`, and an `Int` into a `Row`.

```scala
import shapeless._

implicit object HListEncoder extends RowEncoder[Option[Int] :: String :: Int :: HNil] {
  def apply(input: Option[Int] :: String :: Int :: HNil): Row =
    Row(NonEmptyList.of(CellEncoder[Option[Int]].apply(input.head), input.tail.head, input.tail.tail.head.toString))
}

// .tail drops the header line
val row = Stream(Option(3) :: "test" :: 42 :: HNil)
row.through(lowlevel.encode).compile.toList
```

--------------------------------

### Implement RowDecoder for Shapeless HList

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/csv/index.md

Create a `RowDecoder` for a shapeless `HList` to decode CSV rows based on field positions. This example decodes an `Option[Int]`, a `String`, and an `Int` from the row.

```scala
import shapeless._

implicit object HListDecoder extends RowDecoder[Option[Int] :: String :: Int :: HNil] {
  def apply(row: Row): DecoderResult[Option[Int] :: String :: Int :: HNil] =
    if(row.values.size < 3)
      Left(new DecoderError("row is too short"))
    else
      for {
        i <- if(row.values.head.isEmpty) Right(None) else CellDecoder[Int].apply(row.values.head).map(Some(_))
        s <- CellDecoder[String].apply(row.values.tail.head)
        j <- CellDecoder[Int].apply(row.values.tail.tail.head)
      } yield i :: s :: j :: HNil
}

// .tail drops the header line
val hlists = noh.tail.through(lowlevel.decode[Fallible, Option[Int] :: String :: Int :: HNil])
hlists.compile.toList
```

--------------------------------

### Define and Encode Scala Case Classes as JSON Events

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/json/index.md

Defines a sealed trait `Event` with case classes and provides a Circe encoder for them. This setup is necessary for generating JSON streams from these case classes.

```scala
sealed trait Event
case class CreateCounter(name: String, initialValue: Int) extends Event
case class RemoveCounter(name: String) extends Event
case class IncreaseCounter(name: String) extends Event

object Event {
  import _root_.io.circe.Encoder
  import _root_.io.circe.generic.extras.Configuration
  import _root_.io.circe.generic.extras.semiauto._
  implicit val configuration = Configuration.default.withDiscriminator("type")

  implicit val encoder: Encoder[Event] = deriveConfiguredEncoder
}
```

--------------------------------

### Initialize JSON Stream

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/json/jq.md

Sets up an initial JSON string and converts it into a stream of tokens. Requires fs2 and fs2-data-json libraries.

```scala
import cats.effect.SyncIO
import cats.syntax.all._

import fs2._
import fs2.data.json._

val input = """
{
              |"field1": 0,
              |"field2": "test",
              |"field3": [1, 2, 3]
              |}"""

val stream = Stream.emit(input).through(tokens[SyncIO, String])
```

--------------------------------

### Initialize Pagefind UI

Source: https://github.com/gnieh/fs2-data/blob/main/site/helium/templates/mainNav.template.html

Initializes the Pagefind UI component on the DOMContentLoaded event. Configure options like showing sub-results or images.

```javascript
window.addEventListener('DOMContentLoaded', (event) => { new PagefindUI({ element: "#search", showSubResults: true, showImages: false }); });
```

--------------------------------

### Handle invalid CBOR item stream with toBinary

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/cbor/index.md

The `toBinary` pipe fails if the input stream is invalid, such as missing elements in an array. This example demonstrates such a case.

```scala
val invalid =
  Stream
    .emits(List(
      CborItem.StartArray(3),
      CborItem.TextString("an"),
      CborItem.TextString("array")))

invalid
  .through(toBinary[Fallible])
  .compile
  .drain
```

--------------------------------

### Stream indefinite arrays/maps in serialized CBOR

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/cbor/index.md

Control whether arrays or maps are streamed in the serialized form by setting the `indefinite` flag to `true`. This example demonstrates making top-level arrays streamed.

```scala
valueStream
  .map {
    // make top-level arrays streamed
    case CborValue.Array(elements, _) => CborValue.Array(elements, true)
    case value                        => value
  }
  .through(toItems)
  .compile
  .toList
```

--------------------------------

### Deserialize Byte Stream to MsgpackItem

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/msgpack/index.md

Use the `fromBinary` pipe to convert a stream of `Byte`s into a stream of `MsgpackItem`s. This process is the inverse of serialization, excluding extension types.

```scala
val itemStream = binaryStream.through(data.msgpack.low.fromBinary[Fallible])

itemStream.compile.toList == inputStream.compile.toList
```

--------------------------------

### Define JSON Input and Token Stream

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/json/jsonpath.md

Sets up a multi-line JSON string and converts it into a stream of tokens using fs2-data-json. Ensure fs2 and cats-syntax are imported.

```scala
import cats.syntax.all._

import fs2._
import fs2.data.json._

val input = """
{
              |  "field1": 0,
              |  "field2": "test",
              |  "field3": [1, 2, 3]
              |}
              |{
              |  "field1": 2,
              |  "field3": []
              |}"""

val stream = Stream.emit(input).through(tokens[Fallible, String])
```

--------------------------------

### Deserialization from Binary

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/msgpack/index.md

The `fromBinary` pipe converts a stream of `Byte`s into a stream of `MsgpackItem`s. This is generally the inverse of the serialization process.

```APIDOC
## fromBinary

### Description
Converts a stream of `Byte`s into a stream of `MsgpackItem`s. This process is the inverse of serialization, excluding extension types.

### Method
Pipe

### Endpoint
N/A (Pipe)

### Parameters
N/A (Pipe)

### Request Example
```scala
val itemStream = binaryStream.through(data.msgpack.low.fromBinary[Fallible])

itemStream.compile.toList == inputStream.compile.toList
```

### Response
#### Success Response
- `MsgpackItem`: A stream of MessagePack items.

#### Response Example
(Stream of MsgpackItems)
```

--------------------------------

### Serialization to Binary

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/msgpack/index.md

The `toBinary` pipe serializes a stream of `MsgpackItem`s into a binary stream of `Byte`s. It includes stream validation.

```APIDOC
## toBinary

### Description
Serializes a stream of `MsgpackItem`s into a binary stream of `Byte`s, performing stream validation.

### Method
Pipe

### Endpoint
N/A (Pipe)

### Parameters
N/A (Pipe)

### Request Example
```scala
import fs2.*
import fs2.data.msgpack.low.*
import scodec.bits.*

val inputStream = Stream[Fallible, MsgpackItem](
  MsgpackItem.Array(2),
  MsgpackItem.UnsignedInt(hex"ab"),
  MsgpackItem.UnsignedInt(hex"abcdef01")
)

val binaryStream = inputStream.through(data.msgpack.low.toBinary)
```

### Response
#### Success Response
- `Byte`: A stream of bytes representing the MessagePack serialized data.

#### Response Example
(Binary stream output)
```

--------------------------------

### Unsafely Emit All Raw XML Matches with XPath

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/xml/xpath.md

The `filter.unsafeRaw` operator emits a stream of all matches, where each match is a nested stream of XML events. These inner streams must be fully consumed to prevent memory leaks. This example processes them in parallel.

```scala
stream
  .lift[IO]
  .through(filter.unsafeRaw(path))
  .parEvalMapUnbounded(_.through(render.raw()).compile.foldMonoid)
  .compile
  .toList
  .unsafeRunSync()
```

--------------------------------

### Handle Optional CSV Fields

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/csv/index.md

Provides methods like `asOptionalAt` and `asOptional` on `Row` and `CsvRow` to manage optional values. It allows customization for treating missing fields, empty strings, or specific placeholder values (like 'null') as `None`.

```scala
val testRow = CsvRow.fromNelHeaders(NonEmptyList.of("first" -> "value1", "second" -> "null"))

def get(field: String) =
  testRow.asOptional[String](field, missing = _ => Right(None), isEmpty = _ == "null")

get("first")
get("second")
get("third")
```

--------------------------------

### Dealing with optional data

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/csv/index.md

Provides methods like `asOptionalAt` and `asOptional` to handle missing, empty, or placeholder optional values in rows.

```APIDOC
## Dealing with optional data

Optional values have no unified representation in CSV. The `Row` and `Row[Header]` types provide `asOptionalAt` and `asOptional` methods to handle optional values.

By default, these methods fail if the field is missing and return `None` if the field is the empty string.

### Customizing optional value handling
Treat missing fields and specific placeholder values (like `"null"`) as `None`.

```scala
val testRow = CsvRow.fromNelHeaders(NonEmptyList.of("first" -> "value1", "second" -> "null"))

def get(field: String) =
  testRow.asOptional[String](field, missing = _ => Right(None), isEmpty = _ == "null")

get("first")
get("second")
get("third")
```
```

--------------------------------

### Render JSON Stream to String (Compact)

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/json/index.md

Collects a JSON token stream into a single compact string. No specific imports are needed beyond the stream and renderers.

```scala
stream.through(render.compact).compile.string
```

--------------------------------

### Write JSON to File (Compact Form)

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/json/index.md

Use this to write a JSON token stream to a file in a compact format without spaces or newlines. Ensure `fs2.io.file.Files` and `fs2.data.json.render` are imported.

```scala
import fs2.io.file.{Files, Flags, Path}

stream
  .through(render.compact)
  .through(text.utf8.encode)
  .lift[IO]
  .through(Files[IO].writeAll(Path("/some/path/to/file.json"), Flags.Write))
  .compile
  .drain
```

--------------------------------

### Build JSON Selector using DSL

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/json/transformations.md

Constructs a JSON selector using the fs2.data.json.selector DSL. The .compile method finalizes the selector.

```scala
import fs2.data.json.selector._

val selectorFromDsl = root.field("field3").iterate.compile
```

--------------------------------

### Use XPath Interpolator

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/xml/xpath.md

Shows how to use the xpath interpolator for creating XPath selectors. This method offers compile-time syntax checking for improved safety.

```scala
import fs2.data.xml.xpath.literals._

val path = xpath"//a"
```

--------------------------------

### Render Raw XML Event Stream to File

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/xml/index.md

Writes a raw XML event stream to a file without formatting. The stream is encoded to UTF-8 before writing.

```scala
import fs2.io.file.{Files, Flags, Path}

stream
  .through(render.raw())
  .through(text.utf8.encode)
  .through(Files[IO].writeAll(Path("/some/path/to/file.xml"), Flags.Write))
  .compile
  .drain
```

--------------------------------

### Serialize MsgpackItem Stream to Binary

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/msgpack/index.md

Use the `toBinary` pipe for stream validation during serialization. This pipe ensures that malformed data is not emitted.

```scala
import fs2.*
import fs2.data.msgpack.low.*
import scodec.bits.*

val inputStream = Stream[Fallible, MsgpackItem](
  MsgpackItem.Array(2),
  MsgpackItem.UnsignedInt(hex"ab"),
  MsgpackItem.UnsignedInt(hex"abcdef01")
)

val binaryStream = inputStream.through(data.msgpack.low.toBinary)
```

--------------------------------

### Parse Text Stream with Char Input

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/index.md

Demonstrates how to use the `tokens` pipe with a stream of `Char` to parse textual data. Ensure the appropriate `CharLikeChunks` instance is in scope.

```scala
// Stream of `Char`
Stream.emits("Some input string").through(tokens[Pure, Char])
```

--------------------------------

### Render JSON Stream to String (Pretty Print)

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/json/index.md

Collects a JSON token stream into a single string with indentation. The default indentation is 2 spaces and a width of 10. You can customize the indent size and width.

```scala
// default indentation is 2 spaces
stream.through(render.prettyPrint(width = 10)).compile.string
```

```scala
// if you are more into 4 spaces (or any other indentation size) you can change the indentation size
stream.through(render.prettyPrint(indent = 4, width = 10)).compile.string
```

--------------------------------

### Parse Text Stream with String Input

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/index.md

Demonstrates how to use the `tokens` pipe with a stream of `String` to parse textual data. Ensure the appropriate `CharLikeChunks` instance is in scope.

```scala
// Stream of `String`
Stream.emit("Some input string").through(tokens[Pure, String])
```

--------------------------------

### Build Lenient JSON Selector with Mandatory Field using DSL

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/json/transformations.md

Demonstrates building a JSON selector with a mandatory field and lenient iteration using the DSL. This allows for more flexible selection criteria.

```scala
val selectorFromDsl = root.field("field3").!.iterate.?.compile
```

--------------------------------

### Serialize a Stream of Custom Objects to MessagePack Bytes

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/msgpack/index.md

Serializes a stream of User objects into a stream of MessagePack bytes using the previously defined MsgpackSerializer. Ensure the Fallible effect type is handled appropriately.

```scala
val inputStream = Stream[Fallible, User](
  User("foo", 30, Instant.parse("2025-02-03T10:00:00.00Z"), List.empty),
  User("bar", 31, Instant.parse("2026-01-17T10:00:00.00Z"), List(1, 3, 7))
)

val byteStream = inputStream.through(data.msgpack.high.serialize[Fallible, User])
```

--------------------------------

### Generate JSON Lines from Sample JSON Array

Source: https://github.com/gnieh/fs2-data/blob/main/site/cookbooks/jsonlines.md

Reads a sample JSON array from a file, filters the first five elements using JSONPath, processes them using `writeJsonLines` to produce JSON Lines format, and then decodes the resulting bytes back to a string. This demonstrates generating JSON Lines data from existing JSON.

```scala
import fs2.data.json.jsonpath.literals._

Files[IO]
  .readAll(Path("site/cookbooks/data/json/sample.json"))
  .through(fs2.text.utf8.decode)
  .through(fs2.data.json.tokens)
  .through(fs2.data.json.jsonpath.filter.values(jsonpath"$$[0:4]"))
  .through(writeJsonLines)
  .through(fs2.text.utf8.decode)
  .compile
  .string
  .unsafeRunSync()
```

--------------------------------

### Parse JSON Input to AST with Circe

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/json/libraries.md

Parses the input JSON stream into an Abstract Syntax Tree (AST) using Circe integration. The resulting ASTs are then formatted with two spaces.

```scala
import fs2.data.json.circe._

val asts = input[Fallible].through(ast.parse)
asts.map(_.spaces2).compile.toList
```

--------------------------------

### Programmatic Usage of csv-spectrum

Source: https://github.com/gnieh/fs2-data/blob/main/csv/shared/src/test/resources/csv-spectrum/readme.md

Use this snippet to load all CSV test cases and their JSON counterparts. The callback receives an error object if any, and an array of test case objects.

```javascript
vsr spectrum = require('csv-spectrum')
spectrum(function(err, data) {
  // data is an array of objects has all the csv and json versions of the tests
})
```

--------------------------------

### Collect AST Values from Byte Stream

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/msgpack/index.md

Collects all MsgpackValue AST nodes from a byte stream into a list. This demonstrates consuming the stream of AST values.

```scala
valueStream.compile.toList
```

--------------------------------

### Serialize high-level CBOR value stream to binary

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/cbor/index.md

Serialize a high-level CBOR value stream into a binary format using the `toBinary` pipe from the `fs2.data.cbor.high` package.

```scala
valueStream
  .through(data.cbor.high.toBinary)
  .compile
  .to(ByteVector)
  .map(_.toHex)
```

--------------------------------

### Parse CBOR binary stream to low-level items

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/cbor/index.md

Use the `items` pipe to parse a CBOR binary stream into a low-level representation. Ensure necessary imports are included.

```scala
import fs2._
import fs2.data.cbor.low._

import scodec.bits._

val byteStream = Stream.chunk(Chunk.byteVector(hex"8301820203820405"))

val itemStream = byteStream.through(items[Fallible])
itemStream.compile.toList
```

--------------------------------

### Compile and Use a Query with FS2 Stream

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/json/jq.md

Compile a query once and reuse it with an FS2 stream for efficient data processing. Ensure the stream is processed through the compiled query.

```scala
val qCompiler = jq.Compiler[SyncIO]

val compiled = qCompiler.compile(query).unsafeRunSync()

stream
  .through(compiled)
  .compile
  .to(collector.pretty())
  .unsafeRunSync()
```

--------------------------------

### StaticHeaders Instance Definition

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/csv/index.md

Define a `StaticHeaders` instance to provide headers for encoding and decoding when they are not present in the data itself. This is useful for pipes like `decodeUsingStaticHeaders` and `encodeUsingStaticHeaders`.

```scala
implicit val staticHeadersForMyRow: StaticHeaders[MyRow, String] = StaticHeaders.instance(NonEmptyList.of("i", "s", "j"))
```

--------------------------------

### Build XML DOM Document

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/xml/index.md

Builds an XML DOM of a specified type from an XML event stream. Requires an implicit DocumentBuilder instance for the target DOM type.

```scala
import dom._

trait SomeDocType

implicit val builder: DocumentBuilder[SomeDocType] = ???
stream.through(documents[IO, SomeDocType])
```

--------------------------------

### Collect Raw XML String

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/xml/index.md

Collects an XML event stream into a raw XML string.

```scala
stream.compile.to(collector.raw()).unsafeRunSync()
```

--------------------------------

### Compile jq query into a pipe

Source: https://github.com/gnieh/fs2-data/blob/main/site/cookbooks/jq.md

Compiles a defined jq query into an fs2 Pipe using the Compiler class. This prepares the query for use in an fs2 stream transformation.

```scala
import fs2.data.json.jq.Compiler

val queryCompiler = Compiler[IO]

val queryPipe = queryCompiler.compile(query).unsafeRunSync()
```

--------------------------------

### Deserialize MessagePack Bytes to a Stream of Custom Objects

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/msgpack/index.md

Deserializes a stream of MessagePack bytes back into a stream of User objects using the defined MsgpackDeserializer. The result is collected into a list.

```scala
byteStream
  .through(data.msgpack.high.deserialize[Fallible, User])
  .compile.toList
```

--------------------------------

### Convert MessagePack Bytes to AST Values

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/msgpack/index.md

Converts a stream of MessagePack bytes into a stream of MsgpackValue AST nodes. This allows for more flexible, albeit potentially less performant, handling of MessagePack data.

```scala
val valueStream = byteStream.through(data.msgpack.high.ast.valuesFromBytes)
```

--------------------------------

### Build AST Values from Token Stream

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/json/index.md

Use the `values` pipe to build JSON ASTs from an existing token stream. Requires an implicit `Builder[Json]` in scope. This method is less efficient than `ast.parse` if only values are required.

```scala
import ast._

trait SomeJsonType

implicit val builder: Builder[SomeJsonType] = ???

stream.through(values[Fallible, SomeJsonType])
```

--------------------------------

### Writing CSV

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/csv/index.md

Provides pipes for encoding rows to CSV format, with or without headers.

```APIDOC
## Writing CSV

There are pipes for encoding rows to CSV, with or without headers.

### Writing without headers

```scala
val testRows = Stream(Row(NonEmptyList.of("3", "", "test")))
testRows
  .through(lowlevel.writeWithoutHeaders)
  .through(lowlevel.toRowStrings(/* separator: Char = ',', newline: String = "\n"*/))
  .compile
  .string
```

### Writing with headers
Use `writeWithGivenHeaders` or `encodeRowWithFirstHeaders` for `CsvRow`. For non-String headers, provide an instance of `WritableHeader`.

```scala
// Example for writing with headers (assuming 'stream' is defined and has headers)
// val withh = stream.through(lowlevel.headers[Fallible, String])
// withh.through(lowlevel.writeWithGivenHeaders).compile.string
```
```

--------------------------------

### Parse XML to Stream of Events

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/xml/index.md

Creates a stream of XML events from an input string. This pipe validates XML structure during parsing.

```scala
import cats.effect._
import cats.effect.unsafe.implicits.global

import fs2._
import fs2.data.xml._

val input = """<a xmlns:ns="http://test.ns">
              |  <ns:b ns:a="attribute">text</ns:b>
              |</a>
              |<a>
              |  <b/>
              |  test entity resolution &amp; normalization
              |</a>""".stripMargin

val stream = Stream.emit(input).through(events[IO, String]())
```

--------------------------------

### Collect Raw XML Matches Non-Deterministically

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/xml/xpath.md

Similar to collecting raw XML matches, but setting `deterministic = false` allows results to be emitted as early as possible, rather than in order.

```scala
stream
  .lift[IO]
  .through(filter.collect(path, collector.raw(), deterministic = false))
  .compile
  .toList
  .unsafeRunSync()
```

--------------------------------

### Decode Textual Input with ISO-8859-1 Encoding

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/index.md

Shows how to decode a stream of bytes from a file into characters using the ISO-8859-1 (latin1) encoding. This is necessary for textual data formats when the file is not UTF-8 encoded. The `tokens` pipe is then applied to the decoded character stream.

```scala
// for instance if your input is encoded in ISO-8859-1 aka latin1
import fs2.data.text.latin1._
// if you have UTF-8 instead:
// import fs2.data.text.utf8._

Files[IO]
  .readAll(Path("/some/path/to/a/file.data"), 1024, Flags.Read)
  // decoding is done by the now in scope `CharLikeChunks[IO, Byte]` instance
  .through(tokens)
  .compile
  .drain
```

--------------------------------

### Parse XPath String

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/xml/xpath.md

Demonstrates parsing an XPath string into a selector using XPathParser.either. This method wraps the result in a MonadError to handle potential parsing errors.

```scala
import fs2.data.xml.xpath._

val selector = XPathParser.either("//a")
```

--------------------------------

### Parse JSON Stream to AST Values

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/json/index.md

Use the `ast.parse` pipe to directly parse an input stream into a stream of AST values. Requires an implicit `Builder[Json]` in scope. This is more efficient than tokenizing first if only values are needed.

```scala
import ast._

trait SomeJsonType

implicit val builder: Builder[SomeJsonType] = ???

Stream.emit(input).covary[Fallible].through(parse)
```

--------------------------------

### Resolve XML Namespaces

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/xml/index.md

Applies namespace resolution to an existing stream of XML events. Requires an IO context.

```scala
val nsResolved = stream.through(namespaceResolver[IO])
```

--------------------------------

### Serialize MsgpackItem Stream to Non-Validated Binary

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/msgpack/index.md

Use the `toNonValidatedBinary` pipe when you are certain the item stream is valid. This bypasses stream validation for potentially faster serialization.

```scala
inputStream.through(data.msgpack.low.toNonValidatedBinary)
```

--------------------------------

### Generate JSON Array Wrapped in Object

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/json/index.md

Wraps a stream of tokenized events into a JSON array under the key "events". Requires `fs2.data.json.circe._` for Circe integration and `fs2.data.json.ast.tokenize`.

```scala
import fs2.data.json.circe._

val wrappedTokens = events.through(ast.tokenize).through(wrap.asArrayInObject(at = "events"))
```

--------------------------------

### Define MsgpackDeserializer for a Custom Case Class

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/msgpack/index.md

Defines an implicit MsgpackDeserializer for a custom User case class. This involves sequencing deserializers for each field and then constructing the User object.

```scala
implicit val userDeserializer: MsgpackDeserializer[User] = 
  for {
    name    <- MsgpackDeserializer[String]
    age     <- MsgpackDeserializer[Int]
    created <- MsgpackDeserializer[Instant]
    visits <- MsgpackDeserializer[List[Long]]
  } yield User(name, age, created, visits)
```

--------------------------------

### Derive CellDecoder and CellEncoder for Unary Products and Aliases

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/csv/generic.md

Shows how to derive CellDecoder and CellEncoder for unary product types (case classes with one field) and how to handle custom CSV value annotations for coproduct members. Requires an implicit CellDecoder[String] for the field.

```scala
import fs2.data.csv.generic.semiauto._

sealed trait Advanced
object Advanced {
  @CsvValue("Active") case object On extends Advanced
  case class Unknown(name: String) extends Advanced
}

// works as we have an implicit CellDecoder[String]
implicit val unknownDecoder = deriveCellDecoder[Advanced.Unknown]
implicit val advancedDecoder = deriveCellDecoder[Advanced]

advancedDecoder("Active")
advancedDecoder("Off")

implicit val unknownEncoder = deriveCellEncoder[Advanced.Unknown]
implicit val advancedEncoder = deriveCellEncoder[Advanced]

advancedEncoder(Advanced.On)
advancedEncoder(Advanced.Unknown("Off"))
```

--------------------------------

### Collect Pretty-Printed XML String

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/xml/index.md

Collects an XML event stream into a pretty-printed XML string.

```scala
stream.compile.to(collector.pretty()).unsafeRunSync()
```

--------------------------------

### Define MsgpackSerializer for a Custom Case Class

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/msgpack/index.md

Defines an implicit MsgpackSerializer for a custom User case class. This involves mapping each field of the case class to its corresponding MsgpackSerializer and concatenating the results.

```scala
import fs2.*
import fs2.data.msgpack.high.*
import java.time.Instant

case class User(name: String, age: Int, created: Instant, visits: List[Long])

implicit val userSerializer: MsgpackSerializer[User] = {
  case User(name, age, created, visits) =>
    for {
      name    <- MsgpackSerializer[String](name)
      age     <- MsgpackSerializer[Int](age)
      created <- MsgpackSerializer[Instant](created)
      visits  <- MsgpackSerializer[List[Long]](visits)
    } yield name ++ age ++ created ++ visits
  }
```

--------------------------------

### Convert AST Values back to MessagePack Bytes

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/msgpack/index.md

Converts a stream of MsgpackValue AST nodes back into a stream of MessagePack bytes. This is useful for re-serializing data that has been processed or manipulated as AST nodes.

```scala
val byteStream2 = valueStream.through(data.msgpack.high.ast.valuesToBytes)
```

--------------------------------

### Convert Values to JSON Tokens

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/json/index.md

Use the `tokenize` pipe to convert a stream of values into a stream of JSON tokens. Requires an implicit `Tokenizer[SomeJsonType]` in scope.

```scala
import ast._

trait SomeJsonType

val v: SomeJsonType = ???

implicit val tokenizer: Tokenizer[SomeJsonType] = ???

Stream.emit(v).through(tokenize[Fallible, SomeJsonType])
```

--------------------------------

### Migrate from circe-fs2 stringStreamParser to fs2-data-json

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/json/libraries.md

Replaces `circe-fs2`'s `stringStreamParser` with `fs2.data.json.ast.parse` for parsing JSON streams. The output is formatted with two spaces and collected into a list.

```scala
import io.circe.fs2._

import cats.effect._

input[SyncIO]
  .through(stringStreamParser)
  .map(_.spaces2)
  .compile
  .toList
  .unsafeRunSync()
```

--------------------------------

### Read JSON file and write to stdout

Source: https://github.com/gnieh/fs2-data/blob/main/site/cookbooks/jq.md

Reads a JSON file using fs2-io and writes its content to standard output. This snippet uses pure fs2 operators and does not involve fs2-data.

```scala
import cats.effect.IO
import cats.effect.unsafe.implicits.global

import fs2.io.file.{Files, Path}
import fs2.io.stdout
import fs2.text.utf8

Files[IO]
  .readUtf8(Path("site/cookbooks/data/json/sample.json"))
  .through(utf8.encode[IO])
  .through(stdout)
  .compile
  .drain
  .unsafeRunSync()
```

--------------------------------

### Create RowEncoder for Case Class

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/csv/index.md

Use `RowEncoder.forColumns` to create a `RowEncoder` for a simple positional case class, simplifying the encoding process. This method takes a function that extracts the fields from the case class.

```scala
case class Triple(i: Int, s: String, j: Int)

val tripleEncoder: RowEncoder[Triple] =
  RowEncoder.forColumns((t: Triple) => (t.i, t.s, t.j))
```

--------------------------------

### Parse JSONPath Selector String

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/json/jsonpath.md

Creates a JSONPath selector by parsing a string. The result is wrapped in a MonadError to handle potential parsing errors. Import fs2.data.json.jsonpath._.

```scala
import fs2.data.json.jsonpath._

val selector = JsonPathParser.either("$.field3[*]")
```

--------------------------------

### CsvRowDecoder and CsvRowEncoder with forColumns Helper

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/csv/index.md

Use the `forColumns` helper to reduce boilerplate when mapping named CSV columns to case class constructor arguments or tuple elements. This method simplifies the creation of decoders and encoders by accepting column names and a mapping function.

```scala
case class NamedRow(i: Int, j: Int, s: String)

val namedDecoder: CsvRowDecoder[NamedRow, String] = 
  // Depending on your Scala version, you might also be able to use .forColumns(NamedRow.apply) instead
  CsvRowDecoder.forColumns("i", "j", "s")(NamedRow(_, _, _))

val namedEncoder: StaticCsvRowEncoder[NamedRow, String] = 
  CsvRowEncoder.forColumns("i", "j", "s")((r: NamedRow) => (r.i, r.j, r.s))
```

--------------------------------

### Parse and pretty-print JSON

Source: https://github.com/gnieh/fs2-data/blob/main/site/cookbooks/jq.md

Parses JSON input from a file and pretty-prints the resulting JSON stream to standard output using fs2-data-json pipes.

```scala
import fs2.data.json

Files[IO]
  .readUtf8(Path("site/cookbooks/data/json/sample.json"))
  .through(json.tokens) // parsing JSON input
  .through(json.render.prettyPrint()) // pretty printing JSON stream
  .through(utf8.encode[IO])
  .through(stdout)
  .compile
  .drain
  .unsafeRunSync()
```

--------------------------------

### Render Wrapped JSON Stream to List

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/json/index.md

Renders the wrapped JSON token stream (generated by `wrap.asArrayInObject`) into a list of strings. This is useful for inspecting the generated JSON.

```scala
wrappedTokens.through(render.compact).compile.toList
```

--------------------------------

### Transform and Serialize JSON Data with Circe

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/json/libraries.md

Transforms JSON data using a specified selector and a wrapping function, then serializes the result into a pretty-printed string. This demonstrates deserializing into `Data`, transforming it, and then serializing `WrappedData`.

```scala
import fs2.data.json.selector._
import fs2.data.json.circe._

val values = stream.through(codec.deserialize[Fallible, Data])
values.compile.toList

def wrap(data: Data): WrappedData =
  WrappedData(data.field1, data.field2, data.field3.map(Wrapped(_)))

val sel = root.field("field3").iterate.compile

val transformed = stream.through(codec.transform(sel, wrap))
transformed.compile.to(collector.pretty())
```

--------------------------------

### Parse and Serialize CSV with Case Classes

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/csv/index.md

Use this snippet for the common task of parsing CSV into a case class or serializing data into CSV. It assumes the CSV contains headers and uses them for encoding. Type class instances for `CsvRowDecoder` and `CsvRowEncoder` determine how data is decoded/encoded.

```scala
import cats.effect._

import fs2._
import fs2.data.csv._
import fs2.data.csv.generic.semiauto._

val input = """i,s,j
1,test,2
,other,-3
""".stripMargin

// Usually this would come from a file, for example using Files[IO].readAll
val textStream = Stream.emit(input).covary[Fallible]
implicit val myRowDecoder: CsvRowDecoder[MyRow, String] = deriveCsvRowDecoder
implicit val myRowEncoder: CsvRowEncoder[MyRow, String] = deriveCsvRowEncoder

// decodeUsingHeaders can take a `Char` indicating the separator to use
// for example `decodeUsingHeaders[MyRow](';') for a semi-colon separated csv
val decodedStream = textStream.through(decodeUsingHeaders[MyRow]())
val caseClasses = decodedStream.compile.toList

val backAsText = decodedStream.through(encodeUsingFirstHeaders(fullRows = true)).compile.string
```

--------------------------------

### Convert XML DOM to Events

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/xml/index.md

Transforms a stream of XML DOM documents back into a stream of XML events. Requires implicit DocumentBuilder and DocumentEventifier instances.

```scala
import dom._

trait SomeDocType

implicit val builder: DocumentBuilder[SomeDocType] = ???
implicit val eventifier: DocumentEventifier[SomeDocType] = ???

stream.through(documents[IO, SomeDocType])
      .through(eventify[IO, SomeDocType])
```

--------------------------------

### Serialize Values to JSON Tokens

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/json/index.md

Use the `serialize` pipe to serialize a stream of values into a stream of JSON tokens. Requires an implicit `Serializer[T]` for the value type `T`.

```scala
import codec._

implicit val serializer: Serializer[String] = ???

Stream("a", "b", "c").through(serialize)
```

--------------------------------

### Decode CBOR Items with Tags to JSON Tokens

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/cbor-json/index.md

Demonstrates decoding CBOR items with specific tags (e.g., base64 encoding, positive big numbers) into JSON tokens. Supports tags for byte string interpretation.

```scala
import fs2.data.cbor.Tags

Stream
  .emits(List(
    CborItem.Tag(Tags.ExpectedBase64Encoding),
    CborItem.StartArray(2),
    CborItem.ByteString(hex"cafebabe"),
    CborItem.ByteString(hex"ff"),
    CborItem.Tag(Tags.PositiveBigNum),
    CborItem.ByteString(hex"1234567890abcdef")
  ))
  .through(cbor.json.decodeItems[Fallible])
  .compile
  .toList
```

--------------------------------

### Case Class Derivation with Default Values

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/csv/generic.md

Supports default values in case classes for decoding. Missing columns or empty cell values are treated as defaults. Refrain from defining defaults or define manually if explicit empty value handling is needed.

```scala
import fs2.data.csv.generic.auto._

case class MyRowDefault(i: Int = 42, j: Int, s: String)

val decoded = stream.through(decodeUsingHeaders[MyRowDefault]())
```

--------------------------------

### Transform JSON data using a compiled jq pipe

Source: https://github.com/gnieh/fs2-data/blob/main/site/cookbooks/jq.md

Applies a compiled jq query pipe to transform JSON data read from a file. The transformed data is then pretty-printed and written to standard output.

```scala
Files[IO]
  .readUtf8(Path("site/cookbooks/data/json/sample.json"))
  .through(json.tokens)
  .through(queryPipe) // the transformation using the query pipe
  .through(json.render.prettyPrint())
  .through(utf8.encode[IO])
  .through(stdout)
  .compile
  .drain
  .unsafeRunSync()
```

--------------------------------

### Define Data Models for Circe Deserialization

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/json/libraries.md

Defines case classes `Data`, `WrappedData`, and `Wrapped` with `@JsonCodec` annotations for use with Circe's generic derivation. These models represent the structure of the JSON data.

```scala
import io.circe.generic.JsonCodec

@JsonCodec
case class Data(field1: Int, field2: Option[String], field3: List[Int])

@JsonCodec
case class WrappedData(field1: Int, field2: Option[String], field3: List[Wrapped])

@JsonCodec
case class Wrapped(test: Int)
```

--------------------------------

### Manual CsvRowDecoder Implementation

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/csv/index.md

Implement a CsvRowDecoder manually to decode CSV rows into a case class. This is useful when the CSV headers do not directly map to constructor arguments or when custom logic is needed for decoding fields. Ensure all required fields are handled.

```scala
case class MyRow(i: Option[Int], j: Int, s: String)

implicit object MyRowDecoder extends CsvRowDecoder[MyRow, String] {
  def apply(row: CsvRow[String]): DecoderResult[MyRow] = 
  for {
    i <- row.as[Int]("i").map(Some(_)).leftFlatMap(_ => Right(None))
    j <- row.as[Int]("j")
    s <- row.as[String]("s")
  } yield MyRow(i, j, s)
}
val decoded = withh.through(lowlevel.decodeRow[Fallible, String, MyRow])
decoded.compile.toList
```

--------------------------------

### Implement Custom CellEncoder for Wrapper

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/csv/index.md

Define a `CellEncoder` for a custom `Wrapper` case class. This can be done directly by providing a function from `Wrapper` to `String`, or by using `contramap` on an existing `CellEncoder` for `String`.

```scala
case class Wrapper(content: String)
implicit val wrapperCellEncoder: CellEncoder[Wrapper] = (w: Wrapper) => w.content
```

```scala
implicit val wrapperCellEncoder2: CellEncoder[Wrapper] = CellEncoder[String].contramap(_.content)
```

--------------------------------

### Derive CellDecoder and CellEncoder for Coproducts

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/csv/generic.md

Demonstrates how to automatically derive CellDecoder and CellEncoder instances for sealed traits (coproducts) using generic derivation. This allows decoding and encoding of custom enumerated types.

```scala
import fs2.data.csv.generic._
import fs2.data.csv.generic.semiauto._

sealed trait State
object State {
  case object On extends State
  case object Off extends State
}

implicit val stateDecoder = deriveCellDecoder[State]
// use stateDecoder to derive decoders for rows...or just test:
stateDecoder("On")
stateDecoder("Off")

// same goes for the encoder
implicit val stateEncoder = deriveCellEncoder[State]
stateEncoder(State.On)
```

--------------------------------

### Parse JSON Stream using fs2-data-json AST

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/json/libraries.md

Parses an input JSON stream into an AST using `fs2.data.json.ast.parse` and formats the output. This is the recommended replacement for `circe-fs2`'s `stringStreamParser`.

```scala
import fs2.data.json._
import fs2.data.json.circe._

input[Fallible]
  .through(ast.parse)
  .map(_.spaces2)
  .compile
  .toList
```

--------------------------------

### Validate MsgpackItem Stream

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/msgpack/index.md

Apply the `validate` pipe directly to a stream of `MsgpackItem`s. This method will raise an error within the effect if the stream is malformed.

```scala
Stream(MsgpackItem.Array(999))
  .through(data.msgpack.low.validate[Fallible])
  .compile.toList
```

--------------------------------

### Validate MsgpackItem Stream

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/msgpack/index.md

The `validate` pipe can be applied directly to a stream of `MsgpackItem`s to validate its structure. It will raise an error within the effect if the stream is malformed.

```APIDOC
## validate

### Description
Validates a stream of `MsgpackItem`s. Raises an error within the effect if the stream is malformed.

### Method
Pipe

### Endpoint
N/A (Pipe)

### Parameters
N/A (Pipe)

### Request Example
```scala
Stream(MsgpackItem.Array(999))
  .through(data.msgpack.low.validate[Fallible])
  .compile.toList
```

### Response
#### Success Response
- `MsgpackItem`: The validated stream of MessagePack items.

#### Response Example
(Stream of MsgpackItems)
```

--------------------------------

### Parse JSON to Token Stream

Source: https://github.com/gnieh/fs2-data/blob/main/site/documentation/json/index.md

Use the `tokens` pipe to create a stream of JSON tokens from an input character stream. This pipe validates the JSON structure and emits tokens as they become available.

```scala
import cats.effect._
import cats.syntax.all._

import fs2._
import fs2.data.json._

val input = """
{
  "field1": 0,
  "field2": "test",
  "field3": [1, 2, 3]
}
{
  "field1": 2,
  "field3": []
}
""".stripMargin

val stream = Stream.emit(input).through(tokens[Fallible, String])
stream.compile.toList
```