Is there a straight forward way of using java.io.ObjectInputStream with java.util.stream.Stream?
It seems that java.io.BufferedReader.lines is provided to convert a buffered reader into a Stream but there doesn't seem to be an equivalent for Object's.
Has anyone have good solution for this conversion?
ObjectInputStream can read a lot more than Objects. The class implements DataInput, so it can read all kinds of data. Then there's special methods like readUnshared. So a Stream would be a pretty limited subset of ObjectInputStream's functonality.
But if you just want to read Objects one by one, you can write your own method:
public Stream<Object> toStream(final ObjectInputStream stream) {
return Stream.generate(() -> readObject(stream)).onClose(
() -> close(stream));
}
private static Object readObject(ObjectInputStream stream) {
try {
return stream.readObject();
} catch (IOException e) {
throw new UncheckedIOException(e);
} catch (ClassNotFoundException e) {
throw new RuntimeException(e);
}
}
private static void close(Closeable c) {
try {
c.close();
} catch (IOException e) {
logger.log(Level.WARNING, "Couldn't close " + c, e);
}
}
You could even have a typed stream:
public <T> Stream<T> toStream(final ObjectInputStream stream,
final Class<T> cls) {
return Stream.generate(() -> cls.cast(readObject(stream))).onClose(
() -> close(stream));
}
Object's captured from a socket through a chain of functions to modify and sort them. Passing them directly into a Stream seemed like it could be a nice way rather than putting them into something like a LinkedBlockingDeque<Object>
map(cls::cast) on it if you are confident that the type cast will always succeed. The tricky thing about the stream is that it will always terminate with an exception if you don’t limit it as Stream.generate creates an infinite stream which will always try to read another object from the input stream while the ObjectInputStream is finite in most cases (but doesn’t tell you how many objects you can read)…
this solution produces a finite stream without throwing EOFException at the end. It also closes the stream gracefully at the end. I had to implement a Spliterator that tries a read on hasNext() and returns the element read on next().
public static <T> Stream<T> toStream(final ObjectInputStream stream,
final Class<T> cls) {
Spliterator<T> spliterator = Spliterators.spliteratorUnknownSize(new Iterator<T>() {
private T next;
private boolean read = false;
@Override
public boolean hasNext() {
if (!read) {
try {
next = cls.cast(stream.readUnshared());
} catch (EOFException e) {
next = null;
} catch (ClassNotFoundException e) {
throw new RuntimeException(e);
} catch (IOException e) {
throw new UncheckedIOException(e);
}
read = true;
}
return next != null;
}
@Override
public T next() {
read = false;
return next;
}
}, Spliterator.DISTINCT | Spliterator.IMMUTABLE |
Spliterator.ORDERED | Spliterator.NONNULL);
return StreamSupport.stream(spliterator, false).onClose(() -> {
try {
stream.close();
} catch (IOException e) {
e.printStackTrace();
}
});
}
