4

这个问题是前一个问题的延续。我编写了以下代码来确定是否File.openRead()创建了可以逐行流式传输的 Stream。事实证明，答案是否定的。读取整个文件，然后将其传递给下一个转换。那么我的问题是：如何在 Dart 中逐行流式传输文件？

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);
  }
}
4

4 回答 4

3

我认为这段代码很有用：

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

main() {
  final file = new File('file.txt');
  Stream<List<int>> inputStream = file.openRead();

  inputStream
    .transform(utf8.decoder)       // Decode bytes to UTF-8.
    .transform(new LineSplitter()) // Convert stream to individual lines.
    .listen((String line) {        // Process results.
        print('$line: ${line.length} bytes');
      },
      onDone: () { print('File is now closed.'); },
      onError: (e) { print(e.toString()); });
}
于 2020-04-23T21:02:13.883 回答
2

如果需要流，您可以从返回的未来创建它readAsLines()：

   Stream<List<String>> stream = 
      new Stream.fromFuture(new File('Data.txt').readAsLines());

然而，对我来说，一一简单地处理这些行看起来更简单，

  List<String> lines = new File('Data.txt').readAsLinesSync();
  for (var line in lines) {
    stdout.writeln(line); 
  }
于 2013-12-29T05:29:21.487 回答
2

转换器startChunkedConversion仅在转换开始时被调用一次。但是，返回的接收器的add方法会被文件的某些部分多次调用。

由源来决定块的大小，但是一个 37MB 的文件（如您之前的问题中提到的）肯定会以较小的块发送。

如果您想查看这些块，您可以拦截startChunkedConversion并返回一个包裹的接收器，或者您可以将自己置于openRead变压器和变压器之间。

截距：

class InterceptSink {
  static int lineSplitCount = 0;

  final _sink;
  InterceptSink(this._sink);
  add(x) {
    print("InterceptSink.add Count: $lineSplitCount");
    lineSplitCount++;
    _sink.add(x);
  }
  close() { _sink.close(); }
}

class LineSplitterIntercept extends Converter {
  convert(x) { throw "unimplemented"; }
  startChunkedConversion(outSink) {
    var lineSink = new LineSplitter().startChunkedConversion(outSink);
    return new InterceptSink(lineSink);
  }
}

之后openRead：

file.openRead()
  .transform(UTF8.decoder)
  .map(x) {
    print("chunk size: ${x.length)");
    return x;
  }
  .transform(new LineSplitter())
  ...
于 2013-12-29T12:34:31.077 回答
2

因为其他答案都不适合我的情况，所以这里有另一种技巧：

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


void main()
{
    var file = File('/path/to/some/file.txt');
    var raf = file.openSync(mode: fileMode.read);
 
    String line;
    while ((line = readLine(raf)) != null)
    {
        print(line);
    }
  }

  String readLine(RandomAccessFile raf, {String lineDelimiter = '\n'}) {
 
    var line = '';
    int byte;
    var priorChar = '';

    var foundDelimiter = false;

    while ((byte = raf.readByteSync()) != -1) {
      var char = utf8.decode([byte]);

      if (isLineDelimiter(priorChar, char, lineDelimiter)) {
        foundDelimiter = true;
        break;
      }

      line += char;
      priorChar = char;
    }
    if (line.isEmpty && foundDelimiter == false) {
      line = null;
    }
    return line;
  }


  bool isLineDelimiter(String priorChar, String char, String lineDelimiter) {
    if (lineDelimiter.length == 1) {
      return char == lineDelimiter;
    } else {
      return priorChar + char == lineDelimiter;
    }
  }
于 2020-01-02T05:29:47.737 回答