1
2
3
4
5
6
7
8
9
10
11
12
13
14
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
36
37
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
54
55
56
57
58 public SctpMessageCompletionHandler(int maxIncompleteSctpMessages, int maxFragments) {
59 this(maxIncompleteSctpMessages, maxFragments, DEFAULT_MAX_BUFFERED_BYTES);
60 }
61
62
63
64
65
66
67
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
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
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
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
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 }