1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16 package io.netty.handler.codec.redis;
17
18 import io.netty.channel.ChannelHandlerContext;
19 import io.netty.handler.codec.CodecException;
20 import io.netty.handler.codec.MessageToMessageDecoder;
21 import io.netty.handler.codec.PrematureChannelClosureException;
22 import io.netty.util.ReferenceCountUtil;
23 import io.netty.util.internal.ObjectUtil;
24 import io.netty.util.internal.UnstableApi;
25
26 import java.util.ArrayDeque;
27 import java.util.ArrayList;
28 import java.util.Deque;
29 import java.util.List;
30
31
32
33
34
35 @UnstableApi
36 public final class RedisArrayAggregator extends MessageToMessageDecoder<RedisMessage> {
37
38 private static final int DEFAULT_MAX_ARRAY_LENGTH = RedisConstants.REDIS_MAX_ARRAY_LENGTH;
39 private final int maxNestedArrayDepth;
40 private final Deque<AggregateState> depths = new ArrayDeque<AggregateState>(4);
41 private final int maxElements;
42
43
44
45
46
47
48
49
50
51
52 @Deprecated
53 public RedisArrayAggregator() {
54
55 this(DEFAULT_MAX_ARRAY_LENGTH, 1024);
56 }
57
58
59
60
61
62
63
64
65
66
67 public RedisArrayAggregator(int maxElements, int maxNestedArrayDepth) {
68 super(RedisMessage.class);
69 this.maxElements = ObjectUtil.checkPositive(maxElements, "maxElements");
70 this.maxNestedArrayDepth = ObjectUtil.checkPositive(maxNestedArrayDepth, "maxNestedArrayDepth");
71 }
72
73 @Override
74 protected void decode(ChannelHandlerContext ctx, RedisMessage msg, List<Object> out) throws Exception {
75 if (msg instanceof ArrayHeaderRedisMessage) {
76 msg = decodeRedisArrayHeader((ArrayHeaderRedisMessage) msg);
77 if (msg == null) {
78 return;
79 }
80 } else {
81 ReferenceCountUtil.retain(msg);
82 }
83
84 while (!depths.isEmpty()) {
85 AggregateState current = depths.peek();
86 current.children.add(msg);
87
88
89 if (current.children.size() == current.length) {
90 msg = new ArrayRedisMessage(current.children);
91 depths.pop();
92 } else {
93
94 return;
95 }
96 }
97
98 out.add(msg);
99 }
100
101 private CodecException clearAndCreateException(String msg) {
102 releaseAndClearDepths();
103 return new CodecException(msg);
104 }
105
106 private RedisMessage decodeRedisArrayHeader(ArrayHeaderRedisMessage header) {
107 if (header.isNull()) {
108 return ArrayRedisMessage.NULL_INSTANCE;
109 } else if (header.length() == 0L) {
110 return ArrayRedisMessage.EMPTY_INSTANCE;
111 } else if (header.length() > 0L) {
112
113 if (header.length() > maxElements) {
114 throw clearAndCreateException("this codec doesn't support longer length than " + maxElements);
115 }
116
117 if (depths.size() >= maxNestedArrayDepth) {
118 throw clearAndCreateException("max nested array depth exceeded: " + maxNestedArrayDepth);
119 }
120
121 depths.push(new AggregateState((int) header.length()));
122 return null;
123 } else {
124 throw clearAndCreateException("bad length: " + header.length());
125 }
126 }
127
128 private static final class AggregateState {
129 private final int length;
130 private final List<RedisMessage> children;
131 AggregateState(int length) {
132 this.length = length;
133 this.children = new ArrayList<RedisMessage>(length);
134 }
135 }
136
137 @Override
138 public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
139 super.handlerRemoved(ctx);
140 releaseAndClearDepths();
141 }
142
143 private void releaseAndClearDepths() {
144 for (AggregateState state : depths) {
145 for (RedisMessage message : state.children) {
146 ReferenceCountUtil.safeRelease(message);
147 }
148 }
149 depths.clear();
150 }
151
152 @Override
153 public void channelInactive(ChannelHandlerContext ctx) throws Exception {
154 super.channelInactive(ctx);
155
156 if (!depths.isEmpty()) {
157 ctx.fireExceptionCaught(new PrematureChannelClosureException(
158 "channel gone inactive with " + depths.size() +
159 " messages still incomplete"));
160 }
161 }
162 }