条件操作符

iff

在订阅的那一刻决定走哪条路。

First Tab

import { iif, of, defer } from 'rxjs';

const isDev = process.env.NODE_ENV === 'development';

// 根据环境决定输出日志的方式
const logger$ = iif(
  () => isDev,
  of('Pipe to Console (Dev Mode)'),
  of('Pipe to Log File (Prod Mode)')
);

logger$.subscribe(console.log);

Second Tab

defaultEmpty

如果流里什么都没产生就结束了，给它一个“低保”。

Node.js 场景

配置文件读取

import { EMPTY, defaultIfEmpty } from "rxjs";

const dbResult$ = EMPTY.pipe(
    defaultIfEmpty({ host: 'localhost', port: 5432 })
)

dbResult$.subscribe(config => {
    console.log(`DB Config: ${JSON.stringify(config)}`)
});

DOM 场景

every

检查流中的所有数据是否都满足条件

Node.js 场景

批量权限检查

import { from, every } from "rxjs"

const userPermissions = ['read', 'write', 'execute'];

// 检查用户是否拥有所有必须的权限
from(userPermissions).pipe(
    every(permission => ['read', 'write', 'execute', 'delete'].includes(permission))
).subscribe(hasAllPermissions => console.log(`User has all permissions: ${hasAllPermissions}`));

DOM 场景

takeUntil

最常用的条件操作符之一，用于手动管理资源释放。

Node.js 场景

超时关闭服务

import { interval, takeUntil } from "rxjs"

// 模拟每秒一次的心跳，10s后停止
const hearthbeat$ = interval(1000);
const stopSignal$ = interval(10000);

hearthbeat$.pipe(
    takeUntil(stopSignal$)
).subscribe({
    next: (val) => console.log(`Heartbeat ${val}`),
    complete: () => console.log('Heartbeat stopped after 10 seconds')
});

DOM 场景

sequenceEqual

sequenceEqual 会订阅两个流，并将它们发出的每一个值进行一一对比。如果任何一个值不匹配，或者其中一个流提前结束，它会立即发出 false 并结束。

Node.js 场景

import { delay, from, sequenceEqual } from "rxjs"

const expectedSequence = from(["READ", "WRITE", "DELETE"]);

// 模拟用户的实时操作指令流（可能来自不同的异步功能）
const userActions$ = from(["READ", "WRITE"]).pipe(
    delay(1000) // 模拟网络延迟，即便时间不同，只要顺序和值对，结果就是 true
)

userActions$.pipe(
    sequenceEqual(expectedSequence)
).subscribe(isEqual => {
    console.log(`User actions match expected sequence: ${isEqual}`);
})

DOM 场景