Java OutputStream, который постепенно обрабатывает текст

Question

Я хочу постепенно обработать текст, записанный в OutputStream, как он написан.

Например, предположим, что у нас есть эта программа:

import java.io.File;
import java.io.IOException;
import java.io.OutputStream;
import java.io.OutputStreamWriter;
import java.nio.charset.Charset;

public class Streaming {

    // Writes file, incrementally, to OutputStream.
    static void dump(File file, OutputStream out) throws IOException {
        // Implementation omitted
    }

    static int sum = 0;
    public static void main(String[] args) throws IOException {
        Charset charSet = Charset.defaultCharset(); // Interpret the file as having this encoding.
        dump(new File("file.txt"), new OutputStream() {
            @Override
            public void write(int b) throws IOException {
                // Add b to bytes already read,
                // Determine if we have reached the end of the token (using
                //   the default encoding),
                // And parse the token and add it to `sum`
            }
        });
        System.out.println("Sum: " + sum);
    }
}

Предположим, file.txt - это текстовый файл, содержащий разделенный пробелами список целых чисел. В этой программе я хочу найти сумму значений в file.txt, накапливая сумму в переменной sum. Я бы хотел избежать создания строки длиной в миллионы символов.

Меня интересует способ, которым я могу сделать это, используя функцию dump, которая записывает содержимое файла в выходной поток. Я не заинтересован в чтении файла другим способом (например, создание Scanner для file.txt и повторный вызов nextInt на сканере). Я налагаю это ограничение, потому что я использую библиотеку, API которой похож на dump, где клиент должен предоставить OutputStream, и библиотека впоследствии записывает много текста в выходной поток.

Как я могу реализовать метод write, чтобы правильно выполнить описанные шаги? Я хотел бы избежать выполнения токенизации вручную, поскольку такие утилиты, как Scanner, уже способны выполнять токенизацию, и я хочу иметь возможность обрабатывать любую кодировку текста (как указано charSet). Однако я не могу использовать Scanner напрямую, потому что нет способа проверить (неблокирующим способом), доступен ли токен:

    public static void main(String[] args) throws IOException {
        Charset charSet = Charset.defaultCharset();
        PipedInputStream in = new PipedInputStream();
        try (Scanner sc = new Scanner(in, charSet)) {
            dump(new File("file.txt"), new PipedOutputStream(in) {
                @Override
                public void write(byte[] b, int off, int len) throws IOException {
                    super.write(b, off, len);
                    // This will loop infinitely, because `hasNextInt`
                    // will block if there is no int token currently available.
                    if (sc.hasNextInt()) {
                        sum += sc.nextInt();
                    }
                }
            });
        }
        System.out.println("Sum: " + sum);
        System.out.println(charSet);
    }

Существует ли неблокирующая утилита, которая может выполнять для меня токенизацию при записи данных в выходной поток?

Answer 1

Если я правильно понимаю ваш вопрос, FilterOutputStream - это то, что вы хотите сделать подклассом. DigestOutputStream расширяет FilterOutputStream и делает что-то похожее на то, что вы хотите сделать: он отслеживает байты по мере их поступления и передает их в другой класс для обработки.

Одно решение, которое приходит на ум, - это FilterOutputStream для передачи байтов в PipedOutputStream , подключенный к PipedInputStream, который другой поток читает для создания вашей суммы:

PipedOutputStream sumSink = new PipedOutputStream();

Callable<Long> sumCalculator = new Callable<Long>() {
    @Override
    public Long call()
    throws IOException {

        long sum = 0;
        PipedInputStream source = new PipedInputStream(sumSink);

        try (Scanner scanner = new Scanner(source, charSet)) {
            while (scanner.hasNextInt()) {
                sum += scanner.nextInt();
            }
        }

        return sum;
    }
};
Future<Long> sumTask = ForkJoinPool.commonPool().submit(sumCalculator);

OutputStream dest = getTrueDestinationOutputStream();
dest = new FilterOutputStream(dest) {
    @Override
    public void write(int b)
    throws IOException {
        super.write(b);
        sumSink.write(b);
    }

    @Override
    public void write(byte[] b)
    throws IOException {
        super.write(b);
        sumSink.write(b);
    }

    @Override
    public void write(byte[] b,
                      int offset,
                      int len)
    throws IOException {
        super.write(b, offset, len);
        sumSink.write(b, offset, len);
    }

    @Override
    public void flush()
    throws IOException {
        super.flush();
        sumSink.flush();
    }

    @Override
    public void close()
    throws IOException {
        super.close();
        sumSink.close();
    }
};

dump(file, dest);

long sum = sumTask.get();

Answer 2

Основано на ответе @ tevemadar.Читает в строках и пытается разобрать их в целые.Если это не удается, то вы знаете, что число сделано и затем добавляется к сумме.Единственная проблема заключается в том, что мой метод не добавляет последнее число, если оно занимает последние два байта.Чтобы решить эту проблему, вы можете добавить однострочный метод: if(!currNumber.isEmpty()) sum += Integer.parseInt(currNumber);, который можно вызвать после завершения файла.

import java.io.FilterOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.util.Objects;

class SumOutputStream extends FilterOutputStream {
  public int sum = 0;
  String currNumber = "";
  String lastChar = "";

  public SumOutputStream(OutputStream os){
    super(os);
  }

  public void write(byte b[], int off, int len) throws IOException {
      Objects.checkFromIndexSize(off, len, b.length);
      for (int i = 0 ; i < len ; i++) {
          try { 
              if(!lastChar.isEmpty()) {
                  Integer.parseInt(lastChar);
                  currNumber += lastChar;
              }
          } catch(NumberFormatException e) { 
              if(!currNumber.isEmpty()) sum += Integer.parseInt(currNumber);
              currNumber = "";
          }  catch(NullPointerException e) {
              e.printStackTrace();
          }
          write(b[off + i]);
          lastChar = new String(b);
      }
  }
}

Answer 3

В качестве «идиоматического» подхода вам может потребоваться FilterOutputStream:

Эти потоки располагаются поверх уже существующего выходного потока (базового выходного потока), который он использует в качестве основного приемника данных, но, возможно, преобразует данные в процессе или обеспечивает дополнительную функциональность.

По крайней мере, для меня это звучит примерно так, как вы описываете.

Это конкретный класс (в отличие от OutputStream) , поэтому абсолютный минимум, с которым вы можете обойтись, - это предоставить ваш конструктор и реализацию для однобайтовой write() (которая собирается быть вызванным реализациями по умолчанию других write() методов):

public class SumOutputStream extends FilterOutputStream {
  public int sum = 0;
  public SumOutputStream(OutputStream os) {
    super(os);
  }

  private int num = 0;
  public void write(int b) throws IOException {
    if (b >= '0' && b <= '9') {
      sum -= num;
      num = num * 10 + b - '0';
      sum += num;
    } else {
      num = 0;
    }
    out.write(b);
  }

  public static void main(String[] args) throws IOException {
    try (SumOutputStream sos = new SumOutputStream(new FileOutputStream("test.txt"))) {
      sos.write("123 456 78".getBytes());
      System.out.println(sos.sum);
      sos.write('9');
      System.out.println(sos.sum);
    }
  }
}

Это будет суммировать все пропущенные числа, сохраняя sum в актуальном состоянии все время, даже с частичными результатами (это то, что должно показывать разделение 9).