FxTS

website·Indexed 32 days ago

A functional programming library for JavaScript and TypeScript that utilizes Iterable and AsyncIterable. It provides tools for method chaining, lazy evaluation, function composition, and concurrency handling, including a concurrentPool for managing uneven workloads and processing LLM token streams.

Tokens: 43.8K
Snippets: 339
Records: 415
Agent score: 56%

What's inside FxTS

Understand Lazy Evaluation in FxTS
FxTS provides lazy evaluation for Iterable and AsyncIterable data sets. Unlike native Array methods (like .filter() or .map()) which create new arrays and traverse the entire collection at each step, FxTS lazy functions evaluate values only as needed. This allows for efficient processing of large or infinite data sets and enables early termination (e.g., using take(n)) without processing the entire source.
```
import { pipe, range, filter, map, take, reduce } from 'fxts';

const sum = (a: number, b: number) => a + b;

pipe(
  range(Infinity), // Infinite sequence
  filter((a) => a % 2 === 0),
  map((a) => a * a),
  take(2), // Stops evaluation after 2 matches are found
  reduce(sum),
); // Result: 4
```
https://fxts.dev/llms-full.txt?llms-doc=lazy-evaluationwebsitellms
Compose functions using pipe
The pipe function allows you to connect multiple functions by passing the output of one function as the input to the next. This avoids deeply nested function calls and improves readability, especially when performing a sequence of operations like filter, map, and reduce.
```
import { filter, map, pipe, reduce } from "@fxts/core";

const sum = (a: number, b: number) => a + b;
const arr = [1, 2, 3, 4, 5];

pipe(
  arr,
  filter((a) => a % 2 === 0),
  map((a) => a + 10),
  reduce(sum),
);
```
https://fxts.dev/llms-full.txt?llms-doc=function-compositionwebsitellms
Use toAsync to handle asynchronous callbacks in a pipe
The toAsync function is used within a pipe to transition a data stream into an asynchronous context. This is necessary when you want to use an async callback function (e.g., inside partition) or when the input data consists of Promises. It allows the pipe to handle AsyncIterable data and asynchronous effects.
```
// Example 1: Using an asynchronous callback function
await pipe(
  Promise.resolve([1, 2, 3, 4, 5]),
  toAsync,
  partition(async (a) => a % 2 === 0), // [[2, 4], [1, 3, 5]]
);

// Example 2: Handling a list of Promises
await pipe(
  [Promise.resolve(1), Promise.resolve(2), Promise.resolve(3), Promise.resolve(4), Promise.resolve(5)],
  toAsync,
  partition((a) => a % 2 === 0), // [[2, 4], [1, 3, 5]]
);
```
https://fxts.dev/llms-full.txt?llms-doc=when-to-use-the-toasync-function-part-192websitellms

Handle asynchronous predicates or streams with toAsync

When working with asynchronous data sources (like a Promise of an array or an array of Promises) or when the predicate function in takeUntil is asynchronous, use the toAsync function to convert the stream into an AsyncIterable. This allows the pipe to be awaited and supports async callbacks.

import { pipe, toAsync, takeUntil, toArray } from 'fxts';

// Case 1: Using an asynchronous callback predicate
await pipe(
 Promise.resolve([1, 2, 3, 4, 5, 6]),
 toAsync,
 takeUntil(async (a) => a % 2 === 0),
 toArray,
); // [1, 2]

// Case 2: Processing a stream of Promises
await pipe(
 [Promise.resolve(1), Promise.resolve(2), Promise.resolve(3), 
  Promise.resolve(4), Promise.resolve(5), Promise.resolve(6)],
 toAsync,
 takeUntil(a => a % 2 === 0),
 toArray,
); // [1, 2]

Process LLM token streams using FxTS pipelines

Since major LLM SDKs (like Anthropic and OpenAI) return streams as AsyncIterable, they can be passed directly into FxTS pipe without adapters. This allows for lazy processing of token events, such as filtering for specific event types and mapping to text content.

import Anthropic from "@anthropic-ai/sdk";
import { each, filter, map, pipe } from "@fxts/core";

const client = new Anthropic();
const stream = await client.messages.create({
  model: "claude-sonnet-5",
  max_tokens: 1024,
  messages: [{ role: "user", content: "Write a haiku about iterators" }],
  stream: true,
});

await pipe(
  stream,
  filter((event) => event.type === "content_block_delta"),
  map((event) => ("text" in event.delta ? event.delta.text : "")),
  each((text) => process.stdout.write(text)),
);

Handle arrays of promises using toAsync with zipWithIndex
When dealing with an array where the elements themselves are promises, use toAsync before other lazy operators like zipWithIndex. This ensures that the promises are resolved before the indexing logic is applied, allowing the final result to be awaited as a resolved array of indexed values.
```
import { pipe, toAsync, zipWithIndex, toArray } from 'fxts';

await pipe(
 [Promise.resolve("a"), Promise.resolve("b"), Promise.resolve("c"), Promise.resolve("d")],
 toAsync,
 zipWithIndex,
 toArray,
);
// Result: [[0, "a"], [1, "b"], [2, "c"], [3, "d"]]
```
https://fxts.dev/llms-full.txt?llms-doc=when-to-use-the-toasync-function-part-296websitellms
Use the fx() function to enable method chaining in fxts
The fx function allows developers to use functions provided by fxts within a method chaining sequence. This is particularly useful for functions that are not natively available as methods on the iterable; these can be connected via the chain method when wrapped with fx.
```
declare function fx<T extends Iterable<unknown> | AsyncIterable<unknown>>(a: T): T extends Iterable<unknown> ? FxIterable<IterableInfer<T>> : FxAsyncIterable<IterableInfer<T>>;
```
https://fxts.dev/llms-full.txt?llms-doc=when-to-use-the-toasync-function-part-88websitellms
Convert a synchronous iterable to an asynchronous one using `toAsync`
Use toAsync within a pipe to transform a synchronous iterable (or an array of promises) into an AsyncIterable. This is necessary when you need to use asynchronous callbacks (e.g., an async function inside filter) or when the source data consists of promises that need to be resolved sequentially during iteration.
```
// Case 1: Using an asynchronous callback in a filter
await pipe(
  Promise.resolve([0, 1, 2, 3, 4, 5, 6]),
  toAsync,
  filter(async a => a % 2 === 0),
  toArray,
);

// Case 2: Handling an array of promises
await pipe(
  [Promise.resolve(0), Promise.resolve(1), Promise.resolve(2)],
  toAsync,
  filter(a => a % 2 === 0),
  toArray,
);
```
https://fxts.dev/llms-full.txt?llms-doc=when-to-use-the-toasync-function-part-75websitellms

Convert iterables to asynchronous streams using toAsync

Use the toAsync function within a pipe to convert a synchronous iterable or a collection of promises into an AsyncIterable. This is necessary when you need to use an asynchronous predicate function (an async callback) within operators like takeUntilInclusive or when the source data consists of promises.

import { pipe, toAsync, takeUntilInclusive, toArray } from 'fxts';

// Scenario 1: Using an asynchronous predicate callback
await pipe(
 Promise.resolve([1, 2, 3, 4, 5, 6]),
 toAsync,
 takeUntilInclusive(async (a) => a % 2 === 0),
 toArray,
); // [1, 2]

// Scenario 2: Handling a source array of Promises
await pipe(
 [
  Promise.resolve(1), 
  Promise.resolve(2), 
  Promise.resolve(3), 
  Promise.resolve(4), 
  Promise.resolve(5), 
  Promise.resolve(6)
 ],
 toAsync,
 takeUntilInclusive(a => a % 2 === 0),
 toArray,
); // [1, 2]

Process asynchronous tasks concurrently with concurrentPool

Use concurrentPool in a pipeline to process items concurrently. Unlike p-map, FxTS allows results to be consumed as a stream (using each or filter) as they arrive, rather than waiting for the entire array to complete. This enables early termination: combining concurrentPool with take will stop scheduling new work once the required number of results is reached.

import { concurrentPool, each, map, pipe, toAsync, filter, take, toArray } from "@fxts/core";

// Example 1: Consuming results as a stream
await pipe(
  toAsync(urls),
  map((url) => fetchItem(url)),
  concurrentPool(5),
  each((item) => render(item)), // handled as each result arrives, in order
);

// Example 2: Early termination with take(3)
const firstThree = await pipe(
  toAsync(candidates),
  map((c) => probe(c)),
  concurrentPool(5),
  filter((r) => r.alive),
  take(3), // stops pulling once three passed - remaining work is not started
  toArray,
);

Handle asynchronous iterables with toAsync and dropRight
When dealing with a collection of promises or an asynchronous data source, use toAsync before applying operations like dropRight. This converts the iterable into an AsyncIterable, allowing the pipeline to correctly await and process the elements before dropping the specified number from the end.
```
await pipe(
 [Promise.resolve(1), Promise.resolve(2), Promise.resolve(3), Promise.resolve(4)],
 toAsync,
 dropRight(2),
 toArray,
); // [1, 2]
```
https://fxts.dev/llms-full.txt?llms-doc=when-to-use-the-toasync-function-part-59websitellms

Handle asynchronous streams with `toAsync` and `reduce`

When working with Promises or asynchronous callbacks in a pipeline, use toAsync to convert the stream into an asynchronous sequence. This allows reduce and other operators to process the data asynchronously. reduce also supports partial lazy forms for backward compatibility with Promises.

import { pipe, map, filter, reduce, toAsync } from 'fxts';
const sum = (a: number, b: number) => a + b;

// 1. Handling a Promise of an array (partial lazy form)
await pipe(
 Promise.resolve([1, 2, 3, 4]),
 map((a) => a + 10),
 filter(a => a % 2 === 0),
 reduce(sum),
); // 26

// 2. Using toAsync for asynchronous callbacks
await pipe(
 Promise.resolve([1, 2, 3, 4]),
 toAsync,
 map(async (a) => a + 10),
 filter(a => a % 2 === 0),
 reduce(sum),
); // 26

// 3. Using toAsync for an array of Promises
await pipe(
 [Promise.resolve(1), Promise.resolve(2), Promise.resolve(3), Promise.resolve(4)],
 toAsync,
 map(a => a + 10),
 filter(a => a % 2 === 0),
 reduce(sum),
); // 26