fs2-data

The fs2-data-json module provides a streaming, declarative implementation of a jq-like query language. This allows you to extract and transform specific parts of a JSON input stream.

Note: The JSON query feature is currently experimental.

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

// Example setup: converting a string input into a stream of JSON tokens
val input = "{\"field1\": 0, \"field2\": \"test\", \"field3\": [1, 2, 3]}"
val stream = Stream.emit(input).through(tokens[SyncIO, String])

The fs2.data.xml.dom package allows you to convert between XML event streams and a Document Object Model (DOM).

Building a DOM from events

Use the documents pipe. This requires an implicit DocumentBuilder[Doc] in scope, which defines how your specific Doc type is constructed from events.

Converting a DOM back to events

Use the eventify pipe to transform a stream of Doc objects back into a stream of XML events. This requires an implicit DocumentEventifier[Doc] in scope.

import dom._

trait SomeDocType

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

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

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

When you derive a CsvRowDecoder or CsvRowEncoder using the generic modules (automatic or semi-automatic), the resulting instances also implement the StaticHeaders type class.

This implementation allows these decoders/encoders to be used with CSV pipes that require header information to be available at compile-time/statically, such as:

• encodeUsingFirstHeaders
• encodeUsingHeaders

The fs2-data-msgpack module is divided into two primary namespaces depending on whether you want to work with Scala types or raw MessagePack structures:

• fs2.data.msgpack.high: The recommended API for most users. It works directly with Scala values. You define MsgpackSerializer[A] and MsgpackDeserializer[A] to map your classes to MessagePack.
• fs2.data.msgpack.low: A low-level API that operates on a flat model of MsgpackItems. This model mimics the raw MessagePack format and serves as the intermediate layer between binary data and the high-level Scala API.

The fs2-data approach to processing JSON follows a linear pipeline:

1. Reading: Use fs2 operators (like fs2-io) to read raw bytes/text.
2. Parsing: Use the json.tokens pipe from fs2-data-json to convert the input stream into JSON Tokens.
3. Transforming: Apply logic to the stream of tokens. This can be done using standard fs2 operators or by using the fs2.data.json.jq.Compiler to turn a jq query into a Pipe.
4. Printing: Use json.render.prettyPrint() to convert the transformed tokens back into a formatted JSON string.
5. Writing: Use fs2 operators (like fs2-io) to write the resulting stream to a destination (e.g., stdout or a file).

The fs2-data-csv module provides two layers of abstraction for processing CSV data:

1. High-level API (fs2.data.csv._): Provides convenient fs2.Pipes designed for common use cases, such as decoding CSV directly into case classes or encoding data with headers. It is built by composing low-level pipes.
2. Low-level API (fs2.data.csv.lowlevel._): Provides granular pipes that perform single steps in a processing pipeline (e.g., parsing raw characters into rows). These are ideal for building complex, custom CSV processing logic.

Both APIs share core data types and type classes located in fs2.data.csv._.

To add support for a new Scala XML DOM library to fs2-data, you must implement the following two traits:

1. DocumentBuilder: Responsible for building a DOM from a stream of events.
2. DocumentEventifier: Responsible for converting a DOM back into a stream of events.

fs2-data provides several implementations out of the box for common libraries.

A query consists of two potential phases:

1. Filtering Phase: Selects elements from the input stream (similar to JSONPath). If no selector is provided, the input is selected as a whole.
2. Construction Phase: Builds the output based on the selected elements. If no constructor is provided, elements are emitted unchanged.

Query Patterns

• filter: Applies only selection (e.g., .field).
• filter | constructor: Selects a subset and builds a new structure (e.g., .field | { "new": . }).
• constructor: Builds a result from the input top-level value (e.g., [ . ]).

Constructors can contain sub-queries applied to each element selected during the filtering phase.

CSV does not have a standard for missing or null values. fs2-data-csv provides asOptionalAt (for Row) and asOptional (for Row[Header]) to handle these cases.

By default, these methods return None if a field is an empty string, but will fail if the field is missing (index out of bounds or header missing). You can customize this behavior by providing missing and isEmpty functions.

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

// Custom logic: treat missing fields as None and the string "null" as None
def get(field: String) =
  testRow.asOptional[String](field, missing = _ => Right(None), isEmpty = _ == "null")

get("first")  // Some(value1)
get("second") // None
get("third")  // None