掌握Dart Stream Transformers：打造高效能資料處理管道的實用指南

void main() {

    Stream<int> temperatures = Stream.fromIterable([15, 20, 25, 10, 30]);



    temperatures

        .transform(TemperatureFilter(20))

        .listen(

            (temp) => print('篩選後的溫度: $temp'),

            onError: (error) => print('發生錯誤: $error'),

        );

    // 輸出: 篩選後的溫度: 20, 25, 30

}

<dynamic> mixedData = Stream.fromIterable([15, "20", 25, null, 30]);

TypeSafeTemperatureFilter extends StreamTransformerBase<dynamic, int> {  final int threshold;    TypeSafeTemperatureFilter(this.threshold);    @override  Stream<int> bind(Stream<dynamic> stream) {    return stream.transform(      StreamTransformer<dynamic, int>.fromHandlers(        handleData: (data, sink) {          if (data == null) {            sink.addError('溫度讀取為空');            return;          }                    final int? temperature = int.tryParse(data.toString());          if (temperature == null) {            sink.addError('無效的溫度格式: $data');            return;          }                    if (temperature >= threshold) {            sink.add(temperature);          },      ),    );  }}

Stream<int> temperatures = Stream.fromIterable([15, 20, 25]);

StreamController<int> controller = StreamController<int>();

void safelyProcessTemperatures() async {

  if (controller.isClosed) {

    print('流已經關閉');

    return;

  }

  

  await for (final temp in temperatures) {

    if (controller.isClosed) break;

    controller.add(temp);

  }

}

stream.map(expensiveOperation).where(filterCondition);

stream.where(filterCondition).map(expensiveOperation);

stream    .bufferCount(10)  // 每次處理 10 個項目    

.map((batch) => batch.map(expensiveOperation).toList());

Stream<int> highFrequencyData = ...;

final throttledStream = highFrequencyData    

.transform(StreamTransformer.fromHandlers(      

handleData: (data, sink) {        

// 每 100 毫秒最多處理一次事件        

Future.delayed(Duration(milliseconds: 100), () => sink.add(data));      

},    

));