import com.ociweb.pronghorn.pipe.DataOutputBlobWriter; //导入依赖的package包/类
private boolean publishOnPrivateTopic(int token, Writable writable) {
	//this is a private topic            
	Pipe<MessagePrivate> output = publishPrivateTopics.getPipe(token);
	if (PipeWriter.tryWriteFragment(output, MessagePrivate.MSG_PUBLISH_1)) {
				
		DataOutputBlobWriter<MessagePrivate> writer = PipeWriter.outputStream(output);
		DataOutputBlobWriter.openField(writer);
		writable.write(writer);
		DataOutputBlobWriter.closeHighLevelField(writer, MessagePrivate.MSG_PUBLISH_1_FIELD_PAYLOAD_3);
		
		PipeWriter.publishWrites(output);

		return true;
	} else {
		return false;
	}
}
 

import com.ociweb.pronghorn.pipe.DataOutputBlobWriter; //导入依赖的package包/类
private int tokenForPrivateTopic(TopicWritable topic) {
	if (null==publishPrivateTopics) {
		return -1;
	}
	if (null==tempTopicPipe) {
		tempTopicPipe = RawDataSchema.instance.newPipe(2, maxDynamicTopicLength);
		tempTopicPipe.initBuffers();
	}

	int size = Pipe.addMsgIdx(tempTopicPipe, RawDataSchema.MSG_CHUNKEDSTREAM_1);
	DataOutputBlobWriter<RawDataSchema> output = Pipe.openOutputStream(tempTopicPipe);
	topic.write(output);
	DataOutputBlobWriter.closeLowLevelField(output);
	Pipe.confirmLowLevelWrite(tempTopicPipe, size);
	Pipe.publishWrites(tempTopicPipe);			
       
	Pipe.takeMsgIdx(tempTopicPipe);
	
	int token = publishPrivateTopics.getToken(tempTopicPipe);
			
	Pipe.confirmLowLevelRead(tempTopicPipe, size);
	Pipe.releaseReadLock(tempTopicPipe);
	return token;
}
 

import com.ociweb.pronghorn.pipe.DataOutputBlobWriter; //导入依赖的package包/类
private static void writeHeader(NetResponseWriter outputStream, final int headerBlobPosition, 
		                        final long positionOfLen,
								int statusCode, final int context, 
								HTTPContentType contentType,
								int length, boolean chunked) {
	//logger.info("writing head at position {} ", headerBlobPosition);
	DataOutputBlobWriter.openFieldAtPosition(outputStream, headerBlobPosition);

	byte[] revisionBytes = HTTPRevisionDefaults.HTTP_1_1.getBytes();		
	byte[] etagBytes = null;//TODO: nice feature to add later		
	int connectionIsClosed = 1&(context>>ServerCoordinator.CLOSE_CONNECTION_SHIFT);
	
	AbstractRestStage.writeHeader(revisionBytes, statusCode, 0, etagBytes, null!=contentType?contentType.getBytes():null, 
				                  length, chunked, false, null, 0,0,0, outputStream, connectionIsClosed);

	int propperLength = DataOutputBlobWriter.length(outputStream);
	Pipe.validateVarLength(outputStream.getPipe(), propperLength);
	Pipe.setIntValue(propperLength, outputStream.getPipe(), positionOfLen); //go back and set the right length.
	outputStream.getPipe().closeBlobFieldWrite();

}
 

import com.ociweb.pronghorn.pipe.DataOutputBlobWriter; //导入依赖的package包/类
private void writeChunkPrefix(NetResponseWriter outputStream, 
   		                      int headerBlobPosition, 
   		                      long positionOfLen,
   		                      int len) {
   	
   	//logger.trace("writing chunk at position {} ", headerBlobPosition);
	DataOutputBlobWriter.openFieldAtPosition(outputStream, headerBlobPosition);

	//since block has return on the end we must subtract 2 from what we wrote
       final int adj = 2+(5*(1&(context>>ServerCoordinator.END_RESPONSE_SHIFT))); 	
       
   	Appendables.appendHexDigitsRaw(outputStream, len-adj);
   	outputStream.write(RETURN_NEWLINE);
   	
	int propperLength = DataOutputBlobWriter.length(outputStream);
	Pipe.validateVarLength(outputStream.getPipe(), propperLength);
	Pipe.setIntValue(propperLength, outputStream.getPipe(), positionOfLen); //go back and set the right length.
	outputStream.getPipe().closeBlobFieldWrite();    	
}
 

import com.ociweb.pronghorn.pipe.DataOutputBlobWriter; //导入依赖的package包/类
private void publishConnect(Pipe<MQTTClientRequestSchema> output) {

		PipeWriter.presumeWriteFragment(output, MQTTClientRequestSchema.MSG_CONNECT_1);
			
	    PipeWriter.writeInt(output,MQTTClientRequestSchema.MSG_CONNECT_1_FIELD_KEEPALIVESEC_28, keepAliveSeconds);
	    PipeWriter.writeInt(output,MQTTClientRequestSchema.MSG_CONNECT_1_FIELD_FLAGS_29, flags);
	    PipeWriter.writeUTF8(output,MQTTClientRequestSchema.MSG_CONNECT_1_FIELD_CLIENTID_30, (CharSequence) clientId);
	    PipeWriter.writeUTF8(output,MQTTClientRequestSchema.MSG_CONNECT_1_FIELD_WILLTOPIC_31, (CharSequence) lastWillTopic);
	    
	    DataOutputBlobWriter<MQTTClientRequestSchema> writer = PipeWriter.outputStream(output);
	    DataOutputBlobWriter.openField(writer);	
	    if(null!= lastWillPayload) {
	    	lastWillPayload.write((MQTTWriter)writer);
	    }
	    DataOutputBlobWriter.closeHighLevelField(writer, MQTTClientRequestSchema.MSG_CONNECT_1_FIELD_WILLPAYLOAD_32);
	    
	    PipeWriter.writeUTF8(output,MQTTClientRequestSchema.MSG_CONNECT_1_FIELD_USER_33, (CharSequence) user);
	    PipeWriter.writeUTF8(output,MQTTClientRequestSchema.MSG_CONNECT_1_FIELD_PASS_34, (CharSequence) pass);
	    PipeWriter.publishWrites(output);
	}
 

import com.ociweb.pronghorn.pipe.DataOutputBlobWriter; //导入依赖的package包/类
private void convertMQTTTopicsToLocal(final byte[] backing1, final int pos1, final int len1, final int mask1) {
	int size = Pipe.addMsgIdx(tempSubject, RawDataSchema.MSG_CHUNKEDSTREAM_1);
	DataOutputBlobWriter<RawDataSchema> stream = Pipe.outputStream(tempSubject);
	DataOutputBlobWriter.openField(stream);
	TrieParserReader.parseSetup(topicConversionTrieReader, backing1, pos1, len1, mask1);
	boolean foundEnd = false;
	while (TrieParserReader.parseHasContent(topicConversionTrieReader)) {
					
		if (foundEnd) {
			throw new UnsupportedOperationException("Invalid topic if /# is used it must only be at the end.");
		}
		
		int token = (int)TrieParserReader.parseNext(topicConversionTrieReader, topicConversionTrie);
		if (-1 == token) {				
			stream.write(TrieParserReader.parseSkipOne(topicConversionTrieReader));
		} if (1 == token) {
			stream.write(WILD_POUND_THE_END); //  /#
			foundEnd = true;
		} if (2 == token) {
			stream.write(WILD_PLUS_THE_SEGMENT); //  +/
		}
	}
	DataOutputBlobWriter.closeLowLevelField(stream);
	Pipe.confirmLowLevelWrite(tempSubject, size);
	Pipe.publishWrites(tempSubject);
}
 

import com.ociweb.pronghorn.pipe.DataOutputBlobWriter; //导入依赖的package包/类
private void streamingRequest(Pipe<ClientHTTPRequestSchema> pipe, int httpRequestResponseId) {
		
	PipeWriter.presumeWriteFragment(pipe, ClientHTTPRequestSchema.MSG_HTTPGET_100);
	assert(httpRequestResponseId>=0);
	PipeWriter.writeInt(pipe, ClientHTTPRequestSchema.MSG_HTTPGET_100_FIELD_DESTINATION_11, httpRequestResponseId);
	PipeWriter.writeInt(pipe, ClientHTTPRequestSchema.MSG_HTTPGET_100_FIELD_SESSION_10, httpRequestResponseId);
	PipeWriter.writeInt(pipe, ClientHTTPRequestSchema.MSG_HTTPGET_100_FIELD_PORT_1, port);
	PipeWriter.writeUTF8(pipe, ClientHTTPRequestSchema.MSG_HTTPGET_100_FIELD_HOST_2, host);
	PipeWriter.writeUTF8(pipe, ClientHTTPRequestSchema.MSG_HTTPGET_100_FIELD_PATH_3, path);
			
	DataOutputBlobWriter<ClientHTTPRequestSchema> stream = PipeWriter.outputStream(pipe);
	DataOutputBlobWriter.openField(stream);
	myAuth.addHeaders(stream, "GET").append("\r\n");
	
	//myAuth.addHeaders(System.out, "GET");
	
	DataOutputBlobWriter.closeHighLevelField(stream, ClientHTTPRequestSchema.MSG_HTTPGET_100_FIELD_HEADERS_7);

	PipeWriter.publishWrites(pipe);
}
 

import com.ociweb.pronghorn.pipe.DataOutputBlobWriter; //导入依赖的package包/类
private boolean sendCommands(int [] cmd, int start, int length, int finalTargetIndex){
	DataOutputBlobWriter<I2CCommandSchema> i2cPayloadWriter = ch.i2cCommandOpen(i2c_address);
	length = length / 2; //we need to send two bytes for each command
	int i;
	
	for (i = start; i < Math.min(start + length, finalTargetIndex); i++){
		i2cPayloadWriter.write(COMMAND_MODE);
		i2cPayloadWriter.write(cmd[i]);
	}
	ch.i2cCommandClose(i2cPayloadWriter);
	ch.i2cFlushBatch();
	
	if (i == finalTargetIndex){
		return true;
	}
	return sendCommands(cmd, i, BATCH_SIZE, finalTargetIndex); //calls itself recursively until we reach finalTargetIndex
}
 

import com.ociweb.pronghorn.pipe.DataOutputBlobWriter; //导入依赖的package包/类
@Override
public void run() {
	
	while ((activeChannelId != -1) 
			&& (null!=activeOutput) 
			&& Pipe.hasRoomForWrite(activeOutput) ) {
		
		Pipe.addMsgIdx(activeOutput, ServerResponseSchema.MSG_TOCHANNEL_100);
		Pipe.addLongValue(activeChannelId, activeOutput);
		Pipe.addIntValue(activeSequenceNo, activeOutput);
			    
	    DataOutputBlobWriter<ServerResponseSchema> outputStream = Pipe.openOutputStream(activeOutput);

	    appendRemainingPayload(activeOutput);
		
	}		
	
	//only do when previous is complete.
	if (null == activeOutput) {
		int i = this.inputs.length;
		while ((--i >= 0) && (activeChannelId == -1)) {			
			process(inputs[i], outputs[i]);			
		}
	}
	
}
 

import com.ociweb.pronghorn.pipe.DataOutputBlobWriter; //导入依赖的package包/类
/**
 * The private method required. for {@link BinaryOLED#sendData(int[], int, int)} to function.
 * @param data
 * @param start
 * @param length
 * @param finalTargetIndex
 * @return true if the i2c bus is ready, false otherwise.
 */

private boolean sendData(int [] data, int start, int length, int finalTargetIndex){
	DataOutputBlobWriter<I2CCommandSchema> i2cPayloadWriter = ch.i2cCommandOpen(i2c_address);
	i2cPayloadWriter.write(DATA_MODE);
	int i;
	for (i = start; i < Math.min(start + length - 1, finalTargetIndex); i++){
		i2cPayloadWriter.write(data[i]);
	}
	ch.i2cCommandClose(i2cPayloadWriter);
	ch.i2cFlushBatch();
	if (i == finalTargetIndex){
		return true;
	}
	return sendData(data, i, BATCH_SIZE, finalTargetIndex); //calls itself recursively until we reach finalTargetIndex
}
 

import com.ociweb.pronghorn.pipe.DataOutputBlobWriter; //导入依赖的package包/类
private int endOfHeaderProcessing(int i, final int stateIdx, DataOutputBlobWriter<NetResponseSchema> writer) {
	int state;
	//all done with header move on to body
	writer.writeShort((short)-1); //END OF HEADER FIELDS 		
	
	//Now write header message, we know there is room because we checked before starting.
					    
	if (payloadLengthData[i] == -1) {
		positionMemoData[stateIdx]= state= 3;	
		payloadLengthData[i] = 0;//starting chunk size.			

	} else {
		positionMemoData[stateIdx]= state= 2;
	}
	return state;
}
 

import com.ociweb.pronghorn.pipe.DataOutputBlobWriter; //导入依赖的package包/类
private void requestPing(long now, long connectionId, Pipe<NetPayloadSchema> server) {
	
	Pipe.presumeRoomForWrite(server);
	int size = Pipe.addMsgIdx(server, NetPayloadSchema.MSG_PLAIN_210);
	Pipe.addLongValue(connectionId, server);
	Pipe.addLongValue(now, server);
	Pipe.addLongValue(0, server); //always use zero for client requests

	DataOutputBlobWriter<NetPayloadSchema> output = Pipe.openOutputStream(server);
	output.writeByte(0xC0);
	output.writeByte(0x00);	
	output.closeLowLevelField();
	
	Pipe.confirmLowLevelWrite(server, size);
	Pipe.publishWrites(server);
	//logger.info("wrote block of {}",len2);
}
 

import com.ociweb.pronghorn.pipe.DataOutputBlobWriter; //导入依赖的package包/类
private void buildPublishMessage(DataOutputBlobWriter<NetPayloadSchema> output, int qos, int packetId, int retain,
		int topicMeta, int topicLength, int payloadMeta, int payloadLength) {
	
	final int pubHead = 0x30 | (0x06&(qos<<1)) | 1&retain; //bit 3 dup is zero which is modified later
	output.writeByte((0xFF&pubHead));
	encodeVarLength(output, topicLength + 2 + payloadLength + (packetId>=0 ? 2 : 0)); //const and remaining length, 2  bytes

	//variable header
	output.writeShort(topicLength);
	Pipe.readBytes(input, output, topicMeta, topicLength);
	
	if (packetId>=0) {
		output.writeShort(packetId);
	}						
	
	Pipe.readBytes(input, output, payloadMeta, payloadLength);
}
 

import com.ociweb.pronghorn.pipe.DataOutputBlobWriter; //导入依赖的package包/类
private void storeReleaseOfId(long fieldBlockId) {

		 //keep in sync				     
		 recordReleaseId(fieldBlockId);
		 
		 //write this id in case of power drop
		 Pipe<RawDataSchema> out = fileOutput[fileOutput.length-1];
		 int chunkSize = Pipe.addMsgIdx(out, RawDataSchema.MSG_CHUNKEDSTREAM_1);
		 //logger.info("send out release data to be decrypted");
		 DataOutputBlobWriter<RawDataSchema> chunkStr = Pipe.outputStream(out);
		 DataOutputBlobWriter.openField(chunkStr);
		 chunkStr.writePackedLong(fieldBlockId);
		 DataOutputBlobWriter.closeLowLevelField(chunkStr);
		 
		 Pipe.confirmLowLevelWrite(out, chunkSize);
		 Pipe.publishWrites(out);
		 Pipe<SequentialCtlSchema> output = fileControl[fileControl.length-1];
		 
		 Pipe.presumeRoomForWrite(output);
		 FragmentWriter.writeL(output, SequentialCtlSchema.MSG_IDTOSAVE_4, fieldBlockId);
	
		 requestsInFlight++;
		  		 
	}
 

import com.ociweb.pronghorn.pipe.DataOutputBlobWriter; //导入依赖的package包/类
@Override
public void startup() {
    reader = new DataInputBlobReader<RawDataSchema>(input);
    output2Writer = new DataOutputBlobWriter<RawDataSchema>(output2);
    
    int maxValue = PhastCodecSchema.FROM.messageStarts.length+1;
    idxReverseLookup = new int[maxValue];
    int i = PhastCodecSchema.FROM.fieldIdScript.length;
    while (--i>=0) {
        long id = PhastCodecSchema.FROM.fieldIdScript[i];
        if (id>10000) {
            idxReverseLookup[(int)(id-10000)] =  i;
        }
    }          

}
 

import com.ociweb.pronghorn.pipe.DataOutputBlobWriter; //导入依赖的package包/类
private void writeBytesToOutput(Pipe<PhastCodecSchema> output1, Pipe<RawDataSchema> output2, DataOutputBlobWriter<RawDataSchema> output2Writer, DataInputBlobReader<RawDataSchema> reader) {
    Pipe.addMsgIdx(output1, PhastCodecSchema.MSG_BLOBCHUNK_1000);
    Pipe.confirmLowLevelWrite(output1, Pipe.sizeOf(output1, PhastCodecSchema.MSG_BLOBCHUNK_1000));
    
    Pipe.addMsgIdx(output2, RawDataSchema.MSG_CHUNKEDSTREAM_1);
    
    output2Writer.openField();                            
    DataOutputBlobWriter.writeBytes(output2Writer, reader, bytesRemainingToCopy);
    output2Writer.closeLowLevelField();
    
    bytesRemainingToCopy = 0;
    Pipe.confirmLowLevelWrite(output2, Pipe.sizeOf(output2, RawDataSchema.MSG_CHUNKEDSTREAM_1));
    
    Pipe.publishWrites(output2);
    Pipe.publishWrites(output1);
}
 

import com.ociweb.pronghorn.pipe.DataOutputBlobWriter; //导入依赖的package包/类
@Override
public void startup() {
    startup = System.currentTimeMillis();
    
    input2Reader = new DataInputBlobReader<RawDataSchema>(input2);
    writer = new DataOutputBlobWriter<RawDataSchema>(output);
    int i = PhastCodecSchema.FROM.fieldIdScript.length;
    lengthLookup = new short[i];
    sizeLookup = new int[i];
    while (--i>=0) {
        lengthLookup[i] = (short) (PhastCodecSchema.FROM.fieldIdScript[i]-10000);
        sizeLookup[i] = PhastCodecSchema.FROM.fragDataSize[i];
    }     
    
    int maxFieldsPerMessage = PhastCodecSchema.FROM.messageStarts.length;
    maxBytesPerMessage = maxFieldsPerMessage * 10;

    
}
 

import com.ociweb.pronghorn.pipe.DataOutputBlobWriter; //导入依赖的package包/类
private boolean sendCommandsInQuickSuccession(int start, int length){
	if (!ch.i2cIsReady()){
		return false;
	}
	assert(length < OLED_96x96_Consts.BATCH_SIZE);
	assert(start + length < OLED_96x96_Consts.BATCH_SIZE);
	DataOutputBlobWriter<I2CCommandSchema> i2cPayloadWriter = ch.i2cCommandOpen(i2c_address);
	i2cPayloadWriter.write(SSD1327_Consts.MCU);
	i2cPayloadWriter.write(SSD1327_Consts.UNLOCK_CMD_ENTERING);
	for (int i = start; i < start + length; i ++){

		i2cPayloadWriter.write(cmd_out[i]);
	}
	ch.i2cCommandClose(i2cPayloadWriter);
	ch.i2cFlushBatch();

	return true;
}
 

import com.ociweb.pronghorn.pipe.DataOutputBlobWriter; //导入依赖的package包/类
@Test
public void testEncodeString() throws IOException{
	//create a new blob pipe to put a string on 
	Pipe<RawDataSchema> pipe = new Pipe<RawDataSchema>(new PipeConfig<RawDataSchema>(RawDataSchema.instance, 100, 4000));
	pipe.initBuffers();
	DataOutputBlobWriter<RawDataSchema> writer = new DataOutputBlobWriter<RawDataSchema>(pipe);
	
	//encode a string on blob using the static method
	StringBuilder testString = new StringBuilder("This is a test");
	PhastEncoder.encodeString(writer, testString , 0, 0, false);
	
	writer.close();
	
	//check what is on the pipe
	DataInputBlobReader<RawDataSchema> reader = new DataInputBlobReader<RawDataSchema>(pipe);
	//should be -63
	int test = reader.readPackedInt();
	//the string
	String value = reader.readUTF();
	
	reader.close();
	
	String s = value.toString();
	assertTrue((test==-63) && (s.compareTo("This is a test")==0));
	
}
 

import com.ociweb.pronghorn.pipe.DataOutputBlobWriter; //导入依赖的package包/类
@Test
public void copyIntTest() throws IOException{
	//create blob for test
	Pipe<RawDataSchema> encodedValuesToValidate = new Pipe<RawDataSchema>(new PipeConfig<RawDataSchema>(RawDataSchema.instance, 100, 4000));
	encodedValuesToValidate.initBuffers();
	DataOutputBlobWriter<RawDataSchema> writer = new DataOutputBlobWriter<RawDataSchema>(encodedValuesToValidate);
	
	//create int dictionary
	int[] intDictionary = new int[5];
	Arrays.fill(intDictionary, 0);
	intDictionary[2] = 5;
	
	//make it increment 2 values 0 and 5
	PhastEncoder.copyInt(intDictionary, writer, 0, 0, 2, 0, false);
	writer.close();
}