class AbstractStreamBuilder<T extends AbstractStreamBuilder<T, P>, P> {

    private static final Logger LOGGER = LoggerManager.getLogger(AbstractStreamBuilder.class);

    protected final PipelineImpl<?> pipeline;
    protected final StreamTerminator streamTerminator;
    protected final Set<BaseStream<?, ?>> streamSet; // Keeps track of the chain of streams so that we can auto-close them all
    private final List<Runnable> closeHandlers;  // The close handlers for this particular stream
    private boolean closed;
    private boolean linkedOrConsumed;

    AbstractStreamBuilder(PipelineImpl<?> pipeline, StreamTerminator streamTerminator, Set<BaseStream<?, ?>> streamSet) {
        this.pipeline = requireNonNull(pipeline);
        this.streamTerminator = requireNonNull(streamTerminator);
        this.closeHandlers = new ArrayList<>();
        this.streamSet = streamSet;
        this.linkedOrConsumed = false;
    }

    protected T append(Action<?, ?> newAction) {
        requireNonNull(newAction);
        pipeline.add(newAction);
        return self();
    }

    public T sequential() {
        pipeline.setParallel(false);
        return self();
    }

    public T parallel() {
        pipeline.setParallel(true);
        return self();
    }

    public boolean isParallel() {
        return pipeline.isParallel();
    }

    public T unordered() {
        pipeline.setOrdered(false);
        return self();
    }

    public T onClose(Runnable closeHandler) {
        requireNonNull(closeHandler);
        closeHandlers.add(closeHandler);
        return self();
    }

    public void close() {
        if (!closed) {
            closed = true;
            try {
                if (!closeHandlers.isEmpty()) {
                    ComposeRunnableUtil.composedRunnable(closeHandlers); // Run this stream's close handlers
                }
            } catch (Exception e) {
                throw new SpeedmentException(e);
            } finally {
                ComposeRunnableUtil.composedClose(streamSet.toArray(new AutoCloseable[0])); // Close the other streams
            }
        }
    }
    
    protected void assertNotLinkedOrConsumedAndSet() {
        if (linkedOrConsumed) {
            throw new IllegalStateException("stream has already been operated upon or has been consumed");
        }
        linkedOrConsumed = true;
    }

    protected P pipeline() {<FILL_FUNCTION_BODY>}

    private T self() {
        @SuppressWarnings("unchecked")
        final T thizz = (T) this;
        return thizz;
    }

    protected boolean finallyCloseBoolean(BooleanSupplier bs) {
        try {
            return bs.getAsBoolean();
        } catch (Exception e) {
            LOGGER.error(e);
            throw e;
        } finally {
            close();
        }
    }

    protected long finallyCloseLong(LongSupplier lp) {
        try {
            return lp.getAsLong();
        } catch (Exception e) {
            LOGGER.error(e);
            throw e;
        } finally {
            close();
        }
    }

    protected int finallyCloseInt(IntSupplier is) {
        try {
            return is.getAsInt();
        } catch (Exception e) {
            LOGGER.error(e);
            throw e;
        } finally {
            close();
        }
    }

    protected double finallyCloseDouble(DoubleSupplier ds) {
        try {
            return ds.getAsDouble();
        } catch (Exception e) {
            LOGGER.error(e);
            throw e;
        } finally {
            close();
        }
    }

    protected void finallyClose(Runnable r) {
        try {
            r.run();
        } catch (Exception e) {
            LOGGER.error(e);
            throw e;
        } finally {
            close();
        }
    }

    protected <T> T finallyCloseReference(Supplier<T> s) {
        try {
            return s.get();
        } catch (Exception e) {
            LOGGER.error(e);
            throw e;
        } finally {
            close();
        }
    }

    protected static Set<BaseStream<?, ?>> newStreamSet() {
        return new HashSet<>();
    }

}

        @SuppressWarnings("unchecked")
        final P result = (P) pipeline;
        return result;
    
<no_super_class>