1 /*
2 * Copyright 2016 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 * https://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
16 package io.netty.handler.codec.redis;
17
18 import io.netty.buffer.ByteBuf;
19 import io.netty.channel.ChannelHandler;
20 import io.netty.channel.ChannelPipeline;
21 import io.netty.handler.codec.MessageAggregator;
22 import io.netty.util.internal.UnstableApi;
23
24 /**
25 * A {@link ChannelHandler} that aggregates an {@link BulkStringHeaderRedisMessage}
26 * and its following {@link BulkStringRedisContent}s into a single {@link FullBulkStringRedisMessage}
27 * with no following {@link BulkStringRedisContent}s. It is useful when you don't want to take
28 * care of {@link RedisMessage}s whose transfer encoding is 'chunked'. Insert this
29 * handler after {@link RedisDecoder} in the {@link ChannelPipeline}:
30 * <pre>
31 * {@link ChannelPipeline} p = ...;
32 * ...
33 * p.addLast("encoder", new {@link RedisEncoder}());
34 * p.addLast("decoder", new {@link RedisDecoder}());
35 * p.addLast("aggregator", <b>new {@link RedisBulkStringAggregator}()</b>);
36 * ...
37 * p.addLast("handler", new HttpRequestHandler());
38 * </pre>
39 * Be aware that you need to have the {@link RedisEncoder} before the {@link RedisBulkStringAggregator}
40 * in the {@link ChannelPipeline}.
41 */
42 @UnstableApi
43 public final class RedisBulkStringAggregator extends MessageAggregator<RedisMessage, BulkStringHeaderRedisMessage,
44 BulkStringRedisContent, FullBulkStringRedisMessage> {
45
46 /**
47 * Creates a new instance.
48 */
49 public RedisBulkStringAggregator() {
50 super(RedisConstants.REDIS_MESSAGE_MAX_LENGTH);
51 }
52
53 @Override
54 protected boolean isStartMessage(RedisMessage msg) throws Exception {
55 return msg instanceof BulkStringHeaderRedisMessage && !isAggregated(msg);
56 }
57
58 @Override
59 protected boolean isContentMessage(RedisMessage msg) throws Exception {
60 return msg instanceof BulkStringRedisContent;
61 }
62
63 @Override
64 protected boolean isLastContentMessage(BulkStringRedisContent msg) throws Exception {
65 return msg instanceof LastBulkStringRedisContent;
66 }
67
68 @Override
69 protected boolean isAggregated(RedisMessage msg) throws Exception {
70 return msg instanceof FullBulkStringRedisMessage;
71 }
72
73 @Override
74 protected boolean isContentLengthInvalid(BulkStringHeaderRedisMessage start, int maxContentLength)
75 throws Exception {
76 return start.bulkStringLength() > maxContentLength;
77 }
78
79 @Override
80 protected Object newContinueResponse(BulkStringHeaderRedisMessage start, int maxContentLength,
81 ChannelPipeline pipeline) throws Exception {
82 return null;
83 }
84
85 @Override
86 protected boolean closeAfterContinueResponse(Object msg) throws Exception {
87 throw new UnsupportedOperationException();
88 }
89
90 @Override
91 protected boolean ignoreContentAfterContinueResponse(Object msg) throws Exception {
92 throw new UnsupportedOperationException();
93 }
94
95 @Override
96 protected FullBulkStringRedisMessage beginAggregation(BulkStringHeaderRedisMessage start, ByteBuf content)
97 throws Exception {
98 return new FullBulkStringRedisMessage(content);
99 }
100 }
以上源码来自： 即时通讯网 - 即时通讯开发者社区！