package net.ihiroky.niotty;

/**
 * A pipeline for the {@link StoreStage}.
 *
 * @param <L> the type of the TaskLoop which executes the stages by default
 */
public class DefaultStorePipeline<L extends TaskLoop>
        extends AbstractPipeline<StoreStage<?, ?>, L> implements StorePipeline {

    private static final String SUFFIX_STORE = "[store]";

    public DefaultStorePipeline(
            String name, AbstractTransport<L> transport, TaskLoopGroup<? extends TaskLoop> taskLoopGroup) {
        super(String.valueOf(name).concat(SUFFIX_STORE), transport, taskLoopGroup);
    }

    @Override
    protected PipelineElement<Object, Object> createContext(
            StageKey key, StoreStage<?, ?> stage, PipelineElementExecutorPool pool) {
        @SuppressWarnings("unchecked")
        StoreStage<Object, Object> s = (StoreStage<Object, Object>) stage;
        return new StoreStageContext<>(this, key, s, pool);
    }

    @Override
    protected Tail<StoreStage<?, ?>> createTail(PipelineElementExecutorPool defaultPool) {
        return new StoreTail(this, IO_STAGE, defaultPool);
    }

    private static class StoreTail extends Tail<StoreStage<?, ?>> {

        private StoreStage<Object, Object> stage_;

        protected StoreTail(AbstractPipeline<?, ?> pipeline, StageKey key, PipelineElementExecutorPool pool) {
            super(pipeline, key, pool);
        }

        @Override
        @SuppressWarnings("unchecked")
        void setStage(StoreStage<?, ?> stage) {
            stage_ = (StoreStage<Object, Object>) stage;
        }

        @Override
        protected Object stage() {
            return stage_;
        }

        @Override
        protected void fire(Object input) {
            stage_.store(this, input);
        }

        @Override
        protected void fire(Object input, TransportParameter parameter) {
            WrappedStageContext<Object> context = new WrappedStageContext<>(this, parameter);
            stage_.store(context, input);
        }

        @Override
        protected void fire(TransportStateEvent event) {
            stage_.store(this, event);
        }

        @Override
        public TransportParameter transportParameter() {
            return DefaultTransportParameter.NO_PARAMETER;
        }
    }
}

@@ -1209,16 +1209,583 @@
 %3E stage_
+ = NULL;%0A%0A        private static final StoreStage%3CObject, Object%3E NULL = new StoreStage%3CObject, Object%3E() %7B%0A            @Override%0A            public void store(StageContext%3CObject%3E context, Object input) %7B%0A                throw new IllegalStateException(%22No I/O stage is initialized. It may not be connected?%22);%0A            %7D%0A%0A            @Override%0A            public void store(StageContext%3CObject%3E context, TransportStateEvent event) %7B%0A                throw new IllegalStateException(%22No I/O stage is initialized. It may not be connected?%22);%0A            %7D%0A        %7D
 ;%0A%0A     

Add message to show a state not connected