Java InfiniteStream with Queue

Question

I did InfiniteStream by implementing Stream and Consumer. You can initiate the stream then consume an object using it. I've added queueing for it to handle storing more than one value from accept. Please provide feedback.

public class InfiniteStream<T> implements Consumer<T>, Stream<T> {
    private static final int LENGTH = 1000;
    private final Stream<T> stream;
    private final Queueing q;
    private final int length;

    public InfiniteStream(int length) {
        this.length = length;
        this.q = new Queueing(this.length);
        this.stream = Stream.generate(q);
    }

    public InfiniteStream() {
        this(LENGTH);
    }

    @Override
    public void accept(T t) {
        q.accept(t);
    }

    @Override
    public Iterator<T> iterator() {
        return stream.iterator();
    }

    @Override
    public Spliterator<T> spliterator() {
        return stream.spliterator();
    }

    @Override
    public boolean isParallel() {
        return stream.isParallel();
    }

    @Override
    public Stream<T> sequential() {
        return stream.sequential();
    }

    @Override
    public Stream<T> parallel() {
        return stream.parallel();
    }

    @Override
    public Stream<T> unordered() {
        return stream.unordered();
    }

    @Override
    public Stream<T> onClose(Runnable closeHandler) {
        return stream.onClose(closeHandler);
    }

    @Override
    public void close() {
        stream.close();
    }

    @Override
    public Stream<T> filter(Predicate<? super T> predicate) {
        return stream.filter(predicate);
    }

    @Override
    public <R> Stream<R> map(Function<? super T, ? extends R> mapper) {
        return stream.map(mapper);
    }

    @Override
    public IntStream mapToInt(ToIntFunction<? super T> mapper) {
        return stream.mapToInt(mapper);
    }

    @Override
    public LongStream mapToLong(ToLongFunction<? super T> mapper) {
        return stream.mapToLong(mapper);
    }

    @Override
    public DoubleStream mapToDouble(ToDoubleFunction<? super T> mapper) {
        return stream.mapToDouble(mapper);
    }

    @Override
    public <R> Stream<R> flatMap(
            Function<? super T, ? extends Stream<? extends R>> mapper) {
        return stream.flatMap(mapper);
    }

    @Override
    public IntStream flatMapToInt(
            Function<? super T, ? extends IntStream> mapper) {
        return stream.flatMapToInt(mapper);
    }

    @Override
    public LongStream flatMapToLong(
            Function<? super T, ? extends LongStream> mapper) {
        return stream.flatMapToLong(mapper);
    }

    @Override
    public DoubleStream flatMapToDouble(
            Function<? super T, ? extends DoubleStream> mapper) {
        return stream.flatMapToDouble(mapper);
    }

    @Override
    public Stream<T> distinct() {
        return stream.distinct();
    }

    @Override
    public Stream<T> sorted() {
        return stream.sorted();
    }

    @Override
    public Stream<T> sorted(Comparator<? super T> comparator) {
        return stream.sorted(comparator);
    }

    @Override
    public Stream<T> peek(Consumer<? super T> action) {
        return stream.peek(action);
    }

    @Override
    public Stream<T> limit(long maxSize) {
        return stream.limit(maxSize);
    }

    @Override
    public Stream<T> skip(long n) {
        return stream.skip(n);
    }

    @Override
    public void forEach(Consumer<? super T> action) {
        stream.forEach(action);
    }

    @Override
    public void forEachOrdered(Consumer<? super T> action) {
        stream.forEachOrdered(action);
    }

    @Override
    public Object[] toArray() {
        return stream.toArray();
    }

    @Override
    public <A> A[] toArray(IntFunction<A[]> generator) {
        return stream.toArray(generator);
    }

    @Override
    public T reduce(T identity, BinaryOperator<T> accumulator) {
        return stream.reduce(identity, accumulator);
    }

    @Override
    public Optional<T> reduce(BinaryOperator<T> accumulator) {
        return stream.reduce(accumulator);
    }

    @Override
    public <U> U reduce(U identity, BiFunction<U, ? super T, U> accumulator,
            BinaryOperator<U> combiner) {
        return stream.reduce(identity, accumulator, combiner);
    }

    @Override
    public <R> R collect(Supplier<R> supplier,
            BiConsumer<R, ? super T> accumulator, BiConsumer<R, R> combiner) {
        return stream.collect(supplier, accumulator, combiner);
    }

    @Override
    public <R, A> R collect(Collector<? super T, A, R> collector) {
        return stream.collect(collector);
    }

    @Override
    public Optional<T> min(Comparator<? super T> comparator) {
        return stream.min(comparator);
    }

    @Override
    public Optional<T> max(Comparator<? super T> comparator) {
        return stream.max(comparator);
    }

    @Override
    public long count() {
        return stream.count();
    }

    @Override
    public boolean anyMatch(Predicate<? super T> predicate) {
        return stream.anyMatch(predicate);
    }

    @Override
    public boolean allMatch(Predicate<? super T> predicate) {
        return stream.allMatch(predicate);
    }

    @Override
    public boolean noneMatch(Predicate<? super T> predicate) {
        return stream.noneMatch(predicate);
    }

    @Override
    public Optional<T> findFirst() {
        return stream.findFirst();
    }

    @Override
    public Optional<T> findAny() {
        return stream.findAny();
    }

    final private class Queueing implements Consumer<T>, Supplier<T> {
        private final int length;
        private final BlockingQueue<T> q;

        private Queueing(int length) {
            this.length = length;
            q = new LinkedBlockingQueue<T>(this.length);
        }

        @Override
        public T get() {
            try {
                return q.take();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new RuntimeException(e);
            }
        }

        @Override
        public void accept(T t) {
            try {
                q.put(t);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new RuntimeException(e);
            }
        }

    }
}

Adding How to use it:

public static void main(String[] args) {
    InfiniteStream<Integer> is = new InfiniteStream<Integer>(10);
    CompletableFuture.runAsync(() -> {
        while (true) {
            try {
                Thread.sleep(1000);
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
            IntStream.range(0, 100).boxed().forEach(is);
        }
    });
    is.forEach(System.out::println);
}

Answer 1

What you have implemented here is a "wrapper" class for a stream. In almost all instances you just pass through the behaviour you want to the wrapped stream instance.

A better solution would be to use the existing Stream code supplied in StreamSupport. This allows you to use code that is well tested, and part of the library. When I played with your code, I ended up changing the API a little, and it helps a lot.

The best way I can think to review this, is to show what I have done, and why it is "better".

import java.util.ArrayList;
import java.util.List;
import java.util.Spliterator;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.function.Consumer;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;

/**
 * Implement a buffered consumer that feeds consumed items in to a stream
 * 
 * @author rolf
 *
 * @param <V>
 *            the generic type of the data being streamed.
 */
public class ConsumerToStream<V> implements Consumer<V> {

    private static final class QSpliterator<T> implements Spliterator<T> {

        private final BlockingQueue<T> queue;

        public QSpliterator(BlockingQueue<T> queue) {
            this.queue = queue;
        }

        @Override
        public boolean tryAdvance(Consumer<? super T> action) {
            try {
                action.accept(queue.take());
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new IllegalStateException("Take interrupted.", e);
            }
            return true;
        }

        @Override
        public Spliterator<T> trySplit() {
            try {
                final int size = queue.size();
                List<T> vals = new ArrayList<>(size + 1);
                vals.add(queue.take());
                queue.drainTo(vals);
                return vals.spliterator();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new IllegalStateException("Thread interrupted during trySplit.", e);
            }
        }


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

        @Override
        public int characteristics() {
            return Spliterator.CONCURRENT;
        }

    }

    private final Stream<V> outstream;
    private final BlockingQueue<V> blockingQueue;

    private final Spliterator<V> splitter;

    /**
     * Construct an instance of the consumer buffer with the supplied maximum
     * capacity
     * 
     * @param bufferSize
     *            the amount of space to set aside for buffered items.
     */
    public ConsumerToStream(int bufferSize) {
        this.blockingQueue = new LinkedBlockingQueue<>(bufferSize);
        this.splitter = new QSpliterator<>(blockingQueue);
        this.outstream = StreamSupport.stream(splitter, false);
    }

    /**
     * Get the stream this buffer outputs to.
     * 
     * @return the output stream.
     */
    public Stream<V> stream() {
        return outstream;
    }

    @Override
    public void accept(V t) {
        try {
            blockingQueue.put(t);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new IllegalStateException("Interrupted accepting a new value.", e);
        }
    }

}

Let's run through the parts there:

• there's the main class ConsumerToStream which is a Consumer, and has an accept method which it overrides from the interface.
• there's the QSpliterator class which is a private class - it is what is used to feed the output Stream - by being "wrapped" by StreamSupport.
• there's the queue, which is used to buffer and thread-manage the inputs to the stream.

Note that the QSpliterator supports trySplit() by feeding off a possibly sized-1 partition, leaving nothing in the main split.

Of significance in this implementation, is that all the methods are useful - there's no blind "wrapping" code... it's "clean".

The use case would look like:

public static void main(String[] args) {
    ConsumerToStream<Integer> cts = new ConsumerToStream<>(10);
    CompletableFuture.runAsync(() -> {
        while (true) {
            try {
                Thread.sleep(1000);
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
            IntStream.range(0, 100).boxed().forEach(cts);
        }
    });
    cts.stream().parallel().forEach(System.out::println);
}    

Oh, and by-the-way - your handling of interrupted exceptions was a pleasure to see. Well done.