Каков наилучший способ создания java.util.stream.Stream из InputStream? - PullRequest
Новая
       23

Каков наилучший способ создания java.util.stream.Stream из InputStream?

2 голосов
/ 07 ноября 2019

Как я понимаю, InputStream - это поток байтов. Я заинтересован в преобразовании объекта InputStream в поток байтов. В основном, реализация следующего метода. 

public Stream<byte[]> toStream(final InputStream is, final int bufferSize);

Каков наилучший способ сделать это? Размер буфера - это количество байтов, считанных из InputStream за раз.

1 Ответ

2 голосов
/ 07 ноября 2019

Вы должны написать свой собственный Spliterator, что-то вроде этого: 

public final class ChunkingInputStreamSpliterator implements Spliterator<byte[]> {

    private final InputStream is;
    private final int bufferSize;

    public ChunkingInputStreamSpliterator(InputStream is, int bufferSize) {
        this.is = is;
        this.bufferSize = bufferSize;
    }

    @Override
    public boolean tryAdvance(Consumer<? super byte[]> action) {
        byte[] bytes;
        try {
            bytes = this.is.readNBytes(this.bufferSize);
        } catch (IOException e) {
            throw new UncheckedIOException(e);
        }
        if (bytes.length == 0)
            return false;
        action.accept(bytes);
        return true;
    }

    @Override
    public Spliterator<byte[]> trySplit() {
        return null; // cannot split an InputStream
    }

    @Override
    public long estimateSize() {
        return Long.MAX_VALUE; // unknown
    }

    @Override
    public int characteristics() {
        return Spliterator.ORDERED | Spliterator.NONNULL;
    }

}

Затем реализуйте свой метод так: 

public static Stream<byte[]> toStream(InputStream is, int bufferSize) {
    return StreamSupport.stream(new ChunkingInputStreamSpliterator(is, bufferSize), false);
}

Если выу вас нет Java 11, поэтому у вас нет очень удобного readNBytes метода, затем выполните эту часть самостоятельно так: 

public boolean tryAdvance(Consumer<? super byte[]> action) {
    byte[] bytes = new byte[this.bufferSize];
    int len = 0;
    try {
        for (int read; len < bytes.length; len += read)
            if ((read = this.is.read(bytes, len, bytes.length - len)) <= 0)
                break;
    } catch (IOException e) {
        throw new UncheckedIOException(e);
    }
    if (len == 0)
        return false;
    if (len < bytes.length)
        bytes = Arrays.copyOfRange(bytes, 0, len);
    action.accept(bytes);
    return true;
}
...