package com.ociweb.gl.impl.stage;

import com.ociweb.pronghorn.pipe.Pipe;
import com.ociweb.pronghorn.stage.PronghornStage;

public class ReactiveManagerPipeConsumer {

	public final Pipe[] inputs;
	private final ReactiveOperator[] operators;
	public final Object obj;
	
	public ReactiveManagerPipeConsumer(Object obj, ReactiveOperators operators, Pipe[] inputs) {
		
		this.obj = obj;
		this.inputs = inputs;
		assert(PronghornStage.noNulls(inputs));
		this.operators = new ReactiveOperator[inputs.length];
		
		int i = inputs.length;
		while (--i>=0) {
			this.operators[i] = operators.getOperator(inputs[i]);
		}
	}
	
	public final void process(ReactiveListenerStage r) {
		applyReactiveOperators(r, inputs, obj, operators, inputs.length);
	}

	private static void applyReactiveOperators(ReactiveListenerStage r, Pipe[] localInputs, Object localObj,
			ReactiveOperator[] localOperators, int count) {
		
		int hasMoreWork;
		do {
			hasMoreWork = 0;
			int i = count;
			while (--i>=0) {
				Pipe localPipe = localInputs[i];
				if (Pipe.contentRemaining(localPipe)>0) {
					localOperators[i].apply(i, localObj, localPipe, r);
					if (Pipe.contentRemaining(localPipe)>0) {
						hasMoreWork++;
					}
				}			
			}
		} while (--hasMoreWork>=0);
	}

	public boolean swapIfFound(Pipe oldPipe, Pipe newPipe) {		
		int i = inputs.length;
		while (--i>=0) {
			if (inputs[i] == oldPipe) {
				inputs[i] = newPipe;
				return true;
			}
		}
		return false;
	}
	
}

@@ -808,17 +808,20 @@
 Stage r,
- 
+%0A%09%09%09
 Pipe%5B%5D l
@@ -913,47 +913,48 @@
 int 
-hasMoreWork;%0A%09%09do %7B%0A%09%09%09hasMoreWork = 0;
+passes = 0;%0A%09%09int countDown = -2;%0A%09%09do %7B
 %0A%09%09%09
@@ -1181,27 +1181,22 @@
 %7B%0A%09%09%09%09%09%09
-hasMoreWork
+passes
 ++;%0A%09%09%09%09
@@ -1218,30 +1218,81 @@
 %7D%0A%09%09
-%7D while (--hasMoreWork
+%09if (-2==countDown) %7B%0A%09%09%09%09countDown = passes;%0A%09%09%09%7D%0A%09%09%7D while (--countDown
 %3E=0)

fix hang when out put of reactor is too short