FxTS Documentation

When using the concurrent operator, FxTS behaves similarly to Promise.all. If an error occurs in one of the concurrent tasks, other tasks that have already been initiated will continue to be evaluated. The error is caught by the try-catch block once the current batch of concurrent requests is processed or the error is propagated.

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

const fetchAsyncError = (a) => {
  if (a === 3) {
    return Promise.reject(`err ${a}`);
  }
  return a;
};

try {
  await pipe(
    [
      Promise.resolve(1),
      Promise.resolve(2),
      Promise.resolve(3), // When this item is evaluated, `map` function throws an error.
      Promise.resolve(4), // This item is also evaluated.
      Promise.resolve(5), // Is is not evaluated from this item.
      Promise.resolve(6),
    ],
    toAsync,
    map(fetchAsyncError),
    filter((a) => a % 2 === 0),
    concurrent(2), // request 2
    toArray,
  );
} catch (err) {
  // handle err
}

The position of concurrent in a pipeline determines which stage of the process is parallelized.

1. Applying to the data stream: If concurrent is placed after a map or filter that produces asynchronous values, it controls the concurrency of those specific asynchronous operations.
2. Behavior with Iterable: concurrent always applies to the Iterable before the length is changed by subsequent operators like take.

If you need to evaluate an asynchronous predicate within a filter concurrently, you must ensure the stream is prepared correctly. For example, if you want to sequentially map values but then run the filter predicate concurrently, you might need to collect values into an array first or place concurrent specifically after the asynchronous filter stage.

// Example: Evaluating an asynchronous predicate in filter concurrently
await pipe(
  range(Infinity),
  toAsync,
  map(fetchApi),
  toArray,
  filter((a) => delay(100, a % 2 === 0)),
  take(3),
  concurrent(3),
  each(console.log),
);

FxTS supports lazy evaluation, allowing you to process large or infinite data sets efficiently. Unlike standard array methods (like Array.prototype.filter or map) which create new intermediate arrays and traverse the entire collection at every step, FxTS uses Iterable and AsyncIterable to evaluate values only as needed.

When using pipe with lazy functions, operations like take(n) will stop the evaluation of the upstream pipeline as soon as the requirement is met. This prevents unnecessary computations and traversals.

Key distinctions:

• Lazy functions: Do not evaluate actual values immediately. They return an iterable structure that represents the computation pipeline (similar to a generator).
• Strict functions: These are terminal operations that trigger the actual evaluation of the lazy pipeline (e.g., toArray, reduce, head).

// Standard array approach (Eager/Not Lazy)
// Creates intermediate arrays at every step
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
  .filter((a) => a % 2 === 0)
  .map((a) => a * a)
  .reduce((a, b) => a + b);

// FxTS approach (Lazy)
// Only evaluates what is necessary to satisfy 'take(2)'
pipe(
  [0, 1, 2, 3, 4, 5, 6, 7, 8, 9],
  filter((a) => a % 2 === 0),
  map((a) => a * a),
  take(2),
  reduce((a, b) => a + b),
);

In lazy-evaluated pipelines, functions return an IterableIterator rather than the actual data. Because nothing is processed until a terminal function like toArray is called, using tap will only show you the IterableIterator object itself, which is not helpful for inspecting the data flowing through the pipeline.

To inspect the actual values being yielded during lazy evaluation, use the peek function instead of tap. peek allows you to observe each item as it is lazily produced by the iterator.

// Using tap in lazy pipelines only shows the iterator object:
pipe(
  range(1, Infinity),
  map(addDateFrom(new Date(2000, 0, 1))),
  filter(is13thOfFriday),
  tap(console.log), // Logs: IterableIterator
  map(formatYYYYMMDD),
  tap(console.log), // Logs: IterableIterator
  take(5),
  toArray,
);

// Use peek to see the actual values during lazy evaluation:
pipe(
  range(1, Infinity),
  map(addDateFrom(new Date(2000, 0, 1))),
  filter(is13thOfFriday),
  peek(console.log), // Logs actual values as they are yielded
  map(formatYYYYMMDD),
  peek(console.log), // Logs actual values as they are yielded
  take(5),
  toArray,
);