Как сделать так, чтобы File.lines автоматически закрывался в Flowable.generate - PullRequest
Новая
       5

Как сделать так, чтобы File.lines автоматически закрывался в Flowable.generate

0 голосов
/ 02 ноября 2018

У меня есть следующий фрагмент, и он генерирует Flowable<String>. Я не уверен, как я могу сделать Files.lines Autoclosable. Мне нужно было передать iterator в качестве второго аргумента, чтобы прочитать строку одну за другой, как она потребляется.

Обратите внимание, что я не использовал третий аргумент (disposeState), как в generate(Callable<S> initialState, BiConsumer<S,Emitter<T>> generator, Consumer<? super S> disposeState), потому что нет смысла передавать iterator как disposeState. 

private Flowable<String> generateFlowable(File file) {
    return Flowable.generate(
            () -> Files.lines(Paths.get(file.toURI()), StandardCharsets.UTF_8).iterator(),
            (iterator, emitter) -> {
                if (iterator.hasNext()) {
                    emitter.onNext(iterator.next());
                } else {
                    emitter.onComplete();
                }
            }
    );
}

Строки расходуются и анализируются одна за другой в другом методе. Когда я запустил lsof, я обнаружил, что обработчик не был закрыт. Кто-нибудь может подсказать мне, как мы можем это сделать?

1 Ответ

0 голосов
/ 02 ноября 2018

Существует два возможных способа автоматического закрытия Flowable. Первый использует Flowable::using: 

private Flowable<String> generateFlowable(File file) {
  return Flowable.using(
          () -> Files.lines(file.toPath(), StandardCharsets.UTF_8),
          stream -> Flowable.fromIterable(stream::iterator),
          Stream::close
  );
}

Второй использует Flowable::generate, но использует BufferedReader: 

private Flowable<String> generateFlowable(File file) {
  return Flowable.generate(
          () -> Files.newBufferedReader(Paths.get(file.toURI()), StandardCharsets.UTF_8),
          (reader, emitter) -> {
            String line = reader.readLine();
            if (line != null) {
              emitter.onNext(line);
            } else {
              emitter.onComplete();
            }
          }, BufferedReader::close);
}
...