<commit_before>package com.ociweb.gl.api;

import com.ociweb.pronghorn.network.config.HTTPContentType;
import com.ociweb.pronghorn.network.config.HTTPRevisionDefaults;
import com.ociweb.pronghorn.network.module.AbstractAppendablePayloadResponseStage;
import com.ociweb.pronghorn.network.schema.ServerResponseSchema;
import com.ociweb.pronghorn.pipe.DataOutputBlobWriter;
import com.ociweb.pronghorn.pipe.FieldReferenceOffsetManager;
import com.ociweb.pronghorn.pipe.Pipe;
import com.ociweb.pronghorn.pipe.PipeWriter;
import com.ociweb.pronghorn.util.Appendables;

public class HTTPResponseService {

	private final MsgCommandChannel<?> msgCommandChannel;
	
	public HTTPResponseService(MsgCommandChannel<?> msgCommandChannel) {
		this.msgCommandChannel = msgCommandChannel;
		msgCommandChannel.initFeatures |= MsgCommandChannel.NET_RESPONDER;
	}
	
	public HTTPResponseService(MsgCommandChannel<?> msgCommandChannel,
			int queueLength, int maxMessageSize) {
		this.msgCommandChannel = msgCommandChannel;
		MsgCommandChannel.growCommandCountRoom(msgCommandChannel, queueLength);
		msgCommandChannel.initFeatures |= MsgCommandChannel.NET_RESPONDER;    	
		
		msgCommandChannel.pcm.ensureSize(ServerResponseSchema.class, queueLength, maxMessageSize);
	}
	
	public boolean hasRoomFor(int messageCount) {
		return null==msgCommandChannel.goPipe || Pipe.hasRoomForWrite(msgCommandChannel.goPipe, 
		FieldReferenceOffsetManager.maxFragmentSize(Pipe.from(msgCommandChannel.goPipe))*messageCount);
	}
	

	public boolean publishHTTPResponse(HTTPFieldReader<?> reqeustReader, int statusCode) {
		assert((0 != (msgCommandChannel.initFeatures & MsgCommandChannel.NET_RESPONDER))) : "CommandChannel must be created with NET_RESPONDER flag";
		
		//logger.info("Building response for connection {} sequence {} ",w.getConnectionId(),w.getSequenceCode());
		
		return publishHTTPResponse(reqeustReader.getConnectionId(), reqeustReader.getSequenceCode(),
				statusCode,false,null,Writable.NO_OP); //no type and no body so use null
	}
	
	public boolean publishHTTPResponse(HTTPFieldReader<?> reqeustReader, 
			int statusCode,
		    HTTPContentType contentType,
		   	Writable writable) {
		assert((0 != (msgCommandChannel.initFeatures & MsgCommandChannel.NET_RESPONDER))) : "CommandChannel must be created with NET_RESPONDER flag";
		
		return publishHTTPResponse(reqeustReader.getConnectionId(), reqeustReader.getSequenceCode(),
									statusCode, false, contentType, writable);
	}
	
	public boolean publishHTTPResponse(HTTPFieldReader<?> reqeustReader, 
			   int statusCode, boolean hasContinuation,
			   HTTPContentType contentType,
			   Writable writable) {
		assert((0 != (msgCommandChannel.initFeatures & MsgCommandChannel.NET_RESPONDER))) : "CommandChannel must be created with NET_RESPONDER flag";
		
		return publishHTTPResponse(reqeustReader.getConnectionId(), reqeustReader.getSequenceCode(),
				                statusCode, hasContinuation, contentType, writable);	
	}
	
	public boolean publishHTTPResponse(long connectionId, long sequenceCode, int statusCode) {
		return publishHTTPResponse(connectionId, sequenceCode, statusCode, false, null, Writable.NO_OP);
	}
	
	public boolean publishHTTPResponse(long connectionId, long sequenceCode, 
            int statusCode, 
            boolean hasContinuation,
            HTTPContentType contentType,
            Writable writable) {
		assert(null!=writable);
		assert((0 != (msgCommandChannel.initFeatures & MsgCommandChannel.NET_RESPONDER))) : "CommandChannel must be created with NET_RESPONDER flag";
		
		final int sequenceNo = 0xFFFFFFFF & (int)sequenceCode;
		final int parallelIndex = 0xFFFFFFFF & (int)(sequenceCode>>32);
		
		assert(1==msgCommandChannel.lastResponseWriterFinished) : "Previous write was not ended can not start another.";
			
		Pipe<ServerResponseSchema> pipe = msgCommandChannel.netResponse.length>1 ? msgCommandChannel.netResponse[parallelIndex] : msgCommandChannel.netResponse[0];
		
		//logger.info("try new publishHTTPResponse "+pipe);
		if (!Pipe.hasRoomForWrite(pipe, 
				2*Pipe.sizeOf(pipe, ServerResponseSchema.MSG_TOCHANNEL_100))) {
			return false;
		}		
		//simple check to ensure we have room.
		assert(Pipe.workingHeadPosition(pipe)<(Pipe.tailPosition(pipe)+ pipe.sizeOfSlabRing  /*    pipe.slabMask*/  )) : "Working position is now writing into published(unreleased) tail "+
		Pipe.workingHeadPosition(pipe)+"<"+Pipe.tailPosition(pipe)+"+"+pipe.sizeOfSlabRing /*pipe.slabMask*/+" total "+((Pipe.tailPosition(pipe)+pipe.slabMask));
		
		
		///////////////////////////////////////
		//message 1 which contains the headers
		//////////////////////////////////////		
		msgCommandChannel.holdEmptyBlock(connectionId, sequenceNo, pipe);
		
		
		//check again because we have taken 2 spots now
		assert(Pipe.workingHeadPosition(pipe)<(Pipe.tailPosition(pipe)+ pipe.sizeOfSlabRing  /*    pipe.slabMask*/  )) : "Working position is now writing into published(unreleased) tail "+
		Pipe.workingHeadPosition(pipe)+"<"+Pipe.tailPosition(pipe)+"+"+pipe.sizeOfSlabRing /*pipe.slabMask*/+" total "+((Pipe.tailPosition(pipe)+pipe.slabMask));
		
		//////////////////////////////////////////
		//begin message 2 which contains the body
		//////////////////////////////////////////
		
		Pipe.addMsgIdx(pipe, ServerResponseSchema.MSG_TOCHANNEL_100);
		Pipe.addLongValue(connectionId, pipe);
		Pipe.addIntValue(sequenceNo, pipe);
		
		NetResponseWriter outputStream = (NetResponseWriter)Pipe.outputStream(pipe);
		
		int context;
		if (hasContinuation) {
			context = 0;
			msgCommandChannel.lastResponseWriterFinished = 0;
		} else {
			context = HTTPFieldReader.END_OF_RESPONSE;
			msgCommandChannel.lastResponseWriterFinished = 1;	
		}
		
		//NB: context passed in here is looked at to know if this is END_RESPONSE and if so
		//then the length is added if not then the header will designate chunked.
		outputStream.openField(statusCode, context, contentType);
		writable.write(outputStream); 
		
		if (hasContinuation) {
			// for chunking we must end this block			
			outputStream.write(MsgCommandChannel.RETURN_NEWLINE);
		}
		
		outputStream.publishWithHeader(msgCommandChannel.block1HeaderBlobPosition, msgCommandChannel.block1PositionOfLen); //closeLowLevelField and publish 
		
		return true;
				
	}

	public boolean publishHTTPResponse(HTTPFieldReader<?> reqeustReader, 
	           HeaderWritable headers, Writable writable) {
		return publishHTTPResponse(reqeustReader.getConnectionId(), reqeustReader.getSequenceCode(),
				false, headers, 200, writable
				);
	}
	
	public boolean publishHTTPResponse(long connectionId, long sequenceCode, 
	           boolean hasContinuation, HeaderWritable headers, int statusCode,
	           Writable writable) {
		assert((0 != (msgCommandChannel.initFeatures & MsgCommandChannel.NET_RESPONDER))) : "CommandChannel must be created with NET_RESPONDER flag";
		
		final int sequenceNo = 0xFFFFFFFF & (int)sequenceCode;
		final int parallelIndex = 0xFFFFFFFF & (int)(sequenceCode>>32);
		
		assert(1==msgCommandChannel.lastResponseWriterFinished) : "Previous write was not ended can not start another.";
		
		Pipe<ServerResponseSchema> pipe = msgCommandChannel.netResponse.length>1 ? msgCommandChannel.netResponse[parallelIndex] : msgCommandChannel.netResponse[0];
		
		if (!Pipe.hasRoomForWrite(pipe)) {
			return false;
		}		
						
		///////////////////////////////////////
		//message 1 which contains the headers
		//////////////////////////////////////		
		msgCommandChannel.holdEmptyBlock(connectionId, sequenceNo, pipe);
		<commit_msg>added assert check for payload of JSON<commit_after>package com.ociweb.gl.api;

import com.ociweb.pronghorn.network.config.HTTPContentType;
import com.ociweb.pronghorn.network.config.HTTPContentTypeDefaults;
import com.ociweb.pronghorn.network.config.HTTPRevisionDefaults;
import com.ociweb.pronghorn.network.module.AbstractAppendablePayloadResponseStage;
import com.ociweb.pronghorn.network.schema.ServerResponseSchema;
import com.ociweb.pronghorn.pipe.DataOutputBlobWriter;
import com.ociweb.pronghorn.pipe.FieldReferenceOffsetManager;
import com.ociweb.pronghorn.pipe.Pipe;
import com.ociweb.pronghorn.pipe.PipeWriter;
import com.ociweb.pronghorn.util.Appendables;

public class HTTPResponseService {

	private final MsgCommandChannel<?> msgCommandChannel;
	
	public HTTPResponseService(MsgCommandChannel<?> msgCommandChannel) {
		this.msgCommandChannel = msgCommandChannel;
		msgCommandChannel.initFeatures |= MsgCommandChannel.NET_RESPONDER;
	}
	
	public HTTPResponseService(MsgCommandChannel<?> msgCommandChannel,
			int queueLength, int maxMessageSize) {
		this.msgCommandChannel = msgCommandChannel;
		MsgCommandChannel.growCommandCountRoom(msgCommandChannel, queueLength);
		msgCommandChannel.initFeatures |= MsgCommandChannel.NET_RESPONDER;    	
		
		msgCommandChannel.pcm.ensureSize(ServerResponseSchema.class, queueLength, maxMessageSize);
	}
	
	public boolean hasRoomFor(int messageCount) {
		return null==msgCommandChannel.goPipe || Pipe.hasRoomForWrite(msgCommandChannel.goPipe, 
		FieldReferenceOffsetManager.maxFragmentSize(Pipe.from(msgCommandChannel.goPipe))*messageCount);
	}
	

	public boolean publishHTTPResponse(HTTPFieldReader<?> reqeustReader, int statusCode) {
		assert((0 != (msgCommandChannel.initFeatures & MsgCommandChannel.NET_RESPONDER))) : "CommandChannel must be created with NET_RESPONDER flag";
		
		//logger.info("Building response for connection {} sequence {} ",w.getConnectionId(),w.getSequenceCode());
		
		return publishHTTPResponse(reqeustReader.getConnectionId(), reqeustReader.getSequenceCode(),
				statusCode,false,null,Writable.NO_OP); //no type and no body so use null
	}
	
	public boolean publishHTTPResponse(HTTPFieldReader<?> reqeustReader, 
			int statusCode,
		    HTTPContentType contentType,
		   	Writable writable) {
		assert((0 != (msgCommandChannel.initFeatures & MsgCommandChannel.NET_RESPONDER))) : "CommandChannel must be created with NET_RESPONDER flag";
		
		return publishHTTPResponse(reqeustReader.getConnectionId(), reqeustReader.getSequenceCode(),
									statusCode, false, contentType, writable);
	}
	
	public boolean publishHTTPResponse(HTTPFieldReader<?> reqeustReader, 
			   int statusCode, boolean hasContinuation,
			   HTTPContentType contentType,
			   Writable writable) {
		assert((0 != (msgCommandChannel.initFeatures & MsgCommandChannel.NET_RESPONDER))) : "CommandChannel must be created with NET_RESPONDER flag";
		
		return publishHTTPResponse(reqeustReader.getConnectionId(), reqeustReader.getSequenceCode(),
				                statusCode, hasContinuation, contentType, writable);	
	}
	
	public boolean publishHTTPResponse(long connectionId, long sequenceCode, int statusCode) {
		return publishHTTPResponse(connectionId, sequenceCode, statusCode, false, null, Writable.NO_OP);
	}
	
	public boolean publishHTTPResponse(long connectionId, long sequenceCode, 
            int statusCode, 
            boolean hasContinuation,
            HTTPContentType contentType,
            Writable writable) {
		assert(null!=writable);
		assert((0 != (msgCommandChannel.initFeatures & MsgCommandChannel.NET_RESPONDER))) : "CommandChannel must be created with NET_RESPONDER flag";
		
		final int sequenceNo = 0xFFFFFFFF & (int)sequenceCode;
		final int parallelIndex = 0xFFFFFFFF & (int)(sequenceCode>>32);
		
		assert(1==msgCommandChannel.lastResponseWriterFinished) : "Previous write was not ended can not start another.";
			
		Pipe<ServerResponseSchema> pipe = msgCommandChannel.netResponse.length>1 ? msgCommandChannel.netResponse[parallelIndex] : msgCommandChannel.netResponse[0];
		
		//logger.info("try new publishHTTPResponse "+pipe);
		if (!Pipe.hasRoomForWrite(pipe, 
				2*Pipe.sizeOf(pipe, ServerResponseSchema.MSG_TOCHANNEL_100))) {
			return false;
		}		
		//simple check to ensure we have room.
		assert(Pipe.workingHeadPosition(pipe)<(Pipe.tailPosition(pipe)+ pipe.sizeOfSlabRing  /*    pipe.slabMask*/  )) : "Working position is now writing into published(unreleased) tail "+
		Pipe.workingHeadPosition(pipe)+"<"+Pipe.tailPosition(pipe)+"+"+pipe.sizeOfSlabRing /*pipe.slabMask*/+" total "+((Pipe.tailPosition(pipe)+pipe.slabMask));
		
		
		///////////////////////////////////////
		//message 1 which contains the headers
		//////////////////////////////////////		
		msgCommandChannel.holdEmptyBlock(connectionId, sequenceNo, pipe);
		
		
		//check again because we have taken 2 spots now
		assert(Pipe.workingHeadPosition(pipe)<(Pipe.tailPosition(pipe)+ pipe.sizeOfSlabRing  /*    pipe.slabMask*/  )) : "Working position is now writing into published(unreleased) tail "+
		Pipe.workingHeadPosition(pipe)+"<"+Pipe.tailPosition(pipe)+"+"+pipe.sizeOfSlabRing /*pipe.slabMask*/+" total "+((Pipe.tailPosition(pipe)+pipe.slabMask));
		
		//////////////////////////////////////////
		//begin message 2 which contains the body
		//////////////////////////////////////////
		
		Pipe.addMsgIdx(pipe, ServerResponseSchema.MSG_TOCHANNEL_100);
		Pipe.addLongValue(connectionId, pipe);
		Pipe.addIntValue(sequenceNo, pipe);
		
		NetResponseWriter outputStream = (NetResponseWriter)Pipe.outputStream(pipe);
		
		int context;
		if (hasContinuation) {
			context = 0;
			msgCommandChannel.lastResponseWriterFinished = 0;
		} else {
			context = HTTPFieldReader.END_OF_RESPONSE;
			msgCommandChannel.lastResponseWriterFinished = 1;	
		}
		
		//NB: context passed in here is looked at to know if this is END_RESPONSE and if so
		//then the length is added if not then the header will designate chunked.
		outputStream.openField(statusCode, context, contentType);
		writable.write(outputStream); 
		
		if (hasContinuation) {
			// for chunking we must end this block			
			outputStream.write(MsgCommandChannel.RETURN_NEWLINE);
		}
		
		assert(isValidContent(contentType,outputStream)) : "content type is not matching payload";
				
		
		outputStream.publishWithHeader(msgCommandChannel.block1HeaderBlobPosition, msgCommandChannel.block1PositionOfLen); //closeLowLevelField and publish 
		
		return true;
				
	}

	private boolean isValidContent(HTTPContentType contentType, NetResponseWriter outputStream) {
		StringBuilder target = new StringBuilder();
		
		if (HTTPContentTypeDefaults.JSON == contentType ||
			HTTPContentTypeDefaults.TXT == contentType) {		
			//simple check to make sure JSON starts with text not the length
			outputStream.debugAsUTF8(target);
			if (target.length()>0) {
				if (target.charAt(0)<32) {
					return false;
				}
				if (target.length()>1) {
					if (target.charAt(1)<32) {
						return false;
					}	
				}
			}
		}
		
		return true;
	}

	public boolean publishHTTPResponse(HTTPFieldReader<?> reqeustReader, 
	           HeaderWritable headers, Writable writable) {
		return publishHTTPResponse(reqeustReader.getConnectionId(), reqeustReader.getSequenceCode(),
				false, headers, 200, writable
				);
	}
	
	public boolean publishHTTPResponse(long connectionId, long sequenceCode, 
	           boolean hasContinuation, HeaderWritable headers, int statusCode,
	           Writable writable) {
		assert((0 != (msgCommandChannel.initFeatures & MsgCommandChannel.NET_RESPONDER))) : "CommandChannel must be created with NET_RESPONDER flag";
		
		final int sequenceNo = 0xFFFFFFFF & (int)sequenceCode;
		final int parallelIndex = 0xFFFFFFFF & (int)(sequenceCode>>32);
		
		assert(1==msgCommandChannel.lastResponseWriterFinished) : "Previous write was not ended can not start another.";
		
		Pipe<ServerResponseSchema> pipe = msgCommandChannel.netResponse.length>1 ? msgCommandChannel.netResponse[parallelIndex] : msgCommandChannel.netResponse[0];
		
		if (!Pipe.hasRoomForWrite(pipe)) {
			return false;
		}		
						
		///////////////////////////////////////
		//message 1 which contains the headers
		//////////////////////////////////////		
		msgCommandChannel.holdEmptyBlock(connectionId, sequenceNo, pipe);
		
{'fraction_non_alphanumeric': 0.12185905913470621, 'fraction_numerical': 0.005352981925813968, 'mean_word_length': 4.260019874130506, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 35, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 8, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6611104', 'n_tokens_mistral': 5002, 'n_tokens_neox': 4387, 'n_words': 927}