RxJS中如何按换行符条件缓冲流值实现逐行发射输出

问题背景 #

现有字符串流需要实现按换行符切分、逐行发射值的能力：

• 输入流定义：

let stream$ = from(['hello\n', 'world ', ' at\nhome\n and', ' more'])

• 预期流依次发射值：

'hello' 
'world  at'
'home'
' and more'

初始实现的问题 #

最初尝试通过外部Subject配合buffer做切分，代码如下：

let break$ = new Subject()
stream$.pipe(
    flatMap(x => x.match(/[^\n]+\n?|\//g)),
    map(x => {
      if (x.endsWith('\n')) {
        break$.next(true)
        return x
      }
      return x
    })
    .buffer(break$)
)

该实现运行后仅发射单个聚合数组，不符合预期：

[ 'hello\n', 'world ', ' at\n', 'home\n', ' and', ' more' ]

预期的分组结构应为：

[
  ['hello\n'],
  ['world ', ' at\n'],
  ['home\n'],
  [' and', ' more'],
]

现有可运行方案会主动触发订阅，需要纯管道形式的实现，保留流的懒评估特性。

实现方案 #

不需要创建外部Subject，全程在管道内通过状态累计即可实现，不会提前触发订阅，完全符合懒加载要求：

import { from } from 'rxjs';
import { mergeMap, scan, endWith } from 'rxjs/operators';

const stream$ = from(['hello\n', 'world ', ' at\nhome\n and', ' more']);

const lineStream$ = stream$.pipe(
  // 流结束时注入标记，触发最后一段未尾随换行的内容输出
  endWith(null),
  scan((acc, chunk) => {
    // 处理流结束逻辑
    if (chunk === null) {
      return {
        lines: acc.remaining ? [acc.remaining] : [],
        remaining: ''
      };
    }
    // 拼接上一段未完成的缓存和当前数据块
    const mergedContent = acc.remaining + chunk;
    const splitParts = mergedContent.split('\n');
    return {
      // 除最后一段外都是完整行
      lines: splitParts.slice(0, -1),
      // 最后一段没有遇到换行，存入缓存等下一个数据块拼接
      remaining: splitParts.at(-1)
    };
  }, { lines: [], remaining: '' }),
  // 将累计得到的完整行逐个发射
  mergeMap(segment => from(segment.lines))
);

订阅后输出顺序和预期完全一致：

hello
world  at
home
 and more

实现逻辑说明 #

• 无外部副作用：没有手动创建的外部可观察对象、没有主动触发的订阅逻辑，所有操作都在pipe内部完成，只有订阅时才会执行逻辑
• 兼容所有拆分场景：通过scan维护跨数据块的行缓存，既支持单个数据块包含多个换行，也支持换行符被拆分到两个不同数据块的情况
• 时序稳定：不需要依赖buffer、外部通知类Subject等容易出现竞态的操作符，逻辑完全随数据流推进执行
• 自动处理尾行：通过endWith注入结束标记，保证最后一段没有尾随换行符的内容也能正常发射

内容的提问来源于stack exchange，提问作者SystematicFrank