核心概念

理解 rx4u 的基础概念，有助于你构建高效、可维护的响应式应用。

函数式响应式编程

rx4u 通过三个核心原则实现函数式响应式编程（FRP）：

1. 纯函数

操作符是函数，而不是流对象上的方法。有状态操作符会在每个订阅中维护自己的状态：

TypeScript
import { map } from 'rx4u';

// Pure function - same input always produces same output
const double = map(x => x * 2);

// Can be reused safely
const stream1$ = pipeFrom(source1$, double);
const stream2$ = pipeFrom(source2$, double);

2. 不可变性

流转换会创建新流实例，而不是修改已有实例：

JavaScript
const originalStream$ = of(1, 2, 3);
const mappedStream$ = pipeFrom(originalStream$, map(x => x * 2));

// originalStream$ is unchanged
// mappedStream$ is a new, independent stream

3. 函数组合

复杂行为源于对简单函数的组合：

JavaScript
const processNumbers = (source$) => pipeFrom(
  source$,
  filter(x => x > 0),
  map(x => x * 2),
  distinctUntilChanged(),
  debounceTime(100)
);

流的表示（Sub）

在 rx4u 中，流由 Sub<T> 类型表示，它是一个接受回调函数的函数：

TypeScript
type Sub<T> = (
  next?: (value: T) => void,
  error?: (error: unknown) => void,
  complete?: () => void
) => () => void; // Returns unsubscribe function

使用示例

TypeScript
const stream$: Sub<number> = of(1, 2, 3);

const unsubscribe = stream$(
  value => console.log('Received:', value),
  error => console.error('Error:', error),
  () => console.log('Stream completed')
);

// Later...
unsubscribe();

惰性执行

rx4u 流采用惰性执行，只有订阅后才会计算：

JavaScript
const expensiveOperation$ = pipeFrom(
  of(1, 2, 3, 4, 5),
  map(x => {
    console.log('Processing:', x); // Only runs when subscribed
    return heavyComputation(x);
  })
);

// No computation has happened yet

expensiveOperation$(console.log); // Now the work begins

订阅模型

独立订阅

默认情况下，每次订阅都会创建独立的执行：

JavaScript
const stream$ = pipeFrom(
  interval(1000),
  map(() => Math.random())
);

stream$(x => console.log('Sub 1:', x)); // Gets random values
stream$(x => console.log('Sub 2:', x)); // Gets different random values

共享订阅

使用 share 操作符在订阅者之间共享执行：

JavaScript
const sharedStream$ = pipeFrom(
  interval(1000),
  map(() => Math.random()),
  share()
);

sharedStream$(x => console.log('Sub 1:', x)); // Gets same values
sharedStream$(x => console.log('Sub 2:', x)); // Gets same values

使用 pipeFrom 组合操作符

pipeFrom 函数实现优雅的操作符组合：

TypeScript
import { pipeFrom, map, filter, take, debounceTime } from 'rx4u';

const processedStream$ = pipeFrom(
  sourceStream$,
  filter(x => x.length > 0),
  map(x => x.trim().toLowerCase()),
  debounceTime(300),
  distinctUntilChanged(),
  take(10)
);

类型安全

操作符具有完整类型信息，确保整条管道的类型安全：

TypeScript
const numbers$ = of(1, 2, 3, 4, 5); // Sub<number>

const result$ = pipeFrom(
  numbers$,
  filter(x => x > 2),        // Sub<number>
  map(x => x.toString()),    // Sub<string>
  map(x => x.length)         // Sub<number>
);

错误处理

rx4u 提供多种错误处理策略：

catchError

处理错误并使用回退流继续执行：

JavaScript
const resilientStream$ = pipeFrom(
  riskyStream$,
  catchError(error => of('fallback value'))
);

retry

自动重试失败的操作：

JavaScript
const retriedStream$ = pipeFrom(
  unreliableStream$,
  retry(3) // Retry up to 3 times
);

内存管理

rx4u 专为高效使用内存而设计：

  • 操作符会在完成或取消订阅后释放订阅资源
  • 取消订阅会停止后续发出；已经运行的 Promise 本身无法取消
  • 共享流和状态流提供显式的清理行为
JavaScript
// Automatic cleanup
const unsubscribe = stream$(handleValue);
unsubscribe(); // Stops future emissions and releases subscription resources

下一步