@ Override public final int encode ( ValueReference < ByteBuffer > data , int size ) { int bufferSize = size ; ByteBuffer buf = data . get ( ) ; if ( buf == null ) { buf = this . buffer ; bufferSize = this . bufferSize ; buffer . clear ( ) ; } if ( inProgress == null ) { return 0 ; } int pos = 0 ; buf . limit ( buf . capacity ( ) ) ; while ( pos < bufferSize ) { // If there are no more data to return, run the state machine. // If there are still no data, return what we already have // in the buffer. if ( toWrite == 0 ) { if ( newMsgFlag ) { inProgress = null ; break ; } next ( ) ; } // If there are no data in the buffer yet and we are able to // fill whole buffer in a single go, let's use zero-copy. // There's no disadvantage to it as we cannot stuck multiple // messages into the buffer anyway. Note that subsequent // write(s) are non-blocking, thus each single write writes // at most SO_SNDBUF bytes at once not depending on how large // is the chunk returned from here. // As a consequence, large messages being sent won't block // other engines running in the same I/O thread for excessive // amounts of time. if ( pos == 0 && data . get ( ) == null && toWrite >= bufferSize ) { writeBuf . limit ( writeBuf . capacity ( ) ) ; data . set ( writeBuf ) ; pos = toWrite ; writeBuf = null ; toWrite = 0 ; return pos ; } // Copy data to the buffer. If the buffer is full, return. int toCopy = Math . min ( toWrite , bufferSize - pos ) ; int limit = writeBuf . limit ( ) ; writeBuf . limit ( Math . min ( writeBuf . capacity ( ) , writeBuf . position ( ) + toCopy ) ) ; int current = buf . position ( ) ; buf . put ( writeBuf ) ; toCopy = buf . position ( ) - current ; writeBuf . limit ( limit ) ; pos += toCopy ; toWrite -= toCopy ; } data . set ( buf ) ; return pos ; }
is NULL ) encoder will provide buffer of its own .