View Javadoc
1   /*
2    * Copyright 2012 The Netty Project
3    *
4    * The Netty Project licenses this file to you under the Apache License,
5    * version 2.0 (the "License"); you may not use this file except in compliance
6    * with the License. You may obtain a copy of the License at:
7    *
8    * https://www.apache.org/licenses/LICENSE-2.0
9    *
10   * Unless required by applicable law or agreed to in writing, software
11   * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12   * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13   * License for the specific language governing permissions and limitations
14   * under the License.
15   */
16  
17  package io.netty.handler.codec.sctp;
18  
19  import io.netty.buffer.ByteBuf;
20  import io.netty.buffer.CompositeByteBuf;
21  import io.netty.channel.ChannelHandlerContext;
22  import io.netty.channel.ChannelInboundHandler;
23  import io.netty.channel.sctp.SctpMessage;
24  import io.netty.handler.codec.CodecException;
25  import io.netty.handler.codec.MessageToMessageDecoder;
26  import io.netty.util.collection.IntObjectHashMap;
27  import io.netty.util.collection.IntObjectMap;
28  
29  import java.util.ArrayList;
30  import java.util.List;
31  
32  import static io.netty.util.internal.ObjectUtil.checkPositive;
33  
34  /**
35   * {@link MessageToMessageDecoder} which will take care of handle fragmented {@link SctpMessage}s, so
36   * only <strong>complete</strong> {@link SctpMessage}s will be forwarded to the next
37   * {@link ChannelInboundHandler}.
38   */
39  public class SctpMessageCompletionHandler extends MessageToMessageDecoder<SctpMessage> {
40      private static final int DEFAULT_MAX_BUFFERED_BYTES = 16 * 1024 * 1024;
41  
42      private final IntObjectMap<List<ByteBuf>> incompleteSctpMessages = new IntObjectHashMap<List<ByteBuf>>();
43      private final int maxIncompleteSctpMessages;
44      private final int maxFragments;
45      private final int maxBufferedBytes;
46      private long bufferedBytes;
47  
48      public SctpMessageCompletionHandler() {
49          this(128, 128);
50      }
51  
52      /**
53       * Create a new instance.
54       *
55       * @param maxIncompleteSctpMessages the maximum number of incomplete sctp message inflight.
56       * @param maxFragments              the maximum number of fragments per sctp message.
57       */
58      public SctpMessageCompletionHandler(int maxIncompleteSctpMessages, int maxFragments) {
59          this(maxIncompleteSctpMessages, maxFragments, DEFAULT_MAX_BUFFERED_BYTES);
60      }
61  
62      /**
63       * Create a new instance.
64       *
65       * @param maxIncompleteSctpMessages the maximum number of incomplete sctp message inflight.
66       * @param maxFragments              the maximum number of fragments per sctp message.
67       * @param maxBufferedBytes          the maximum number of bytes buffered by incomplete sctp messages.
68       */
69      public SctpMessageCompletionHandler(int maxIncompleteSctpMessages, int maxFragments, int maxBufferedBytes) {
70          super(SctpMessage.class);
71          this.maxIncompleteSctpMessages = checkPositive(maxIncompleteSctpMessages, "maxIncompleteSctpMessages");
72          this.maxFragments = checkPositive(maxFragments, "maxFragments");
73          this.maxBufferedBytes = checkPositive(maxBufferedBytes, "maxBufferedBytes");
74      }
75  
76      @Override
77      protected void decode(ChannelHandlerContext ctx, SctpMessage msg, List<Object> out) throws Exception {
78          final ByteBuf byteBuf = msg.content();
79          final int protocolIdentifier = msg.protocolIdentifier();
80          final int streamIdentifier = msg.streamIdentifier();
81          final boolean isComplete = msg.isComplete();
82          final boolean isUnordered = msg.isUnordered();
83  
84          List<ByteBuf> frag = incompleteSctpMessages.get(streamIdentifier);
85          if (frag == null) {
86              // No previous fragments.
87              if (isComplete) {
88                  out.add(msg.retain());
89              } else {
90                  if (maxIncompleteSctpMessages <= incompleteSctpMessages.size()) {
91                      throw new CodecException(
92                              "Too many incomplete sctp messages in flight: " + maxIncompleteSctpMessages);
93                  }
94                  checkBufferedBytes(byteBuf);
95                  //first incomplete message
96                  frag = new ArrayList<ByteBuf>();
97                  frag.add(byteBuf.retain());
98                  bufferedBytes += byteBuf.readableBytes();
99                  incompleteSctpMessages.put(streamIdentifier, frag);
100             }
101         } else {
102             if (maxFragments <= frag.size()) {
103                 throw new CodecException("Too many fragments for sctp message: " + maxFragments);
104             }
105             checkBufferedBytes(byteBuf);
106             frag.add(byteBuf.retain());
107             bufferedBytes += byteBuf.readableBytes();
108             if (isComplete) {
109                 // Is complete so remove it.
110                 incompleteSctpMessages.remove(streamIdentifier);
111                 CompositeByteBuf composite = ctx.alloc().compositeBuffer();
112 
113                 for (int i = 0; i < frag.size(); i++) {
114                     composite.addComponent(true, frag.get(i));
115                 }
116                 // last message to complete
117                 SctpMessage assembledMsg = new SctpMessage(
118                         protocolIdentifier,
119                         streamIdentifier,
120                         isUnordered,
121                         composite);
122                 out.add(assembledMsg);
123                 removeBufferedBytes(frag);
124             }
125         }
126     }
127 
128     private void checkBufferedBytes(ByteBuf byteBuf) {
129         int readableBytes = byteBuf.readableBytes();
130         if (readableBytes > maxBufferedBytes - bufferedBytes) {
131             throw new CodecException("Too many buffered bytes for incomplete sctp messages: " + maxBufferedBytes);
132         }
133     }
134 
135     private void removeBufferedBytes(List<ByteBuf> buffers) {
136         for (ByteBuf buffer : buffers) {
137             bufferedBytes -= buffer.readableBytes();
138         }
139     }
140 
141     @Override
142     public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
143         for (List<ByteBuf> buffers: incompleteSctpMessages.values()) {
144             for (ByteBuf buffer: buffers) {
145                 buffer.release();
146             }
147         }
148         incompleteSctpMessages.clear();
149         bufferedBytes = 0;
150         super.handlerRemoved(ctx);
151     }
152 
153     @Override
154     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
155         super.exceptionCaught(ctx, cause);
156         ctx.close();
157     }
158 }