Dart 中的 Stream<List<int>> 和 Stream<int> 有什么区别

Question

我正在尝试围绕 Dart Streams。特别是这个example of the command line utility cat 有以下代码行：

 Stream<List<int>> stream = new File(path).openRead();

  // Transform the stream using a `StreamTransformer`. The transformers
  // used here convert the data to UTF8 and split string values into
  // individual lines.
  return stream
      .transform(UTF8.decoder)
      .transform(const LineSplitter())
      .listen((line) {
        if (showLineNumbers) {
          stdout.write('${lineNumber++} ');
        }
        stdout.writeln(line);
      }).asFuture().catchError((_) => _handleError(path));

1. Stream<T> 声明为Stream<List<int>> 让我有点困惑。为什么不声明为Stream<int>。 List 类型如何使它与众不同。如果是 List，订阅者事件是否以某种方式缓冲？

2. 什么类型（如<T>）被传递给第一个转换？是int 还是List<int>？

3. 将什么类型传递给每个下一个转换以及决定它们的类型的因素。

4. 此示例是否在将转换结果传递给下一个转换之前读取整个文件？如果是这样，是否有一个示例说明如何流式传输类似于此节点问题Parsing huge logfiles in Node.js - read in line-by-line

的非常大的文件

Answer 1

1. 好问题。
2. UTF8 是一个扩展 Codec<String, List<int>> 的 Utf8Codec。所以UTF8.decoder 是一个Converter<List<int>, String>，它以List<int> 作为参数。
3. LineSplitter 是 Converter<String, List<String>>。所以它将String 作为参数。 .transform(const LineSplitter()) 的结果流是 Stream<String>，每行都在其中发送。
4. File.openRead 在将第一个字节写入流之前不会读取整个文件。所以处理大文件没有问题。

Answer 2

Alexandre Ardhuin 的前三个问题是正确的。然而，第四个问题不是。在把它拆开并删除我自己的代码版本后，我确定了以下内容：

即使是 37Mb 的文件，转换也只会被调用一次。

这是我用来弄清楚的代码。

import 'dart:async';
import 'dart:convert';
import 'dart:io';


void main(List<String> arguments) {

  Stream<List<int>> stream = new File('Data.txt').openRead();

   stream
      .transform(const Utf8InterceptDecoder())
        .transform(const LineSplitterIntercept())
          .listen((line) {
//            stdout.writeln(line);
          }).asFuture().catchError((_) => print(_));
}

int lineSplitCount = 0;

class LineSplitterIntercept extends LineSplitter {

  const LineSplitterIntercept() : super();
  // Never gets called
  List<String> convert(String data) {
    stdout.writeln("LineSplitterIntercept.convert : Data:" + data);
    return super.convert(data);
  }

  StringConversionSink startChunkedConversion(ChunkedConversionSink<String> sink) {
    stdout.writeln("LineSplitterIntercept.startChunkedConversion Count:"+lineSplitCount.toString()+ " Sink: " + sink.toString());
    lineSplitCount++;
    return super.startChunkedConversion(sink);
  }
}

int utfCount = 0;

class Utf8InterceptDecoder extends Utf8Decoder {

  const Utf8InterceptDecoder() : super();

  //never gets called
  String convert(List<int> codeUnits) {
    stdout.writeln("Utf8InterceptDecoder.convert : codeUnits.length:" + codeUnits.length.toString());
    return super.convert(codeUnits);
  }


  ByteConversionSink startChunkedConversion(ChunkedConversionSink<String> sink) {
    stdout.writeln("Utf8InterceptDecoder.startChunkedConversion Count:"+ utfCount.toString() + " Sink: "+ sink.toString());
    utfCount++;
    return super.startChunkedConversion(sink);
  }
}