How to obtain Inputstream from Java 8 Streams?

Question

I have some data streamed from different files. It is in the following format:

Stream<String> linesModifiedAndAppendedFromVariousFiles=getLines();

However, I need to feed this into an library method that accepts InputStream or Reader as a parameter.

How can I feed this Java 8 stream into an InputStream or a type of Reader?

P.S: this is not about wrapping java.util.streams.Stream around an InputStream. What I am looking for is the other way around.

Intimidate

A Stream has nothing to do with InputStream. Those two are entirely two different packages. – Ucayali

Possible duplicate of How can I use Java 8 Streams with an InputStream? – Ucayali

@MuratKaragoz I know! There is nothing in questions that claims it is. There should be (i hope) a way to create an Inputsream that accepts Java 8 streams as a source. – Intimidate

@MuratKaragöz No, that one is about wrapping java.util.stream.Stream around an InputStream. What I need is the other way around. – Intimidate

Answer 1

You can do this with PipedReader and PipedWriter.

PipedReader reader = new PipedReader();

Runnable feeder = new Runnable() {
    @Override
    public void run() {
        try (PipedWriter writer = new PipedWriter(reader)) {
            linesModifiedAndAppendedFromVariousFiles.forEachOrdered(line -> {
                try {
                    writer.write(line);
                    writer.write('\n');
                } catch (IOException e) {
                    throw new UncheckedIOException(e);
                }
            });
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
};
new Thread(feeder).start();

someLibraryMethod.consumeReader(reader);

Bonham

thanks for the reply. The thing is, here forEachOrdered is a terminal operation hence it terminates the stream. But i want it to be an infinite stream being constantly populated. How can I achieve same thing above given this info? – Intimidate

forEachOrdered will continue until the Stream is closed. forEachOrdered does not itself close or terminate a Stream. – Bonham

Answer 2

A java.util.Stream is conceptually a potentially endless, non-reversing (as in, once you move on past an entry you cant go back to it) sequence, which may potentially allow you to process it in parallel. Crucially, the 'thingies' in the sequence can be ANYTHING at all. For example, you can have a stream of Color objects.

A java.io.InputStream is conceptually a potentially endless, non-reversing, non-parallellizable sequence of bytes.

These 2 things are just not the same.

However, if you have a Stream of bytes specifically, you could feasibly turn that into an input stream. You simply elect not to use the parallellization option inherent in Stream, and then these 2 things start boiling down to the same thing. However, if you have a stream of anything that isn't a byte, you will have to come up with the 'mapping'.

Let's say you have a stream of string objects. Let's say this is a stream of the first 4 numbers in english (so: Arrays.asList("one", "two", "three", "four").stream()).

How do you want to map this stream-of-strings into a stream-of-bytes? One strategy could be to render the strings to bytes using UTF-8 encoding, and to use the 0 char as a separator. In other words, that you want the same result as this hypothetical: new ByteArrayInputStream(new String("one\0two\0three\0four").getBytes(StandardCharsets.UTF_8)).

One can imagine a function that takes in a Stream<Byte> and turns it into an InputStream. However, a Stream<Byte> would be a very inefficient concept. One can also imagine a function that takes a Stream<T> along with a mapping function mapping T to byte[], and a separator constant (or function that generates a value) that produces the separators. For the above example, something like:

toInputStream(oneTwoThreeFour, str -> str.getBytes(StandardCharsets.UTF_8), "\0");

as far as I know, this does not exist in the core libraries nor in places like guava. But writing it should be trivial enough. Maybe half a page's worth of code.

Brassware

Answer 3

Reading stream of bytes:

    PipedInputStream inputStream = new PipedInputStream();
    Thread infinitInputStreamSupplier = infinitInputStreamSupplierThread(inputStream);
    infinitInputStreamSupplier.start();
    //consume input stream here...
    infinitInputStreamSupplier.interrupt();

Here are the methods to generate input stream

private Thread infinitInputStreamSupplierThread(final PipedInputStream inputStream) {
        return new Thread(() -> {
            try (PipedOutputStream outputStream = new PipedOutputStream(inputStream)) {
                Stream<byte[]> infiniteStream = Stream.generate(randomByteSupplier());
                infiniteStream.forEachOrdered(bytes -> {
                    try {
                        outputStream.write(bytes);
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                });
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        });
    }

    private Supplier<byte[]> randomByteSupplier() {
        return () -> {
            byte[] bytes = new byte[100];
            new Random().nextBytes(bytes);

            return bytes;
        };
    }

Nunciature