private void triggerPipeliningLogic ( HttpCarbonMessage outboundResponseMsg ) { String httpVersion = ( String ) inboundRequestMsg . getProperty ( Constants . HTTP_VERSION ) ; if ( outboundResponseMsg . isPipeliningEnabled ( ) && Constants . HTTP_1_1_VERSION . equalsIgnoreCase ( httpVersion ) ) { Queue responseQueue ; synchronized ( sourceContext . channel ( ) . attr ( Constants . RESPONSE_QUEUE ) . get ( ) ) { responseQueue = sourceContext . channel ( ) . attr ( Constants . RESPONSE_QUEUE ) . get ( ) ; Long nextSequenceNumber = sourceContext . channel ( ) . attr ( Constants . NEXT_SEQUENCE_NUMBER ) . get ( ) ; //IMPORTANT:Next sequence number should never be incremented for interim 100 continue response //because the body of the request is yet to come. Only when the actual response is sent out, this //next sequence number should be updated. nextSequenceNumber ++ ; sourceContext . channel ( ) . attr ( Constants . NEXT_SEQUENCE_NUMBER ) . set ( nextSequenceNumber ) ; if ( LOG . isDebugEnabled ( ) ) { LOG . debug ( "Current sequence id of the response : {}" , outboundResponseMsg . getSequenceId ( ) ) ; LOG . debug ( "Updated next sequence id to : {}" , nextSequenceNumber ) ; } } if ( ! responseQueue . isEmpty ( ) ) { if ( LOG . isDebugEnabled ( ) ) { LOG . debug ( "Pipelining logic is triggered from transport" ) ; } //Notify ballerina to send the response which is next in queue. This is needed because, //if the other responses got ready before the nextSequenceNumber gets updated then the //ballerina respond() won't start serializing the responses in queue. This is to trigger //that process again. if ( outboundResponseMsg . getPipeliningFuture ( ) != null ) { EventExecutorGroup pipeliningExecutor = sourceContext . channel ( ) . attr ( Constants . PIPELINING_EXECUTOR ) . get ( ) ; //IMPORTANT:Pipelining logic should never be executed in an I/O thread as it might lead to I/O //thread blocking scenarios in outbound trottling. Here, the pipelining logic runs in a thread that //belongs to the pipelining thread pool. pipeliningExecutor . execute ( ( ) -> outboundResponseMsg . getPipeliningFuture ( ) . notifyPipeliningListener ( sourceContext ) ) ; } } } }
Increment the next expected sequence number and trigger the pipelining logic .