如何通过StreamTransformer将数据库列表流转换为领域模型流并创建StreamController

问题背景 #

我正尝试将ObjectBox数据库返回的数据流从数据库数据模型（PropertyObjectBox）转换为领域层数据模型（Property）。遇到以下困惑：

• 找到的StreamTransformer示例都是处理单个对象的流，但ObjectBox返回的是Stream<List<PropertyObjectBox>>类型的列表流
• 不清楚如何从Stream过渡到StreamController，现有代码存在报错，需要修改

现有代码（带疑问标注）：

/// <PropertyObjectBox> ObjectBox数据库中Property的数据模型
/// <PropertyModel> 数据层的数据模型（我知道其实没必要）
/// <Property> 领域层的数据模型


@override
// -->> 这里应该返回Stream<List<Property>?>还是Stream<Property>?
// -->> 或者为了兼容handleError需要返回其他类型？
Stream<List<Property>?> streamOnDeviceProperties() {
  Stream<List<PropertyObjectBox>> propObStream = objectbox.propertyBox.query()
      .watch(triggerImmediately: true).map((query) => query
  // 监听查询会生成Stream<Query<Property>>
  // 要获取List<Property>中的实际数据，需要调用query的find()方法
      .find());

//-->> 这里应该用PropertyObjectBox还是List<PropertyObjectBox>?
  var streamTransformer = StreamTransformer<PropertyObjectBox, dynamic>.fromHandlers(
    handleData: (PropertyObjectBox data, EventSink sink) {
      final propertyModel = PropertyModel.fromObjectbox(data);
      final Property property = propertyModel.toDomain();
      sink.add(property);
    },
    handleError: (Object error, StackTrace stacktrace, EventSink sink) {
      sink.addError('Something went wrong: $error');
    },
    handleDone: (EventSink sink) => sink.close(),
  );
//-->> 下一行报错：'Stream<List<PropertyObjectBox>>'类型没有定义getter 'stream'  
  var controllerStream = propObStream.stream.transform(streamTransformer);
}

修改后的完整代码 #

/// <PropertyObjectBox> ObjectBox数据库中Property的数据模型
/// <PropertyModel> 数据层的数据模型（我知道其实没必要）
/// <Property> 领域层的数据模型

@override
// 返回Stream<List<Property>>，若允许空结果可改用Stream<List<Property>?>
Stream<List<Property>> streamOnDeviceProperties() {
  // 获取ObjectBox的列表流
  Stream<List<PropertyObjectBox>> propObStream = objectbox.propertyBox.query()
      .watch(triggerImmediately: true)
      .map((query) => query.find());

  // 定义适配列表流的Transformer
  var streamTransformer = StreamTransformer<List<PropertyObjectBox>, List<Property>>.fromHandlers(
    handleData: (List<PropertyObjectBox> dataList, EventSink<List<Property>> sink) {
      // 遍历数据库模型列表，批量转换为领域模型
      final domainList = dataList.map((obModel) {
        final propertyModel = PropertyModel.fromObjectbox(obModel);
        return propertyModel.toDomain();
      }).toList();
      sink.add(domainList);
    },
    handleError: (Object error, StackTrace stacktrace, EventSink<List<Property>> sink) {
      sink.addError('转换数据时出错: $error', stacktrace);
    },
    handleDone: (EventSink<List<Property>> sink) => sink.close(),
  );

  // Stream本身支持transform方法，无需额外调用.stream
  return propObStream.transform(streamTransformer);
}

关键修改点说明 #

1. 返回类型确定

   • 因ObjectBox返回的是列表流，应返回Stream<List<Property>>（若数据库可能返回空结果，可改为Stream<List<Property>?>）
   • 无需返回Stream<Property>，这不符合监听查询全量结果的场景

2. StreamTransformer类型修正

   • 原Transformer针对单个对象，需改为适配列表流的StreamTransformer<List<PropertyObjectBox>, List<Property>>
   • 在handleData中批量处理整个列表，用map遍历转换每个模型

3. 修复报错点

   • propObStream本身就是Stream类型，直接调用.transform()即可，原代码的.stream是多余调用，导致报错

4. 错误处理优化

   • 传递StackTrace到错误回调，方便后续定位问题

5. 可选简化（移除中间层）
   若确实不需要PropertyModel中间层，可直接从数据库模型转领域模型：

   final domainList = dataList.map((obModel) => Property.fromObjectBox(obModel)).toList();
   

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