import com.ociweb.pronghorn.stage.scheduling.GraphManager; //导入依赖的package包/类
public void buildReplicators(GraphManager gm, ArrayList<ReactiveManagerPipeConsumer> consumers) {
	int i = inputPipes.length;
	while (--i>=0) {
		if (1 == groupedPipes[i].length) {
			//swap back to using direct connection
			int c = consumers.size();				
			while (--c>=0) {
				if (consumers.get(c).swapIfFound(groupedPipes[i][0], inputPipes[i])) {
					break;
				}
			}
			if (c<0) {
				//can not optimize this case so just add the extra hop.
				logger.info("internal error unable to find this pipe!  Hello Guys!");
				ReplicatorStage.newInstance(gm, inputPipes[i], groupedPipes[i]);
			}
			
		} else {
			ReplicatorStage.newInstance(gm, inputPipes[i], groupedPipes[i]);
		}
	}
}
 

import com.ociweb.pronghorn.stage.scheduling.GraphManager; //导入依赖的package包/类
protected void logStageScheduleRates() {
    int totalStages = GraphManager.countStages(gm);
       for(int i=1;i<=totalStages;i++) {
           PronghornStage s = GraphManager.getStage(gm, i);
           if (null != s) {
               
               Object rate = GraphManager.getNota(gm, i, GraphManager.SCHEDULE_RATE, null);
               if (null == rate) {
                   logger.debug("{} is running without breaks",s);
               } else  {
                   logger.debug("{} is running at rate of {}",s,rate);
               }
           }
           
       }
}
 

import com.ociweb.pronghorn.stage.scheduling.GraphManager; //导入依赖的package包/类
public EgressMQTTStage(GraphManager graphManager, Pipe<MessageSubscription> input, Pipe<MQTTClientRequestSchema> output,
						CharSequence[] internalTopic,	CharSequence[] externalTopic, EgressConverter[] converter,
						int[] fieldQOS, int[] fieldRetain) {
	super(graphManager, input, output);
	this.input = input;
	this.output = output;
	this.internalTopic = internalTopic;
	this.externalTopic = externalTopic;		
	this.converter = converter;
	this.fieldQOS = fieldQOS;
	this.fieldRetain = fieldRetain;

	this.allTopicsMatch = isMatching(internalTopic,externalTopic,converter,fieldQOS,fieldRetain);
	
	supportsBatchedRelease = false; //must have immediate release
	supportsBatchedPublish = false; //also we want to minimize outgoing latency.
			
}
 

import com.ociweb.pronghorn.stage.scheduling.GraphManager; //导入依赖的package包/类
/**
 * Using real hardware support this stage turns on and off digital pins and sets PWM for analog out.
 * It supports time based blocks (in ms) specific to each connection.  This way no other commands are
 * send to that connection until the time expires.  This is across all pipes.
 * 
 */
public AbstractTrafficOrderedStage(GraphManager graphManager, 
		MsgRuntime<?,?> runtime,
		BuilderImpl hardware,
		Pipe<?>[] output,
		Pipe<TrafficReleaseSchema>[] goPipe,
		Pipe<TrafficAckSchema>[] ackPipe,
		
		Pipe<?> ... otherResponse ) {

	super(graphManager, join(goPipe, output), join(ackPipe, otherResponse));
       
	assert(output.length >= goPipe.length);
	
	this.runtime = runtime;
	this.hardware = hardware;
	this.etcAndDataPipe = output;//the last few pipes align with goPipe
	this.ackPipe = ackPipe;
	this.goPipe = goPipe;
	this.hitPoints = goPipe.length;
	this.graphManager = graphManager;
	

}
 

import com.ociweb.pronghorn.stage.scheduling.GraphManager; //导入依赖的package包/类
@Override 
public void startup() {
    	       
       //processing can be very time critical so this thread needs to be on of the highest in priority.
       Thread.currentThread().setPriority(Thread.MAX_PRIORITY);
    
	connectionBlocker = new Blocker(MAX_DEVICES);
	activeCounts = new int[goPipe.length];
	activeBlocks = new int[MAX_DEVICES];
	Arrays.fill(activeCounts, -1); //0 indicates, need to ack, -1 indicates done and ready for more
	Arrays.fill(activeBlocks, -1);

	startLoopAt = activeCounts.length;
	
	rate = (Number)graphManager.getNota(graphManager, this.stageId,  GraphManager.SCHEDULE_RATE, null);
}
 

import com.ociweb.pronghorn.stage.scheduling.GraphManager; //导入依赖的package包/类
/**
 * Parse HTTP data on feed and sends back an ack to the  SSLEngine as each message is decrypted.
 * 
 * @param graphManager
 * @param hardware
 * @param input
 * @param goPipe
 * @param ackPipe
 * @param output
 */

public HTTPClientRequestTrafficStage(
		GraphManager graphManager, 
		MsgRuntime<?,?> runtime,
		BuilderImpl hardware,
		ClientCoordinator ccm,
		
           Pipe<ClientHTTPRequestSchema>[] input,
           Pipe<TrafficReleaseSchema>[] goPipe,
           Pipe<TrafficAckSchema>[] ackPipe,
           
           Pipe<NetPayloadSchema>[] output
           ) {
	
	super(graphManager, runtime, hardware, input, goPipe, ackPipe, output);
	this.input = input;
	this.output = output;
	this.ccm = ccm;
	
	assert(ccm.isTLS == hardware.getHTTPClientConfig().isTLS());
	
	GraphManager.addNota(graphManager, GraphManager.DOT_BACKGROUND, "lavenderblush", this);
	GraphManager.addNota(graphManager, GraphManager.LOAD_MERGE, GraphManager.LOAD_MERGE, this);
	
}
 

import com.ociweb.pronghorn.stage.scheduling.GraphManager; //导入依赖的package包/类
private FilterStage(GraphManager graphManager, Pipe<ValveSchema> input, Pipe<ValveSchema> output) {
		super(graphManager, input, output);
		this.input=input;
		this.output=output;
		
//		int type = FieldReferenceOffsetManager.extractTypeFromLoc(MSG_LIFECYCLECOUNT_312_FIELD_STATION_1);
//		TypeMask.toString(type);
//		TypeMask.isInt(type);
//		switch(type) {
//		case TypeMask.IntegerUnsigned:
//			break;
//		case TypeMask.TextUTF8:
//			break;
//		}
		
	}
 

import com.ociweb.pronghorn.stage.scheduling.GraphManager; //导入依赖的package包/类
private ConsoleJSONDumpStage buildGraph2(String blockFilePath, File tempFile, StringBuilder results,
		GraphManager gm2) {
	Pipe<RawDataSchema> encryptedDataPipe2 = RawDataSchema.instance.newPipe(10, 1000);
	Pipe<RawDataSchema> resultDataPipe2 = RawDataSchema.instance.newPipe(10, 1000);
			
	Pipe<BlockStorageReceiveSchema> doFinalInput3 = BlockStorageReceiveSchema.instance.newPipe(10, 1000);
	Pipe<BlockStorageXmitSchema> doFinalOutput3 = BlockStorageXmitSchema.instance.newPipe(10, 1000);
	
	BlockStorageStage.newInstance(gm2, blockFilePath, doFinalOutput3, doFinalInput3);
			
	results.setLength(0);
	results.append("single large message: ");
	
	FileBlobReadStage read= new FileBlobReadStage(gm2, encryptedDataPipe2, tempFile.getAbsolutePath());
	RawDataCryptAESCBCPKCS5Stage decrypt2 = new  RawDataCryptAESCBCPKCS5Stage(gm2, pass, false,
			                                                                  encryptedDataPipe2, resultDataPipe2,
			                                                                  doFinalInput3, doFinalOutput3
														);
	
	ConsoleJSONDumpStage lastStage2 = ConsoleJSONDumpStage.newInstance(gm2, resultDataPipe2, results, true);
	return lastStage2;
}
 

import com.ociweb.pronghorn.stage.scheduling.GraphManager; //导入依赖的package包/类
public SequentialFileReadWriteStage(GraphManager graphManager,
 						 Pipe<SequentialCtlSchema>[] control,
 						 Pipe<SequentialRespSchema>[] response,
 						 Pipe<RawDataSchema>[] input,
 		                 Pipe<RawDataSchema>[] output, 
 		                 String[] paths) {
	
    super(graphManager, join(control, input), join(response, output));
    
    this.paths = paths;
    this.output = output;
    this.input = input;
    this.control = control;
    this.response = response;
    
    assert(paths.length == output.length);
    assert(input.length == output.length);
    assert(input.length == control.length);
    assert(response.length == control.length);

}
 

import com.ociweb.pronghorn.stage.scheduling.GraphManager; //导入依赖的package包/类
private static Pipe<NetPayloadSchema>[] buildClientUnwrap(GraphManager gm, ClientCoordinator ccm, Pipe<NetPayloadSchema>[] requests,
		int responseUnwrapCount, Pipe<NetPayloadSchema>[] socketResponse, Pipe<NetPayloadSchema>[] clearResponse,
		Pipe<ReleaseSchema>[] acks) {
	Pipe<NetPayloadSchema>[] hanshakePipes = null;
	if (ccm.isTLS) {
		assert(socketResponse.length>=responseUnwrapCount) : "Can not split "+socketResponse.length+" repsonse pipes across "+responseUnwrapCount+" decrypt units";			
		
		int c = responseUnwrapCount;
		Pipe<NetPayloadSchema>[][] sr = Pipe.splitPipes(c, socketResponse);
		Pipe<NetPayloadSchema>[][] cr = Pipe.splitPipes(c, clearResponse);
		
		hanshakePipes = new Pipe[c];
		
		while (--c>=0) {
			hanshakePipes[c] = new Pipe<NetPayloadSchema>(requests[0].config(),false); 
			SSLEngineUnWrapStage unwrapStage = new SSLEngineUnWrapStage(gm, ccm, sr[c], cr[c], acks[c], hanshakePipes[c], false, 0);
			GraphManager.addNota(gm, GraphManager.DOT_RANK_NAME, "UnWrap", unwrapStage);
		}
		
	}
	return hanshakePipes;
}
 

import com.ociweb.pronghorn.stage.scheduling.GraphManager; //导入依赖的package包/类
public static Pipe<NetPayloadSchema>[] buildRemainderOFServerStages(final GraphManager graphManager,
		ServerCoordinator coordinator, ServerPipesConfig serverConfig,
		Pipe<NetPayloadSchema>[] handshakeIncomingGroup) {
	
//	logger.info("build remainder of server");
	PipeConfig<NetPayloadSchema> fromOrderedConfig = serverConfig.orderWrapConfig();
	Pipe<NetPayloadSchema>[] fromOrderedContent = new Pipe[serverConfig.serverResponseWrapUnitsAndOutputs * serverConfig.serverPipesPerOutputEngine];

	Pipe<NetPayloadSchema>[] toWiterPipes = buildSSLWrapersAsNeeded(graphManager, coordinator, serverConfig, 
			                                                       handshakeIncomingGroup,
			                                                       fromOrderedContent, fromOrderedConfig);
                   
       buildSocketWriters(graphManager, coordinator, serverConfig.serverSocketWriters, toWiterPipes, 
       		           serverConfig.writeBufferMultiplier);

       ServerNewConnectionStage newConStage = new ServerNewConnectionStage(graphManager, coordinator); 
       coordinator.processNota(graphManager, newConStage);

	return fromOrderedContent;
}
 

import com.ociweb.pronghorn.stage.scheduling.GraphManager; //导入依赖的package包/类
public static void buildOrderingSupers(GraphManager graphManager, 
		                               ServerCoordinator coordinator, final int routerCount,
		                               Pipe<ServerResponseSchema>[][] fromModule, 
		                               Pipe<NetPayloadSchema>[] fromSupers) {
	///////////////////
	//we always have a super to ensure order regardless of TLS
	//a single supervisor will group all the modules responses together.
	///////////////////
	//logger.info("build ordering supervisors");
	assert(fromSupers.length >= routerCount) : "reduce router count since we only have "+fromSupers.length+" pipes";
	assert(routerCount>0);
	
	Pipe<NetPayloadSchema>[][] orderedOutput = Pipe.splitPipes(routerCount, fromSupers);
	int k = routerCount;
	while (--k>=0) {
					
		OrderSupervisorStage wrapSuper = new OrderSupervisorStage(graphManager, 
				                    fromModule[k], orderedOutput[k], coordinator);//ensure order   

		coordinator.processNota(graphManager, wrapSuper);
	
	}
}
 

import com.ociweb.pronghorn.stage.scheduling.GraphManager; //导入依赖的package包/类
private static void buildSocketWriters(GraphManager graphManager, ServerCoordinator coordinator, 
										int socketWriters, Pipe<NetPayloadSchema>[] toWiterPipes, 
										int writeBufferMultiplier) {
	///////////////
	//all the writer stages
	///////////////
	
	
	Pipe[][] req = Pipe.splitPipes(socketWriters, toWiterPipes);	
	int w = socketWriters;
	while (--w>=0) {
		
		ServerSocketWriterStage writerStage = new ServerSocketWriterStage(graphManager, coordinator, writeBufferMultiplier, req[w]); //pump bytes out
	    GraphManager.addNota(graphManager, GraphManager.DOT_RANK_NAME, "SocketWriter", writerStage);
	   	coordinator.processNota(graphManager, writerStage);
	}
}
 

import com.ociweb.pronghorn.stage.scheduling.GraphManager; //导入依赖的package包/类
public FieldSplitterStage(GraphManager graphManager, Pipe inputRing, Pipe outputRing) {
	super(graphManager,inputRing,outputRing);
	this.inputRing = inputRing;
	this.outputRing = outputRing;
	
	if (Pipe.from(inputRing) != RawDataSchema.FROM) {
		throw new UnsupportedOperationException("This class can only be used with the very simple RAW_BYTES catalog of messages for input.");
	}
	
	if (Pipe.from(outputRing) != MetaMessageDefs.FROM) {
		throw new UnsupportedOperationException("This class can only be used with the MetaFieldFROM catalog of messages for output.");
	}
	
	typeExtractor = new TypeExtractor(true /* force ASCII */);	
	
}
 

import com.ociweb.pronghorn.stage.scheduling.GraphManager; //导入依赖的package包/类
public MsgCommandChannel(GraphManager gm, B builder,
					  int features,
					  int parallelInstanceId,
					  PipeConfigManager pcm
		             ) {

   this.initFeatures = features;//this is held so we can check at every method call that its configured right
   this.builder = builder;
   this.pcm = pcm;
   this.parallelInstanceId = parallelInstanceId;
}
 

import com.ociweb.pronghorn.stage.scheduling.GraphManager; //导入依赖的package包/类
protected void configureStageRate(Object listener, ReactiveListenerStage stage) {
    //if we have a time event turn it on.
    long rate = builder.getTriggerRate();
    if (rate>0 && listener instanceof TimeListener) {
        stage.setTimeEventSchedule(rate, builder.getTriggerStart());
        //Since we are using the time schedule we must set the stage to be faster
        long customRate =   (rate*nsPerMS)/NonThreadScheduler.granularityMultiplier;// in ns and guanularityXfaster than clock trigger
        long appliedRate = Math.min(customRate,builder.getDefaultSleepRateNS());
        GraphManager.addNota(gm, GraphManager.SCHEDULE_RATE, appliedRate, stage);
    }
}
 

import com.ociweb.pronghorn.stage.scheduling.GraphManager; //导入依赖的package包/类
public BuilderImpl(GraphManager gm, String[] args) {
	
	this.operators = ReactiveListenerStage.reactiveOperators();
	
	this.gm = gm;
	this.getTempPipeOfStartupSubscriptions().initBuffers();
	this.args = args;
	
	this.pcm.addConfig(new PipeConfig<HTTPRequestSchema>(HTTPRequestSchema.instance, 
								                   MINIMUM_INCOMMING_REST_REQUESTS_IN_FLIGHT, 
								                   MAXIMUM_INCOMMING_REST_SIZE));
			
	int requestQueue = 4;
	this.pcm.addConfig(new PipeConfig<NetPayloadSchema>(NetPayloadSchema.instance,
			                                    requestQueue,
			                                    MINIMUM_TLS_BLOB_SIZE)); 		
		
	int maxMessagesQueue = 8;
	int maxMessageSize = 256;
	this.pcm.addConfig(new PipeConfig<MessageSubscription>(MessageSubscription.instance,
			maxMessagesQueue,
			maxMessageSize)); 		


	this.pcm.addConfig(new PipeConfig<TrafficReleaseSchema>(TrafficReleaseSchema.instance, DEFAULT_LENGTH));
	this.pcm.addConfig(new PipeConfig<TrafficAckSchema>(TrafficAckSchema.instance, DEFAULT_LENGTH));

    int defaultCommandChannelLength = 16;
    int defaultCommandChannelHTTPMaxPayload = 1<<14; //must be at least 32K for TLS support	    
	this.pcm.addConfig(new PipeConfig<NetResponseSchema>(NetResponseSchema.instance, defaultCommandChannelLength, defaultCommandChannelHTTPMaxPayload));   

	//for MQTT ingress
	int maxMQTTMessagesQueue = 8;
	int maxMQTTMessageSize = 1024;
	this.pcm.addConfig(new PipeConfig(IngressMessages.instance, maxMQTTMessagesQueue, maxMQTTMessageSize));
	
}
 

import com.ociweb.pronghorn.stage.scheduling.GraphManager; //导入依赖的package包/类
public <R extends ReactiveListenerStage> R createReactiveListener(GraphManager gm,  Behavior listener, 
		                		Pipe<?>[] inputPipes, Pipe<?>[] outputPipes, 
		                		ArrayList<ReactiveManagerPipeConsumer> consumers,
		                		int parallelInstance, String nameId) {
	assert(null!=listener);
	
	return (R) new ReactiveListenerStage(gm, listener, 
			                             inputPipes, outputPipes, 
			                             consumers, this, parallelInstance, nameId);
}
 

import com.ociweb.pronghorn.stage.scheduling.GraphManager; //导入依赖的package包/类
public IngressMQTTStage(GraphManager graphManager, Pipe<MQTTClientResponseSchema> input, Pipe<IngressMessages> output,
						CharSequence[] externalTopic, CharSequence[] internalTopic, IngressConverter[] converter, CharSequence connectionFeedbackTopic) {
	
	super(graphManager, input, output);
	this.input = input;
	this.output = output;
	this.externalTopic = externalTopic;
	this.internalTopic = internalTopic;
	this.connectionFeedbackTopic = connectionFeedbackTopic;
	this.allTopicsMatch = isMatching(internalTopic,externalTopic,converter);
	this.converter = converter;
	
}
 

import com.ociweb.pronghorn.stage.scheduling.GraphManager; //导入依赖的package包/类
public static TrafficCopStage newInstance(GraphManager graphManager, long msAckTimeout, 
				            Pipe<TrafficOrderSchema> primaryIn, 
				            Pipe<TrafficAckSchema>[] ackIn,  
				            Pipe<TrafficReleaseSchema>[] goOut, 
				            MsgRuntime<?,?> runtime,
				            BuilderImpl builder) {

	return new TrafficCopStage(graphManager, msAckTimeout, primaryIn, ackIn, goOut, runtime, builder);	
}