串流帮助

顶上建有一组安全类型流帮助器 效果 提高能效和 控制流处理.

包含的帮助者

过滤器

这个 filter 帮助缩小流,只通过匹配的项目 上观相.

import { filter } from "@effectionx/stream-helpers";
import { each } from "effection";

// Example: Synchronous filtering
function* syncExample(source: Stream<number, unknown>) {	

  const gt5 = filter<number>(function* (x) { return x > 5 });

  for (const value of yield* each(gt5(stream))) {
    console.log(value); // Only values > 5
    yield* each.next();
  }
};

// Example: Asynchronous filtering
function* asyncExample(source: Stream<number, unknown>) {

  const evensOf = filter<number>(function* (x) {
    yield* sleep(100); // Simulate async operation
    return x % 2 === 0; // Keep only even numbers
  });

  for (const value of yield* each(evensOf(stream))) {
    console.log(value); // Only even numbers
    yield* each.next();
  }
});

地图

这个 map 帮助在流中使用提供的函数转换每个项目. 用于您需要处理每个数据转换操作 各个项目.

import { map } from "@effectionx/stream-helpers";
import { each } from "effection";

function* example(stream: Stream<number, unknown>) {
  const double = map<number>(function* (x) {
    return x * 2;
  });

  for (const value of yield* each(double(stream))) {
    console.log(value); // Each value is doubled
    yield* each.next();
  }
}

减少

这个 reduce 帮助工具通过应用在流中转换每个项目 累积值.

import { reduce } from "@effectionx/stream-helpers";
import { each } from "effection";

function* example(stream: Stream<number, unknown>) {
  let sum = reduce(function* (total: number, current: number) {
    return total + current;
  }, 0);

  for (let value of yield* each(sum(streamOf([1,2,3]))) {
    console.log(value) // logs 1 -> 3 -> 6
    yield* each.next();
  }
}

批次

这个 batch 帮助工具在您想要转换单个通过的项目时有用 通过流入数组项目。 批量可以由 指定最大时间或最大大小。 如果两者都明确,则批次 将在满足任一条件时创建.

import { batch } from "@effectionx/stream-helpers";
import { each } from "effection";

// Example: Batch by size
function* exampleBySize(stream: Stream<number, unknown>) {
  const byThree = batch({ maxSize: 3});

  for (const items of yield* each(byThree(stream))) {
    console.log(batch); // [1, 2, 3], [4, 5, 6], ...
    yield* each.next();
  }
};

// Example: Batch by time
function* exampleByTime(stream: Stream<number, unknown>) {
  const stream = batch({ maxTime: 1000 })(sourceStream);

  for (const batch of yield* each(stream)) {
    console.log(batch); // Items received within 1 second
    yield* each.next();
  }
});

// Example: Combined batching
function* exampleCombined(stream: Stream<number, unknown>) {

  const batched = batch({
    maxSize: 5,
    maxTime: 1000,
  });

  for (const batch of yield* each(batched(stream))) {
    console.log(batch); // Up to 5 items within 1 second
    yield* each.next();
  }
});

阀门

允许对源流应用回压来防止压倒 下游消费者。 这对任何生成更快项目的流都有用 比消费者可以消费它们。 它最初设计用于: Kafka , 制作人可以让服务耗尽内存 制作人生产出许多比消费者更快的处理信件. 它会 用于任何无限流的缓冲.

import { valve } from "@effectionx/stream-helpers";
import { each } from "effection";

function* example() {
  const regulated = valve({
    // buffer size threshold when close operation will invoked
    closeAt: 1000,
    *close() {
      // pause the source stream
    },

    // buffer size threshold when open operation will be invoked
    openAt: 100,
    *open() {
      // resume the source stream
    },
  })(stream);

  for (const value of yield* each(regulated)) {
    console.log(value);
    yield* each.next();
  }
}

下水道

这个 drain 帮助将流耗尽,丢弃所有生成的值,然后返回 近值。 当你只关心最后的结果时,这个是有用的 流,而不是中间值.

import { drain } from "@effectionx/stream-helpers";

function* example() {
  // Get the response from a request channel (ignoring any progress)
  const channel = yield* transport.send(request);
  const response = yield* drain(channel);
  console.log(response); // The close value
}

第一届

这个 first 帮助返回由流生成的第一个值,或 undefined 如果流关闭而不产生任何值。 使用 first.expect() 要掷出 错误而不是返回 undefined。 。 。 .

import { first } from "@effectionx/stream-helpers";
import { streamOf } from "@effectionx/stream-helpers";

function* example() {
  const stream = streamOf([1, 2, 3]);
  const value = yield* first(stream);
  console.log(value); // 1

  const empty = streamOf([]);
  const none = yield* first(empty);
  console.log(none); // undefined

  // Use first.expect() to throw if stream is empty
  const required = yield* first.expect(stream); // throws if empty
}

最后一个

这个 last 帮助返回由流产生的最后一个值,或 undefined 如果流关闭而不产生任何值。 它会耗尽整个溪流 查找最后一个值。 使用 last.expect() 来丢出一个错误而不是 返回时 undefined。 。 。 .

import { last } from "@effectionx/stream-helpers";
import { streamOf } from "@effectionx/stream-helpers";

function* example() {
  const stream = streamOf([1, 2, 3]);
  const value = yield* last(stream);
  console.log(value); // 3

  const empty = streamOf([]);
  const none = yield* last(empty);
  console.log(none); // undefined

  // Use last.expect() to throw if stream is empty
  const required = yield* last.expect(stream); // throws if empty
}

拿着

这个 take 帮助创建第一个生成的流变压器 n 数值 从源流中取出,然后用最后的取值关闭.

import { take, forEach, streamOf } from "@effectionx/stream-helpers";
import { pipe } from "remeda";

function* example() {
  const stream = streamOf([1, 2, 3, 4, 5]);
  // yields 1, 2, then closes with 3
  const closeValue = yield* forEach(function* (value) {
    console.log(value); // 1, then 2
  }, take(3)(stream));
  console.log(closeValue); // 3
}

// Works with pipe
const limited = pipe(source, take(3));

慢慢来

这个 takeWhile 帮助创建流式变压器,在同时生成值 上游返回真实。 当上游返回时,溪会关闭 立即(不包括失败的值).

import { takeWhile, forEach, streamOf } from "@effectionx/stream-helpers";
import { pipe } from "remeda";

function* example() {
  const stream = streamOf([1, 2, 3, 4, 5]);
  // yields 1, 2 (stops when value >= 3)
  yield* forEach(function* (value) {
    console.log(value); // 1, then 2
  }, takeWhile((x) => x < 3)(stream));
}

// Works with pipe
const filtered = pipe(source, takeWhile((x) => x.isValid));

使用到结束

这个 takeUntil 帮助创建流式变压器,该变压器输出值直到 上游返回真实。 当上游匹配时,溪流会关闭 匹配值。 这对“在满足条件之前”有用 模式.

import { takeUntil, forEach } from "@effectionx/stream-helpers";
import { pipe } from "remeda";

function* example() {
  // Iterate validation progress until we get a terminal status
  const result = yield* forEach(
    function* (progress) {
      showSpinner(progress.status);
    },
    takeUntil((p) => p.status === "valid" || p.status === "invalid")(
      validationStream,
    ),
  );

  // result is the validation object with terminal status
  if (result.status === "valid") {
    console.log("Validation passed!");
  }
}

// Works with pipe
const untilDone = pipe(source, takeUntil((x) => x.done));

关键差异 takeWhile 财务报告和审定财务报表 takeUntil数字 :

每个

这个 forEach 帮助为每个通过流的项目引用函数. 当您需要在每个项目上执行副作用或操作时,此功能是有用的 不改变溪流本身。 不同于返回的其他流帮助 变了的溪流, forEach 消耗整个溪流并返回 流的近值.

import { forEach } from "@effectionx/stream-helpers";
import { createSignal, spawn } from "effection";

function* example() {
  const stream = createSignal<number, void>();

  // Process each item in the stream
  yield* spawn(() =>
    forEach(function* (item) {
      console.log(`Processing: ${item}`);
      // Perform any side effects here
    }, stream)
  );

  stream.send(1);
  stream.send(2);
  stream.send(3);
  stream.close();
}

// Example: Handling stream close value
function* exampleWithCloseValue() {
  const stream = createSignal<string, number>();

  const result = yield* spawn(() =>
    forEach(function* (item) {
      console.log(`Item: ${item}`);
    }, stream)
  );

  stream.send("hello");
  stream.send("world");
  stream.close(42); // Close with value 42

  const closeValue = yield* result;
  console.log(`Stream closed with: ${closeValue}`); // 42
}

议题

主题帮助器将任意流转换成重播的多播流 新用户的最新值。 类似 RxJS 行为主体。 。 。 .

import { createSubject } from "@effectionx/stream-helpers";
import { createChannel, spawn } from "effection";

function* example() {
  const subject = createSubject<number>();
  const channel = createChannel<number, void>();
  const downstream = subject(channel);

  // First subscriber
  const sub1 = yield* downstream;

  yield* channel.send(1);
  yield* channel.send(2);

  console.log(yield* sub1.next()); // { done: false, value: 1 }
  console.log(yield* sub1.next()); // { done: false, value: 2 }

  // Late subscriber gets the latest value immediately
  const sub2 = yield* downstream;
  console.log(yield* sub2.next()); // { done: false, value: 2 }
}

使用管道操作符将任何流转换成行为主体 :

import { createSubject, map } from "@effectionx/stream-helpers";
import { pipe } from "remeda";

const subject = createSubject<string>();

const stream = pipe(
  source,
  map(function* (x) {
    return x.toString();
  }),
  subject,
);

通过跟踪器

通过 Tracker 流帮助提供了一种了解所有项目的方法 穿过溪流已经处理完毕。 这对你特别有帮助 想要确保所有物品在操作完成前都得到处理.

和其他小溪帮手不一样 因为你必须先打电话 createTracker 函数重击对象。 实际的帮助者在 passthrough 可以调用的方法,并与其他方法一样链条 帮手.

import { each, signal } from "effection";
import { createTracker } from "@ffectionx/stream-helpers"

const source = signal(0);

// create the tracker
const tracker = yield* createTracker();

// create  passthrough stream helper
const track = tracker.passthrough();

for (const value of yield* each(track(source))) {
  // mark items 
  tracker.markOne(value);
  yield* each.next();
}

// will resolve when all items that passed through the stream were seen
yield* tracker;

编组流帮助

你可以用简单的 pipe() 组成一系列的溪流帮手 单位 这个例子,我们用一个来自 医生, (中文(简体) ),

import { batch, filter, forEach, map, valve } from "@effectionx/stream-helpers";
// any standard pipe function should work
import { pipe } from "remeda";

function* example(source: Stream<number, unknown>) {
  // Compose stream helpers using pipe
  const stream = pipe(
    source,
    valve({ open, close, openAt: 100, closeAt: 100 }),
    filter(function* (x) {
      return x > 0;
    }),
    map(function* (x) {
      return x * 20;
    }),
    batch({ maxSize: 50 }),
  );

  // Use forEach to process each value in the composed stream
  yield* forEach(function* (value) {
    console.log(value);
  }, stream);
}

测试流

库包括测试工具, 帮助您测试您的流处理 代码。 这些表格载于 @effectionx/stream-helpers/test-helpers 输出.

福赛特

这个 useFaucet 函数创建可用于测试行为的流 使用后压的溪流 在你的测试中特别有用 需要一个可控源流 .

import { useFaucet } from "@effectionx/stream-helpers/test-helpers";
import { each, run, spawn } from "effection";

await run(function* () {
  const faucet = yield* useFaucet<number>({ open: true });

  // Remember to spawn the stream subscription before sending items to the stream
  yield* spawn(function* () {
    for (let i of yield* each(faucet)) {
      console.log(i);
      yield* each.next();
    }
  });

  // Pass an array of items to send items to the stream one at a time synchronously
  yield* faucet.pour([1, 2, 3]);

  // Pass an operation to control the rate at which items are sent to the stream
  yield* faucet.pour(function* (send) {
    yield* sleep(10);
    yield* send(5);
    yield* sleep(30);
    yield* send(6);
    yield* sleep(10);
    yield* send(7);
  });

  // You can close the faucet to stop items from being sent
  faucet.close();

  // And open it again when needed
  faucet.open();
});

当函数传递到 faucet.pour,则 send 函数将返回一个 操作。 水龙头关闭后 行动就会被阻止 这是 在测试后压时特别有用, 因为您可以发送许多事件 不用担心 open/close 状态.