View Javadoc
1   /*
2    * Copyright 2015 The Netty Project
3    *
4    * The Netty Project licenses this file to you under the Apache License, version 2.0 (the
5    * "License"); you may not use this file except in compliance with the License. You may obtain a
6    * copy of the License at:
7    *
8    * http://www.apache.org/licenses/LICENSE-2.0
9    *
10   * Unless required by applicable law or agreed to in writing, software distributed under the License
11   * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
12   * or implied. See the License for the specific language governing permissions and limitations under
13   * the License.
14   */
15  package io.netty.channel;
16  
17  import io.netty.buffer.ByteBuf;
18  import io.netty.buffer.ByteBufAllocator;
19  import io.netty.buffer.CompositeByteBuf;
20  import io.netty.buffer.Unpooled;
21  import io.netty.util.internal.ObjectUtil;
22  import io.netty.util.internal.PlatformDependent;
23  
24  /**
25   * A FIFO queue of bytes where producers add bytes by repeatedly adding {@link ByteBuf} and consumers take bytes in
26   * arbitrary lengths. This allows producers to add lots of small buffers and the consumer to take all the bytes
27   * out in a single buffer. Conversely the producer may add larger buffers and the consumer could take the bytes in
28   * many small buffers.
29   *
30   * <p>Bytes are added and removed with promises. If the last byte of a buffer added with a promise is removed then
31   * that promise will complete when the promise passed to {@link #remove} completes.
32   *
33   * <p>This functionality is useful for aggregating or partitioning writes into fixed size buffers for framing protocols
34   * such as HTTP2.
35   */
36  public final class CoalescingBufferQueue extends AbstractCoalescingBufferQueue {
37      private final Channel channel;
38  
39      public CoalescingBufferQueue(Channel channel) {
40          this(channel, 4);
41      }
42  
43      public CoalescingBufferQueue(Channel channel, int initSize) {
44          this(channel, initSize, false);
45      }
46  
47      public CoalescingBufferQueue(Channel channel, int initSize, boolean updateWritability) {
48          super(updateWritability ? channel : null, initSize);
49          this.channel = ObjectUtil.checkNotNull(channel, "channel");
50      }
51  
52      /**
53       * Remove a {@link ByteBuf} from the queue with the specified number of bytes. Any added buffer who's bytes are
54       * fully consumed during removal will have it's promise completed when the passed aggregate {@link ChannelPromise}
55       * completes.
56       *
57       * @param bytes the maximum number of readable bytes in the returned {@link ByteBuf}, if {@code bytes} is greater
58       *              than {@link #readableBytes} then a buffer of length {@link #readableBytes} is returned.
59       * @param aggregatePromise used to aggregate the promises and listeners for the constituent buffers.
60       * @return a {@link ByteBuf} composed of the enqueued buffers.
61       */
62      public ByteBuf remove(int bytes, ChannelPromise aggregatePromise) {
63          return remove(channel.alloc(), bytes, aggregatePromise);
64      }
65  
66      /**
67       *  Release all buffers in the queue and complete all listeners and promises.
68       */
69      public void releaseAndFailAll(Throwable cause) {
70          releaseAndFailAll(channel, cause);
71      }
72  
73      @Override
74      protected ByteBuf compose(ByteBufAllocator alloc, ByteBuf cumulation, ByteBuf next) {
75          if (cumulation instanceof CompositeByteBuf) {
76              CompositeByteBuf composite = (CompositeByteBuf) cumulation;
77              composite.addComponent(true, next);
78              return composite;
79          }
80          return composeIntoComposite(alloc, cumulation, next);
81      }
82  
83      @Override
84      protected ByteBuf removeEmptyValue() {
85          return Unpooled.EMPTY_BUFFER;
86      }
87  }