public final ByteBuf remove(ByteBufAllocator alloc, int bytes, ChannelPromise aggregatePromise) {
        checkPositiveOrZero(bytes, "bytes");
        checkNotNull(aggregatePromise, "aggregatePromise");

        // Use isEmpty rather than readableBytes==0 as we may have a promise associated with an empty buffer.
        if (bufAndListenerPairs.isEmpty()) {
            return removeEmptyValue();
        }
        bytes = Math.min(bytes, readableBytes);

        ByteBuf toReturn = null;
        ByteBuf entryBuffer = null;
        int originalBytes = bytes;
        try {
            for (;;) {
                Object entry = bufAndListenerPairs.poll();
                if (entry == null) {
                    break;
                }
                if (entry instanceof ChannelFutureListener) {
                    aggregatePromise.addListener((ChannelFutureListener) entry);
                    continue;
                }
                entryBuffer = (ByteBuf) entry;
                if (entryBuffer.readableBytes() > bytes) {
                    // Add the buffer back to the queue as we can't consume all of it.
                    bufAndListenerPairs.addFirst(entryBuffer);
                    if (bytes > 0) {
                        // Take a slice of what we can consume and retain it.
                        entryBuffer = entryBuffer.readRetainedSlice(bytes);
                        toReturn = toReturn == null ? composeFirst(alloc, entryBuffer)
                                                    : compose(alloc, toReturn, entryBuffer);
                        bytes = 0;
                    }
                    break;
                } else {
                    bytes -= entryBuffer.readableBytes();
                    toReturn = toReturn == null ? composeFirst(alloc, entryBuffer)
                                                : compose(alloc, toReturn, entryBuffer);
                }
                entryBuffer = null;
            }
        } catch (Throwable cause) {
            safeRelease(entryBuffer);
            safeRelease(toReturn);
            aggregatePromise.setFailure(cause);
            throwException(cause);
        }
        decrementReadableBytes(originalBytes - bytes);
        return toReturn;
    }
public final ByteBuf remove ( ByteBufAllocator alloc , int bytes , ChannelPromise aggregatePromise ) { checkPositiveOrZero ( bytes , "bytes" ) ; checkNotNull ( aggregatePromise , "aggregatePromise" ) ; if ( bufAndListenerPairs . isEmpty ( ) ) { return removeEmptyValue ( ) ; } bytes = Math . min ( bytes , readableBytes ) ; ByteBuf toReturn = null ; ByteBuf entryBuffer = null ; int originalBytes = bytes ; try { for ( ; ; ) { Object entry = bufAndListenerPairs . poll ( ) ; if ( entry == null ) { break ; } if ( entry instanceof ChannelFutureListener ) { aggregatePromise . addListener ( ( ChannelFutureListener ) entry ) ; continue ; } entryBuffer = ( ByteBuf ) entry ; if ( entryBuffer . readableBytes ( ) > bytes ) { bufAndListenerPairs . addFirst ( entryBuffer ) ; if ( bytes > 0 ) { entryBuffer = entryBuffer . readRetainedSlice ( bytes ) ; toReturn = toReturn == null ? composeFirst ( alloc , entryBuffer ) : compose ( alloc , toReturn , entryBuffer ) ; bytes = 0 ; } break ; } else { bytes -= entryBuffer . readableBytes ( ) ; toReturn = toReturn == null ? composeFirst ( alloc , entryBuffer ) : compose ( alloc , toReturn , entryBuffer ) ; } entryBuffer = null ; } } catch ( Throwable cause ) { safeRelease ( entryBuffer ) ; safeRelease ( toReturn ) ; aggregatePromise . setFailure ( cause ) ; throwException ( cause ) ; } decrementReadableBytes ( originalBytes - bytes ) ; return toReturn ; }
Remove a {@link ByteBuf} from the queue with the specified number of bytes. Any added buffer who's bytes are
fully consumed during removal will have it's promise completed when the passed aggregate {@link ChannelPromise}
completes.

@param alloc The allocator used if a new {@link ByteBuf} is generated during the aggregation process.
@param bytes the maximum number of readable bytes in the returned {@link ByteBuf}, if {@code bytes} is greater
than {@link #readableBytes} then a buffer of length {@link #readableBytes} is returned.
@param aggregatePromise used to aggregate the promises and listeners for the constituent buffers.
@return a {@link ByteBuf} composed of the enqueued buffers.