使用Thi.ng Transducers实现多路复用并保留最新值

在Thi.ng Transducers库中处理多路数据流时，开发者经常需要将多个转换管道并行执行，同时保留每个管道的最新有效值。本文将深入探讨如何实现这一需求，并介绍几种不同的解决方案。

问题背景

当使用multiplex函数并行处理多个转换管道时，每个管道可能有不同的处理长度。例如：

multiplex(
  comp(take(2), map(t => "a"+t)),
  comp(take(3), map(t => "b"+t))
)

对于输入序列0-4，默认输出会是：

[["a0","b0"],["a1","b1"],[undefined,"b2"],[undefined,undefined]]

但实际需求可能是保留每个管道的最新有效值，输出应为：

[["a0","b0"],["a1","b1"],["a1","b2"]]

解决方案一：使用partitionSync

partitionSync是Thi.ng库中专门用于同步多个数据流的工具。它按照用户定义的键函数分组输入值，并发出对象元组。只有当所有定义的输入/组/键都至少传递了一个值时，它才开始产生结果。

实现代码示例：

const xform = comp(
  multiplex(...),
  flatten1(),
  keep(),
  partitionSync(["a", "b"], { key: (x) => x[0], reset: false }),
  map(({ a, b }) => [a, b])
);

这种方法虽然功能强大，但会产生一些中间结果，可能不是最高效的解决方案。

解决方案二：自定义syncTuples转换器

Thi.ng团队为此专门开发了一个新的转换器syncTuples，它接收固定大小的可能为null的元组，并只发出完全填充的元组（具有非null值），跟踪每个组件的最后一个有效值，并在需要时使用它们来填充空组件。

关键实现逻辑：

1. 维护一个数组保存每个位置的最新有效值
2. 检查当前输入元组是否有任何新值
3. 只有当所有位置都有有效值时才输出结果

const syncTuples = <T>(size: number): Transducer<Nullable<T>[], T[]> => ...

这个转换器已被正式加入Thi.ng Transducers库9.2.0版本中。

处理无限数据流

当处理无限数据源时，可以结合使用takeWhile来提前终止处理：

const xform = comp(
  multiplex(...),
  takeWhile((tuple) => tuple.some((x) => x != null)),
  syncTuples(2)
);

或者使用step函数逐个处理输入值，实现更精细的控制：

const fn = step(xform);
for (let x of range()) {
  const res = fn(x);
  if (!res) break;
  console.log(res);
}

实际应用场景

这种技术在以下场景特别有用：

1. 从多个传感器同步数据
2. 处理不同长度的并行数据流
3. 构建UI组件需要多个异步数据源
4. 动画系统中处理不同属性的时间线

通过Thi.ng Transducers提供的这些工具，开发者可以轻松构建复杂的数据流处理管道，同时保持代码的简洁性和可维护性。