@@ -0,0 +1,1999 @@
+package io.datakernel.stream.processor;%0A%0Aimport io.datakernel.stream.*;%0A%0Aimport static io.datakernel.stream.DataStreams.stream;%0A%0A/**%0A * If stream consumer is not immediately wired, on next eventloop cycle it will error out.%0A * This is because consumers request producers to start producing items on the next cycle after they're wired.%0A%0A * This transformer solves that by storing a data receiver from consumer produce request if it is not wired%0A * and when it is actually wired request his new producer to produce into that stored receiver.%0A */%0Apublic class WaitingStreamConsumer%3CT%3E implements StreamTransformer%3CT, T%3E %7B%0A%0A%09private final AbstractStreamProducer%3CT%3E output = new AbstractStreamProducer%3CT%3E() %7B%0A%09%09@Override%0A%09%09protected void onProduce(StreamDataReceiver%3CT%3E dataReceiver) %7B%0A%09%09%09StreamProducer%3CT%3E producer = input.getProducer();%0A%09%09%09if (producer == null) %7B%0A%09%09%09%09waitingReceiver = dataReceiver;%0A%09%09%09%09return;%0A%09%09%09%7D%0A%09%09%09producer.produce(dataReceiver);%0A%09%09%7D%0A%0A%09%09@Override%0A%09%09protected void onSuspended() %7B%0A%09%09%09StreamProducer%3CT%3E producer = input.getProducer();%0A%09%09%09if (producer != null) %7B%0A%09%09%09%09producer.suspend();%0A%09%09%09%7D%0A%09%09%7D%0A%0A%09%09@Override%0A%09%09protected void onError(Throwable t) %7B%0A%09%09%09input.closeWithError(t);%0A%09%09%7D%0A%09%7D;%0A%0A%09private final AbstractStreamConsumer%3CT%3E input = new AbstractStreamConsumer%3CT%3E() %7B%0A%0A%09%09@Override%0A%09%09protected void onStarted() %7B%0A%09%09%09if (waitingReceiver != null) %7B%0A%09%09%09%09getProducer().produce(waitingReceiver);%0A%09%09%09%7D%0A%09%09%7D%0A%0A%09%09@Override%0A%09%09protected void onEndOfStream() %7B%0A%09%09%09output.sendEndOfStream();%0A%09%09%7D%0A%0A%09%09@Override%0A%09%09protected void onError(Throwable t) %7B%0A%09%09%09output.closeWithError(t);%0A%09%09%7D%0A%09%7D;%0A%0A%09private StreamDataReceiver%3CT%3E waitingReceiver;%0A%0A%09public static %3CT%3E StreamConsumer%3CT%3E wrapper(StreamConsumer%3CT%3E consumer) %7B%0A%09%09WaitingStreamConsumer%3CT%3E transformer = new WaitingStreamConsumer%3C%3E();%0A%09%09stream(transformer.output, consumer);%0A%09%09return transformer.input;%0A%09%7D%0A%0A%09@Override%0A%09public StreamConsumer%3CT%3E getInput() %7B%0A%09%09return input;%0A%09%7D%0A%0A%09@Override%0A%09public StreamProducer%3CT%3E getOutput() %7B%0A%09%09return output;%0A%09%7D%0A%7D%0A

Add WaitingStreamConsumer class