响应式编程模块

脚本入口： eve.Rx()

参考 UniRx 提供基于推送的响应式流：Subject 家族、 LINQ 风格操作符、ReactiveProperty，以及把现有事件队列桥接为 Observable 流的能力。

基本用法

local rx = eve.Rx();
local clicks = rx.newSubject();
 
// 订阅（返回 Subscription，可取消）
local sub = clicks.subscribe(function(v) {
    print("click at " + v);
});
 
// 推送
clicks.onNext("(100, 200)");
clicks.onNext("(300, 50)");
 
// 不再需要时取消订阅
sub.dispose();

对象关系与调用时机

Rx 是工厂模块：newSubject() / newBehaviorSubject(v) / newReplaySubject(n) / newProperty(v) 创建流对象。流对象本身不绑定到帧循环——游戏代码负责在合适的时机调用 onNext / setValue，订阅回调在调用线程同步执行。fromEvent + pump 提供与 event 模块的桥接：pump(ev) 排空事件队列并把匹配名称的消息推给对应流。

目标导向指南

解耦生产者与消费者

生产者持有 Subject 并调用 onNext；消费者持有同一个对象并 subscribe。双方无需互相引用，取消订阅即解除关联。适合按键、点击、资源加载完成等事件广播。

用操作符链做数据变换

local numbers = rx.newSubject();
numbers
    .filter(function(v) { return v % 2 == 0; })
    .map(function(v) { return v * 10; })
    .subscribe(function(v) { print("result " + v); });
numbers.onNext(1);  // 不输出
numbers.onNext(2);  // result 20

观察可变状态

ReactiveProperty 既是读取入口也是可订阅流：状态变化自动通知订阅者，适合 UI 绑定、模型字段监听等场景。

local hp = rx.newProperty(100);
hp.subscribe(function(v) { print("hp -> " + v); });
hp.set(80);  // 打印 hp -> 80
print(hp.get());  // 80

桥接事件队列

local ev = eve.PlatformEvent();
local rx = eve.Rx();
local quest = rx.fromEvent("quest-complete");
quest.subscribe(function(data) { print("quest done: " + data); });
 
// 帧循环或任何时机排空事件队列，匹配的消息进入对应流
rx.pump(ev);

常见问题

忘记 dispose()：订阅者会一直留在流里接收通知。持订阅对象并在不再需要时 dispose。
在 worker 线程推送：回调在 onNext 的调用线程同步执行；脚本闭包只能在主线程被调用， worker 请通过 thread.postMain 回主线程后再推送。
把 ReactiveProperty 当普通变量：写 set() 会触发通知，若只是临时状态请用局部变量。
混淆 subscribe 与 subscribe3：前者只有 onNext 回调；后者是 subscribe3(onNext, onError, onCompleted)，全部必填（不需要就传 null）。

API 快查

下列方法名来自当前 Squirrel 绑定；同一模块创建的辅助对象的方法也列在这里。

模块 Rx：

newSubject() → Subject
newBehaviorSubject(v) → BehaviorSubject
newReplaySubject(capacity) → ReplaySubject
newProperty(v) → ReactiveProperty
fromEvent(name) → Observable
pump(event)：排空事件队列，把匹配 fromEvent 的消息推给对应流

**Observable / Subject / BehaviorSubject / ReplaySubject：**

subscribe(onNext) → Subscription
subscribe3(onNext, onError, onCompleted) → Subscription
map(fn) → Observable
filter(pred) → Observable
take(n) → Observable
skip(n) → Observable
first() → Observable
distinctUntilChanged() → Observable
Subject 额外：onNext(v) / onError(msg) / onCompleted() / hasObservers()
BehaviorSubject 额外：getValue() / setValue(v)
ReactiveProperty：get() / set(v) / subscribe / subscribe3

**Subscription：**

dispose() / isDisposed()

使用要点

流对象和订阅对象应保存在全局或实体状态中，不要在每帧重复创建。
默认回调在当前线程同步执行；跨线程需求先回到主线程再推送。
参数约束、默认值和返回类型以对应模块头文件及 addFunc 绑定为准；本文 API 快查与当前源码同步生成。

引擎内的 Editor 观察会话

Editor automation 直接复用本模块的 C++ Subject 和 distinctUntilChanged 管线，实现 MCP eve_editor_observe_start/poll/close。Editor 持有会话、流和订阅，是生命周期唯一所有者；MCP 的 poll 只在主线程采样 live runtime 并推送事件，不复制场景或材质权威状态。连续相同 JSON 事件不会重复交给 Agent，provider 丢失或 stale identity 则作为可观察的状态变化返回。使用方必须显式 close； target 或引擎对象仍按原所有权销毁，关闭观察会话不会销毁被观察对象。

源码： src/modules/rx/ 相关测试： test/rx.cpp