private void parseViewRows ( boolean last ) { while ( true ) { int openBracketPos = responseContent . bytesBefore ( ( byte ) ' ' ) ; int errorBlockPosition = findErrorBlockPosition ( openBracketPos ) ; if ( errorBlockPosition > 0 && errorBlockPosition < openBracketPos ) { responseContent . readerIndex ( errorBlockPosition + responseContent . readerIndex ( ) ) ; viewRowObservable . onCompleted ( ) ; viewParsingState = QUERY_STATE_ERROR ; return ; } int closeBracketPos = findSectionClosingPosition ( responseContent , ' ' , ' ' ) ; if ( closeBracketPos == - 1 ) { break ; } int from = responseContent . readerIndex ( ) + openBracketPos ; int to = closeBracketPos - openBracketPos - responseContent . readerIndex ( ) + 1 ; viewRowObservable . onNext ( responseContent . slice ( from , to ) . copy ( ) ) ; responseContent . readerIndex ( closeBracketPos ) ; responseContent . discardReadBytes ( ) ; } if ( last ) { viewRowObservable . onCompleted ( ) ; viewErrorObservable . onCompleted ( ) ; viewParsingState = QUERY_STATE_DONE ; } }
Streaming parse the actual rows from the response and pass to the underlying observable .