private void decodeChunk ( InetSocketAddress address , final ByteBuf chunk ) { responseContent . writeBytes ( chunk ) ; String currentChunk = responseContent . toString ( CharsetUtil . UTF_8 ) ; int separatorIndex = currentChunk . indexOf ( "\n\n\n\n" ) ; if ( separatorIndex > 0 ) { String rawConfig = currentChunk . substring ( 0 , separatorIndex ) . trim ( ) . replace ( "$HOST" , address . getAddress ( ) . getHostAddress ( ) ) ; NetworkAddress origin = NetworkAddress . create ( address . getAddress ( ) . getHostAddress ( ) ) ; CouchbaseBucketConfig config = ( CouchbaseBucketConfig ) BucketConfigParser . parse ( rawConfig , environment , origin ) ; synchronized ( currentBucketConfigRev ) { if ( config . rev ( ) > currentBucketConfigRev . get ( ) ) { LOGGER . trace ( "Publishing bucket config: {}" , RedactableArgument . system ( rawConfig ) ) ; currentBucketConfigRev . set ( config . rev ( ) ) ; configStream . onNext ( config ) ; } else { LOGGER . trace ( "Ignoring config, since rev has not changed." ) ; } } responseContent . clear ( ) ; responseContent . writeBytes ( currentChunk . substring ( separatorIndex + 4 ) . getBytes ( CharsetUtil . UTF_8 ) ) ; } }
Helper method to decode and analyze the chunk .