Flutter RxDart BehaviorSubject经transform转换后value取值异常问题

你遇到的问题核心是对Stream.transform的作用理解有误：

• _roomAmount 是原始的BehaviorSubject实例，它内部存储的value永远是你直接调用_roomAmount.add()传入的原始输入值，不会被后续的流转换操作修改。
• 调用.stream.transform(validateRoomsAmount)会生成一个完全独立的新Stream，你在转换器里往sink添加的null、错误、校验后的值，只会发送给监听roomAmount$的订阅者，根本不会反向写入到原始的_roomAmount实例中。

所以你在getHouse()里直接读取_roomAmount.value，拿到的必然是未经过任何校验的原始输入，出现长度超过9的非法值是预期内的结果。
另外你的转换器代码里还有一处笔误：最后写入sink的size.toString()里的size变量从未定义，运行到合法值分支时会直接抛异常，应该替换为传入的参数value。

根据你的使用场景选以下任意一种方式即可：

方案1：需要同步读取校验后的值 #

如果你需要像现在一样同步拿到最新的校验结果创建对象，就把校验后的流转成支持同步读值的广播流，不要直接读原始subject的value：

// 存储原始输入的Subject
final _roomAmount = BehaviorSubject<String>();
// 校验后的流，用shareValueSeeded设置初始null，支持通过.value同步取最新值
Stream<String> get roomAmount$ => _roomAmount.stream
    .transform(validateRoomsAmount)
    .shareValueSeeded(null);

// 修正笔误后的校验转换器
final validateRoomsAmount = StreamTransformer<String, String>.fromHandlers(
    handleData: (value, sink){
      // 输入变化先置空
      sink.add(null);
      if(value == null || value == '') {
        return;
      }
      // 长度非法抛错
      if(value.length > 9) {
        sink.addError("Too long");
        return;
      }
      // 合法值写入，修正原来未定义的size变量
      sink.add(value);
    }
);

House getHouse() {
  // 读取校验后流的最新值，而非原始Subject的value
  return House(roomAmount: roomAmount$.value);
}

shareValueSeeded(null)是rxdart提供的操作符，会把普通流转换成和BehaviorSubject行为一致的广播流，既会给新订阅者发送最新值，也支持通过.value属性同步读取当前最新的发射值，初始传null和你校验逻辑里默认先发射null的逻辑完全匹配。

方案2：异步读取最新合法值 #

如果你不需要同步拿值，也可以直接从校验后流里取最新一次发射的值：

Future<House> getHouse() async {
  // 取流的第一个最新值
  final validAmount = await roomAmount$.first;
  return House(roomAmount: validAmount);
}

注意这个方法是异步的，不能在同步构建场景下直接调用。

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