### Installing sflow with npm or Bun

Source: https://github.com/snomiao/sflow/blob/main/README.md

This snippet provides instructions for installing the sflow library using popular Node.js package managers, npm and Bun. It outlines the commands required to add sflow as a dependency to your project.

```sh
npm install sflow

# or if you are using bun

bun add sflow
```

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

### Merging Distinct Streams in sflow (TypeScript)

Source: https://github.com/snomiao/sflow/blob/main/docs/Examples.md

This example demonstrates how to process multiple streams to get distinct elements and then merge them into a single output stream using `sflow` and `uniq`. It shows how to combine results from independent stream operations into a unified sequence. This is useful for consolidating data from various sources.

```TypeScript
import { sflow } from "sflow";
import { uniq } from "sflow/uniq";

async function example13() {
  const stream1 = sflow([1, 2, 3, 2, 1]);
  const stream2 = sflow([4, 5, 5, 6]);

  const distinctStream1 = stream1.uniq();
  const distinctStream2 = stream2.uniq();

  const result = await sflow([distinctStream1, distinctStream2])
    .merge()
    .log() // Prints: 1, 4, 2, 5, 3, 6
    .toArray();

  console.log(result); // Outputs: [1, 4, 2, 5, 3, 6]
}

example13();
```

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

### Using a Custom Reducer with sflow (TypeScript)

Source: https://github.com/snomiao/sflow/blob/main/docs/Examples.md

This example demonstrates defining and applying a custom reducer with `sflow`'s `reduceEmits` function. The reducer calculates a running sum (`next`) and emits a transformed value (`emit`) based on the sum and current item. This allows for flexible stateful transformations and intermediate output generation.

```TypeScript
import { sflow } from "sflow";
import { reduceEmits } from "sflow/reduceEmits";

async function example15() {
  const data = [1, 2, 3, 4, 5];

  const customReducer = (sum, value) => ({
    next: sum + value,
    emit: sum + value * 2
  });

  const result = await sflow(data)
    .through(reduceEmits(customReducer, 0))
    .log() // Prints: 2, 6, 12, 20, 30
    .toArray();

  console.log(result); // Outputs: [2, 6, 12, 20, 30]
}

example15();
```

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

### Chunking Stream Elements with sflow in TypeScript

Source: https://github.com/snomiao/sflow/blob/main/docs/Examples.md

This example illustrates how to chunk a stream of numbers into arrays of a specified size using the `chunk` operator from the `sflow` library. It processes an array of numbers, grouping them into sub-arrays of three elements each, and then converts the result to an array, demonstrating how to buffer stream elements.

```TypeScript
import { sflow } from "sflow";

async function example2() {
  const result = await sflow([1, 2, 3, 4, 5, 6, 7, 8])
    .chunk(3)
    .toArray();

  console.log(result); // Outputs: [[1, 2, 3], [4, 5, 6], [7, 8]]
}

example2();
```

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

### Handling Cache with Keyv in sflow (TypeScript)

Source: https://github.com/snomiao/sflow/blob/main/docs/Examples.md

This snippet demonstrates how to integrate Keyv for caching stream data using `sflow`'s `cacheTails` utility. It initializes a Keyv store with a TTL, fetches simulated data, and pipes it through the cache, logging the cached items. This is useful for preventing redundant data fetching.

```TypeScript
import { sflow } from "sflow";
import Keyv from "keyv";
import { cacheTails } from "sflow/cacheTails";

async function example10() {
  const store = new Keyv({ ttl: 1000 * 60 * 60 }); // Cache for 1 hour

  const fetchAndProcessData = async () => {
    // Simulate data fetching
    const data = [1, 2, 3, 4, 5];
    return sflow(data);
  };

  const dataStream = await fetchAndProcessData();

  const result = await dataStream
    .through(cacheTails(store, "my-cache-key"))
    .log() // Print cached items
    .toArray();

  console.log(result); // Output: [1, 2, 3, 4, 5]
}

example10();
```

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

### Aggregating Stream Items by Key in sflow (TypeScript)

Source: https://github.com/snomiao/sflow/blob/main/docs/Examples.md

This example illustrates how to aggregate stream items based on a specific key using `sflow`'s `reduceEmits` function. It defines a custom reducer to sum values for each category, emitting intermediate aggregation states. This is useful for summarizing data within a stream.

```TypeScript
import { sflow } from "sflow";
import { reduceEmits } from "sflow/reduceEmits";

async function example11() {
  const data = [
    { category: "A", value: 10 },
    { category: "B", value: 20 },
    { category: "A", value: 5 }
  ];

  const reducer = (acc, { category, value }) => {
    const next = { ...acc, [category]: (acc[category] || 0) + value };
    return { next, emit: next };
  };

  const result = await sflow(data)
    .through(reduceEmits(reducer, {}))
    .log() // Prints: {"A": 10}, {"A": 10, "B": 20}, {"A": 15, "B": 20}
    .toArray();

  console.log(result);
  // Outputs: 
  // [
  //   { "A": 10 },
  //   { "A": 10, "B": 20 },
  //   { "A": 15, "B": 20 }
  // ]
}

example11();
```

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

### Buffering sflow Stream Items by Interval (TypeScript)

Source: https://github.com/snomiao/sflow/blob/main/docs/Examples.md

This example illustrates how to buffer stream items into arrays based on a specified time interval using the `chunkInterval` operator. It's useful for batch processing or grouping events that occur within a certain timeframe.

```typescript
import { sflow } from "sflow";
import { chunkIntervals } from "sflow/chunkIntervals";
import { sleep } from "bun";

async function example19() {
  const result = await sflow([1, 2, 3, 4])
    .forEach(() => sleep(50))
    .chunkInterval(100)
    .log() // Prints: [1, 2], [3, 4]
    .toArray();

  console.log(result); // Outputs: [[1, 2], [3, 4]]
}

example19();
```

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

### Parsing and Formatting TSV Data with sflow (TypeScript)

Source: https://github.com/snomiao/sflow/blob/main/docs/Examples.md

This example demonstrates parsing Tab Separated Values (TSV) data using `sflow`'s `tsvParses` and then formatting it back into TSV using `tsvFormats`. It also includes a transformation step to modify the 'age' field. This is useful for processing structured text data like CSV/TSV files.

```TypeScript
import { sflow } from "sflow";
import { tsvParses, tsvFormats } from "sflow/xsvStreams";

async function example17() {
  const tsvData = "name\tage\nJohn\t30\nJane\t25";
  
  const result = await sflow(tsvData)
    .through(tsvParses("name\tage"))
    .map((record) => ({ ...record, age: Number(record.age) + 1 }))
    .log() // Prints: { name: "John", age: 31 }, { name: "Jane", age: 26 }
    .through(tsvFormats("name\tage"))
    .text();

  console.log(result); // Outputs formatted TSV with updated ages
}

example17();
```

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

### Reducing Stream with Intermediate Emissions using sflow in TypeScript

Source: https://github.com/snomiao/sflow/blob/main/docs/Examples.md

This example illustrates how to use `reduceEmits` with `sflow` to perform a reduction operation while also emitting intermediate states of the reduction. It calculates a running sum of numbers in the stream, and each intermediate sum is emitted as a new stream element, providing visibility into the reduction process and allowing for analysis of partial results.

```TypeScript
import { sflow } from "sflow";
import { reduceEmits } from "sflow/reduceEmits";

async function example8() {
  const reducer = (sum, value) => ({
    next: sum + value,
    emit: sum + value
  });

  const result = await sflow([1, 2, 3, 4])
    .through(reduceEmits(reducer, 0))
    .log() // Prints: 1, 3, 6, 10
    .toArray();

  console.log(result); // Outputs: [1, 3, 6, 10]
}

example8();
```

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

### Initializing sflow Streams from Various Sources in TypeScript

Source: https://github.com/snomiao/sflow/blob/main/README.md

This snippet illustrates how to create sflow instances from different data types, including standard arrays, Promises resolving to arrays, and async iterables. It demonstrates sflow's flexibility in consuming diverse data sources as the starting point for a stream processing pipeline.

```typescript
import { sflow } from "sflow";

// From an array
const flow1 = sflow([1, 2, 3, 4]);

// From a promise
const flow2 = sflow(Promise.resolve([1, 2, 3, 4]));

// From an async iterable
async function* asyncGenerator() {
  yield 1;
  yield 2;
  yield 3;
}
const flow3 = sflow(asyncGenerator());
```

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

### Basic Stream Processing with sflow in TypeScript

Source: https://github.com/snomiao/sflow/blob/main/README.md

This example demonstrates a fundamental sflow pipeline, showcasing how to initialize a stream from an array, apply `map`, `filter`, and `reduce` transformations, and log intermediate results. It highlights the functional chaining of operations to process data from an initial array into a final aggregated result.

```typescript
import { sflow } from "sflow";

async function run() {
  let result = await sflow([1, 2, 3, 4])
    .map((n) => n * 2)
    .log() // this stage prints 2, 4, 6, 8
    .filter((n) => n > 4)
    .log() // this stage prints 6, 8
    .reduce((a, b) => a + b, 0) // first emit 0+6=6, second emit 0+6+8=14
    .log() // this stage prints 6, 14
    .toArray();

  console.log(result); // Outputs: [6, 14]
}

await run();
```

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

### Transforming and Filtering Streams with sflow in TypeScript

Source: https://github.com/snomiao/sflow/blob/main/docs/Examples.md

This snippet demonstrates basic stream transformation and filtering using the `sflow` library. It doubles each number in the stream, logs the intermediate results, filters numbers greater than 6, logs again, and collects the final results into an array. It shows how to chain `map`, `log`, and `filter` operations for sequential data manipulation.

```TypeScript
import { sflow } from "sflow";

async function example1() {
  const result = await sflow([1, 2, 3, 4, 5, 6])
    .map((n) => n * 2)
    .log() // Prints: 2, 4, 6, 8, 10, 12
    .filter((n) => n > 6)
    .log() // Prints: 8, 10, 12
    .toArray();

  console.log(result); // Outputs: [8, 10, 12]
}

example1();
```

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

### Creating and Processing Stream Vectors with sflow (TypeScript)

Source: https://github.com/snomiao/sflow/blob/main/docs/Examples.md

This snippet shows how to create and manipulate a stream vector using `sflow`'s `svector` function. It demonstrates chaining operations like `uniq` (though not effective here as all are unique) and `map` to transform stream elements. This provides a concise way to create streams from a list of values.

```TypeScript
import { svector } from "sflow";

async function example12() {
  const vector = svector(1, 2, 3, 4, 5);

  const result = await vector
    .uniq()
    .map((n) => n * 2)
    .log() // Prints: 2, 4, 6, 8, 10
    .toArray();

  console.log(result); // Outputs: [2, 4, 6, 8, 10]
}

example12();
```

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

### Merging Multiple sflow Streams in TypeScript

Source: https://github.com/snomiao/sflow/blob/main/docs/Examples.md

This example shows how to merge multiple independent `sflow` streams into a single combined stream. It creates two separate streams of numbers and then uses the `merge` operator to interleave their elements into a single sequence, demonstrating concurrent processing of multiple data sources into a unified flow.

```TypeScript
import { sflow } from "sflow";

async function example4() {
  const stream1 = sflow([1, 3, 5]);
  const stream2 = sflow([2, 4, 6]);

  const result = await sflow([stream1, stream2])
    .merge()
    .log() // Prints: 1, 2, 3, 4, 5, 6
    .toArray();

  console.log(result); // Outputs: [1, 2, 3, 4, 5, 6]
}

example4();
```

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

### Custom Transformations with TransformStream in sflow (TypeScript)

Source: https://github.com/snomiao/sflow/blob/main/docs/Examples.md

This snippet demonstrates how to integrate a standard Web Streams API `TransformStream` into an sflow pipeline using the `through` operator. This allows for custom, low-level transformations on stream chunks, such as converting strings to uppercase.

```typescript
import { sflow } from "sflow";

async function example20() {
  const uppercaseTransform = new TransformStream({
    transform(chunk, controller) {
      controller.enqueue(chunk.toUpperCase());
    }
  });

  const result = await sflow(["hello", "world"])
    .through(uppercaseTransform)
    .log() // Prints: "HELLO", "WORLD"
    .toArray();

  console.log(result); // Outputs: ["HELLO", "WORLD"]
}

example20();
```

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

### Performing Sequential Async Operations in sflow (TypeScript)

Source: https://github.com/snomiao/sflow/blob/main/docs/Examples.md

This snippet shows how to integrate asynchronous operations into an `sflow` stream using `asyncMap`. It simulates a delay for each item, ensuring that subsequent items are processed only after the current item's async work is complete. This is crucial for handling I/O-bound or time-consuming tasks sequentially within a stream.

```TypeScript
import { sflow } from "sflow";
import { sleep } from "bun";

async function example16() {
  const data = [1, 2, 3, 4];

  const result = await sflow(data)
    .asyncMap(async (item) => {
      await sleep(item * 100); // Simulate async work
      return item * 2;
    })
    .log() // Prints: 2, 4, 6, 8
    .toArray();

  console.log(result); // Outputs: [2, 4, 6, 8]
}

example16();
```

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

### Unwinding Nested Arrays with sflow in TypeScript

Source: https://github.com/snomiao/sflow/blob/main/docs/Examples.md

This example illustrates how to 'unwind' a nested array within objects in an `sflow` stream. The `unwind` operator flattens the stream by creating a new object for each element in the specified nested array, duplicating the parent object's other properties. This is useful for denormalizing data structures where one-to-many relationships are represented by arrays.

```TypeScript
import { sflow } from "sflow";

async function example6() {
  const data = [
    { id: 1, values: [10, 20, 30] },
    { id: 2, values: [40, 50] }
  ];

  const result = await sflow(data)
    .unwind("values")
    .log() // Prints unwound records
    .toArray();

  console.log(result);
  // Outputs: 
  // [
  //   { id: 1, values: 10 },
  //   { id: 1, values: 20 },
  //   { id: 1, values: 30 },
  //   { id: 2, values: 40 },
  //   { id: 2, values: 50 }
  // ]
}

example6();
```

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

### Conditionally Chunking Stream Data in sflow (TypeScript)

Source: https://github.com/snomiao/sflow/blob/main/docs/Examples.md

This snippet illustrates conditional chunking of a stream using `sflow`'s `chunkIfs` utility. It processes a string character by character, chunking based on a predicate (not being a newline character). This is useful for parsing delimited data or breaking streams into logical blocks.

```TypeScript
import { sflow } from "sflow";
import { chunkIfs } from "sflow/chunkIfs";

async function example14() {
  const data = "a,b,c\n1,2,3\nd,s,f";

  const result = await sflow(data.split(""))
    .through(chunkIfs((char) => char !== "\n"))
    .map((chunk) => chunk.join(""))
    .log() // Prints: "a,b,c", "\n", "1,2,3", "\n", "d,s,f"
    .toArray();

  console.log(result); // Outputs: ["a,b,c", "\n", "1,2,3", "\n", "d,s,f"]
}

example14();
```

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

### Handling Errors in sflow Streams (TypeScript)

Source: https://github.com/snomiao/sflow/blob/main/docs/Examples.md

This snippet demonstrates how to gracefully handle and ignore specific errors within an sflow stream using the `catch` operator combined with `andIgnoreError`. It allows the stream to continue processing valid data while filtering out items that cause predefined errors.

```typescript
import { sflow } from "sflow";
import { andIgnoreError } from "sflow/andIgnoreError";

async function example18() {
  const data = [1, 2, "three", 4];

  const result = await sflow(data)
    .map((x) => {
      if (typeof x !== "number") {
        throw new Error("Not a number");
      }
      return x * 2;
    })
    .catch(andIgnoreError(/Not a number/))
    .log() // Prints: 2, 4
    .toArray();

  console.log(result); // Outputs: [2, 4]
}

example18();
```

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

### Processing Line-by-Line Stream Data with sflow in TypeScript

Source: https://github.com/snomiao/sflow/blob/main/docs/Examples.md

This snippet demonstrates processing a multi-line string stream line by line using `sflow` and the `lines` utility. It splits the input text into individual lines, transforms each line to uppercase, logs the transformed lines, and then joins them back into a single string with newline characters. This is useful for text file processing or parsing structured text data.

```TypeScript
import { sflow } from "sflow";
import { lines } from "sflow/lines";

async function example9() {
  const text = "line1\nline2\nline3";
  
  const result = await sflow(text)
    .through(lines())
    .map((line) => line.toUpperCase())
    .log() // Prints: LINE1, LINE2, LINE3
    .join("\n")
    .text();

  console.log(result); // Outputs: LINE1\nLINE2\nLINE3
}

example9();
```

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

### Processing CSV Data with sflow in TypeScript

Source: https://github.com/snomiao/sflow/blob/main/docs/Examples.md

This snippet demonstrates parsing and formatting CSV data using `sflow` with `csvParses` and `csvFormats` from `sflow/xsvStreams`. It takes a CSV string, parses it into objects, transforms the 'age' field to a number, logs the parsed records, and then formats the modified records back into a CSV string, showcasing end-to-end CSV manipulation.

```TypeScript
import { sflow } from "sflow";
import { csvParses, csvFormats } from "sflow/xsvStreams";

async function example5() {
  const csvData = "name,age\nJohn,30\nJane,25";
  
  const result = await sflow(csvData)
    .through(csvParses("name,age"))
    .map((record) => ({ ...record, age: Number(record.age) }))
    .log() // Prints parsed records
    .through(csvFormats(["name", "age"]))
    .text();

  console.log(result); // Outputs formatted CSV with modified records
}

example5();
```

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

### Piping Data with SNOFLOW Kernels via Native WebStream (TypeScript)

Source: https://github.com/snomiao/sflow/blob/main/docs/README_deprecated.md

This example illustrates how to integrate SNOFLOW kernels with native WebStreams using the `pipeThrough` method. This approach leverages the standard WebStream API, enabling tree-shaking for optimized bundle sizes. It demonstrates similar stream operations as the simple pipe style but by explicitly using kernel functions like `buffers`, `debounces`, `maps`, etc., and finally piping to a `nils()` sink.

```TypeScript
await new ReadableStream({
  start: (ctrl) => {
    [1, 2, 3].map((x) => ctrl.enqueue(x));
  },
})
  .pipeThrough(buffers(2))
  .pipeThrough(debounces(100))
  .pipeThrough(filters())
  .pipeThrough(maps((n) => [String(n)]))
  .pipeThrough(flats())
  .pipeThrough(flatMaps((n) => [String(n)]))
  .pipeThrough(teess((s) => s.pipeTo(nils())))
  .pipeThrough(limits(1))
  .pipeThrough(maps(() => 1))
  .pipeThrough(peeks(() => {}))
  .pipeThrough(
    reduces((a, b) => a + b),
    0,
  )
  .pipeThrough(skips(1))
  .pipeThrough(tails(1))
  .pipeThrough(throttles(100))
  .pipeTo(nils());
```

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

### Applying Core Stream Transformations with sflow in TypeScript

Source: https://github.com/snomiao/sflow/blob/main/README.md

This code provides examples of common stream transformation methods like `map`, `filter`, and `reduce` available in sflow. These methods enable modifying stream elements, selecting elements based on a condition, and aggregating elements into a single value, respectively, demonstrating fundamental functional programming patterns.

```typescript
// Mapping
flow1.map((n) => n * 2);

// Filtering
flow1.filter((n) => n % 2 === 0);

// Reducing
flow1.reduce((a, b) => a + b, 0);
```

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

### Utilizing Advanced Stream Control and Merging with sflow in TypeScript

Source: https://github.com/snomiao/sflow/blob/main/README.md

This code showcases advanced sflow utilities for controlling stream flow and combining multiple streams, including `throttle` for rate limiting, `debounce` for delaying emissions, `toArray` for collecting all elements, and `merge` for combining streams. It also includes an example of `chunkIfs` for custom splitting, demonstrating powerful stream manipulation techniques.

```typescript
// Throttling
flow1.throttle(100);

// Debouncing
flow1.debounce(200);

// Converting to array
flow1.toArray();

// Merging multiple streams
const mergedFlow = sflow([flow1, flow2]).merge();

// Use chunkIf to split tokens by line
await sflow("a,b,c\n\n1,2,3\nd,s,f".split(""))
  .through(chunkIfs((e: string) => e.indexOf("\n") === -1))
  .map((chars) => chars.join(""))
  .toArray(); // ["a,b,c\n",'\n', "1,2,3\n", "d,s,f"]
```

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

### Debouncing and Throttling Streams with sflow in TypeScript

Source: https://github.com/snomiao/sflow/blob/main/docs/Examples.md

This snippet demonstrates applying `debounce` and `throttle` operations to an `sflow` stream. It debounces events by 100ms, ensuring only the last event in a rapid sequence is processed, and then throttles events by 200ms, dropping subsequent events within the throttle window. The `sleep` dependency simulates asynchronous event timing for realistic behavior.

```TypeScript
import { sflow } from "sflow";
import { sleep } from "bun";

async function example3() {
  const result = await sflow([1, 2, 3, 4, 5])
    .debounce(100)
    .log() // Prints debounced numbers
    .throttle(200, { drop: true })
    .log() // Prints throttled numbers
    .toArray();

  console.log(result); // Outputs: [2, 4]
}

example3();
```

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

### Processing Streams Concurrently with sflow pMap in TypeScript

Source: https://github.com/snomiao/sflow/blob/main/docs/Examples.md

This snippet demonstrates using the `pMap` operator in `sflow` to process stream items concurrently. It simulates asynchronous work for each number and processes up to 2 items in parallel, showcasing how to improve performance for I/O-bound or CPU-bound tasks within a stream pipeline. The `sleep` dependency simulates the async work, highlighting real-world use cases.

```TypeScript
import { sflow } from "sflow";
import { sleep } from "bun";

async function example7() {
  const result = await sflow([1, 2, 3, 4])
    .pMap(async (n) => {
      await sleep(n * 100); // Simulate async work
      return n * 2;
    }, { concurrency: 2 })
    .log() // Prints: 2, 4, 6, 8
    .toArray();

  console.log(result); // Outputs: [2, 4, 6, 8]
}

example7();
```

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

### Piping Data with SNOFLOW's Simple Pipe Style (TypeScript)

Source: https://github.com/snomiao/sflow/blob/main/docs/README_deprecated.md

This snippet demonstrates the concise piping syntax of SNOFLOW, allowing chaining of various stream operations like buffering, debouncing, filtering, mapping, flattening, peeking, reducing, and throttling. It processes an array of numbers, transforms them, and performs aggregations before completing the stream. This style offers a clear and readable way to define complex data flows.

```TypeScript
await sflow([1, 2, 3])
  .buffer(2)
  .debounce(100)
  .filter()
  .map((n) => [String(n)])
  .flat()
  .flatMap((n) => [String(n)])
  .tees((s) => s.pipeTo(nils()))
  .limit(1)
  .map(() => 1)
  .peek(() => {})
  .reduce(0, (a, b) => a + b)
  .skip(1)
  .tail(1)
  .throttle(100)
  .done();
```

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

### Chunking and Buffering sflow Streams in TypeScript

Source: https://github.com/snomiao/sflow/blob/main/README.md

This snippet demonstrates sflow's capabilities for organizing stream data into chunks based on different criteria such as count, time intervals, or custom logic. These methods are crucial for processing data in batches, managing flow control, or grouping related elements within a stream.

```typescript
// Chunking by count
flow1.chunk(2); // [[1, 2], [3, 4]]

// Buffering within a time interval
flow1.chunkByInterval(1000);

// Custom chunking
flow1.chunkBy((x) => Math.floor(x / 2));
```

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

### Ensuring Type Safety with sflow and TypeScript Unwind/MapAddField

Source: https://github.com/snomiao/sflow/blob/main/README.md

This snippet illustrates sflow's type-safe features when used with TypeScript, specifically demonstrating the `unwind` method to flatten nested arrays within objects and `mapAddField` to add new properties based on existing ones. This highlights how sflow leverages TypeScript for robust and predictable data transformations on complex structures.

```typescript
import { sflow } from "sflow";

const typedFlow = sflow([{ a: 1, b: [1, 2, 3] }])
  .unwind("b") // Use `unwind` for objects with nested arrays
  .mapAddField("newField", (item) => item.a + item.b);
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.