private void writeAndFlushNextMessageIfPossible ( Channel channel ) { if ( channelError . get ( ) != null || ! channel . isWritable ( ) ) { return ; } while ( true ) { RemoteInputChannel inputChannel = inputChannelsWithCredit . poll ( ) ; // The input channel may be null because of the write callbacks // that are executed after each write. if ( inputChannel == null ) { return ; } //It is no need to notify credit for the released channel. if ( ! inputChannel . isReleased ( ) ) { AddCredit msg = new AddCredit ( inputChannel . getPartitionId ( ) , inputChannel . getAndResetUnannouncedCredit ( ) , inputChannel . getInputChannelId ( ) ) ; // Write and flush and wait until this is done before // trying to continue with the next input channel. channel . writeAndFlush ( msg ) . addListener ( writeListener ) ; return ; } } }
Tries to write&flush unannounced credits for the next input channel in queue .