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.buffer.ByteBufAllocator;
20 import io.netty.buffer.ByteBufUtil;
21 import io.netty.channel.ChannelHandlerContext;
22 import io.netty.handler.codec.CodecException;
23 import io.netty.handler.codec.MessageToMessageEncoder;
24 import io.netty.util.internal.ObjectUtil;
25 import io.netty.util.internal.UnstableApi;
26
27 import java.util.List;
28
29 /**
30 * Encodes {@link RedisMessage} into bytes following
31 * <a href="https://redis.io/topics/protocol">RESP (REdis Serialization Protocol)</a>.
32 */
33 @UnstableApi
34 public class RedisEncoder extends MessageToMessageEncoder<RedisMessage> {
35
36 private final RedisMessagePool messagePool;
37
38 /**
39 * Creates a new instance with default {@code messagePool}.
40 */
41 public RedisEncoder() {
42 this(FixedRedisMessagePool.INSTANCE);
43 }
44
45 /**
46 * Creates a new instance.
47 * @param messagePool the predefined message pool.
48 */
49 public RedisEncoder(RedisMessagePool messagePool) {
50 this.messagePool = ObjectUtil.checkNotNull(messagePool, "messagePool");
51 }
52
53 @Override
54 protected void encode(ChannelHandlerContext ctx, RedisMessage msg, List<Object> out) throws Exception {
55 try {
56 writeRedisMessage(ctx.alloc(), msg, out);
57 } catch (CodecException e) {
58 throw e;
59 } catch (Exception e) {
60 throw new CodecException(e);
61 }
62 }
63
64 private void writeRedisMessage(ByteBufAllocator allocator, RedisMessage msg, List<Object> out) {
65 if (msg instanceof InlineCommandRedisMessage) {
66 writeInlineCommandMessage(allocator, (InlineCommandRedisMessage) msg, out);
67 } else if (msg instanceof SimpleStringRedisMessage) {
68 writeSimpleStringMessage(allocator, (SimpleStringRedisMessage) msg, out);
69 } else if (msg instanceof ErrorRedisMessage) {
70 writeErrorMessage(allocator, (ErrorRedisMessage) msg, out);
71 } else if (msg instanceof IntegerRedisMessage) {
72 writeIntegerMessage(allocator, (IntegerRedisMessage) msg, out);
73 } else if (msg instanceof FullBulkStringRedisMessage) {
74 writeFullBulkStringMessage(allocator, (FullBulkStringRedisMessage) msg, out);
75 } else if (msg instanceof BulkStringRedisContent) {
76 writeBulkStringContent(allocator, (BulkStringRedisContent) msg, out);
77 } else if (msg instanceof BulkStringHeaderRedisMessage) {
78 writeBulkStringHeader(allocator, (BulkStringHeaderRedisMessage) msg, out);
79 } else if (msg instanceof ArrayHeaderRedisMessage) {
80 writeArrayHeader(allocator, (ArrayHeaderRedisMessage) msg, out);
81 } else if (msg instanceof ArrayRedisMessage) {
82 writeArrayMessage(allocator, (ArrayRedisMessage) msg, out);
83 } else {
84 throw new CodecException("unknown message type: " + msg);
85 }
86 }
87
88 private static void writeInlineCommandMessage(ByteBufAllocator allocator, InlineCommandRedisMessage msg,
89 List<Object> out) {
90 writeString(allocator, RedisMessageType.INLINE_COMMAND, msg.content(), out);
91 }
92
93 private static void writeSimpleStringMessage(ByteBufAllocator allocator, SimpleStringRedisMessage msg,
94 List<Object> out) {
95 writeString(allocator, RedisMessageType.SIMPLE_STRING, msg.content(), out);
96 }
97
98 private static void writeErrorMessage(ByteBufAllocator allocator, ErrorRedisMessage msg, List<Object> out) {
99 writeString(allocator, RedisMessageType.ERROR, msg.content(), out);
100 }
101
102 private static void writeString(ByteBufAllocator allocator, RedisMessageType type, String content,
103 List<Object> out) {
104 ByteBuf buf = allocator.ioBuffer(type.length() + ByteBufUtil.utf8MaxBytes(content) +
105 RedisConstants.EOL_LENGTH);
106 type.writeTo(buf);
107 ByteBufUtil.writeUtf8(buf, content);
108 buf.writeShort(RedisConstants.EOL_SHORT);
109 out.add(buf);
110 }
111
112 private void writeIntegerMessage(ByteBufAllocator allocator, IntegerRedisMessage msg, List<Object> out) {
113 ByteBuf buf = allocator.ioBuffer(RedisConstants.TYPE_LENGTH + RedisConstants.LONG_MAX_LENGTH +
114 RedisConstants.EOL_LENGTH);
115 RedisMessageType.INTEGER.writeTo(buf);
116 buf.writeBytes(numberToBytes(msg.value()));
117 buf.writeShort(RedisConstants.EOL_SHORT);
118 out.add(buf);
119 }
120
121 private void writeBulkStringHeader(ByteBufAllocator allocator, BulkStringHeaderRedisMessage msg, List<Object> out) {
122 final ByteBuf buf = allocator.ioBuffer(RedisConstants.TYPE_LENGTH +
123 (msg.isNull() ? RedisConstants.NULL_LENGTH :
124 RedisConstants.LONG_MAX_LENGTH + RedisConstants.EOL_LENGTH));
125 RedisMessageType.BULK_STRING.writeTo(buf);
126 if (msg.isNull()) {
127 buf.writeShort(RedisConstants.NULL_SHORT);
128 } else {
129 buf.writeBytes(numberToBytes(msg.bulkStringLength()));
130 buf.writeShort(RedisConstants.EOL_SHORT);
131 }
132 out.add(buf);
133 }
134
135 private static void writeBulkStringContent(ByteBufAllocator allocator, BulkStringRedisContent msg,
136 List<Object> out) {
137 out.add(msg.content().retain());
138 if (msg instanceof LastBulkStringRedisContent) {
139 out.add(allocator.ioBuffer(RedisConstants.EOL_LENGTH).writeShort(RedisConstants.EOL_SHORT));
140 }
141 }
142
143 private void writeFullBulkStringMessage(ByteBufAllocator allocator, FullBulkStringRedisMessage msg,
144 List<Object> out) {
145 if (msg.isNull()) {
146 ByteBuf buf = allocator.ioBuffer(RedisConstants.TYPE_LENGTH + RedisConstants.NULL_LENGTH +
147 RedisConstants.EOL_LENGTH);
148 RedisMessageType.BULK_STRING.writeTo(buf);
149 buf.writeShort(RedisConstants.NULL_SHORT);
150 buf.writeShort(RedisConstants.EOL_SHORT);
151 out.add(buf);
152 } else {
153 ByteBuf headerBuf = allocator.ioBuffer(RedisConstants.TYPE_LENGTH + RedisConstants.LONG_MAX_LENGTH +
154 RedisConstants.EOL_LENGTH);
155 RedisMessageType.BULK_STRING.writeTo(headerBuf);
156 headerBuf.writeBytes(numberToBytes(msg.content().readableBytes()));
157 headerBuf.writeShort(RedisConstants.EOL_SHORT);
158 out.add(headerBuf);
159 out.add(msg.content().retain());
160 out.add(allocator.ioBuffer(RedisConstants.EOL_LENGTH).writeShort(RedisConstants.EOL_SHORT));
161 }
162 }
163
164 /**
165 * Write array header only without body. Use this if you want to write arrays as streaming.
166 */
167 private void writeArrayHeader(ByteBufAllocator allocator, ArrayHeaderRedisMessage msg, List<Object> out) {
168 writeArrayHeader(allocator, msg.isNull(), msg.length(), out);
169 }
170
171 /**
172 * Write full constructed array message.
173 */
174 private void writeArrayMessage(ByteBufAllocator allocator, ArrayRedisMessage msg, List<Object> out) {
175 if (msg.isNull()) {
176 writeArrayHeader(allocator, msg.isNull(), RedisConstants.NULL_VALUE, out);
177 } else {
178 writeArrayHeader(allocator, msg.isNull(), msg.children().size(), out);
179 for (RedisMessage child : msg.children()) {
180 writeRedisMessage(allocator, child, out);
181 }
182 }
183 }
184
185 private void writeArrayHeader(ByteBufAllocator allocator, boolean isNull, long length, List<Object> out) {
186 if (isNull) {
187 final ByteBuf buf = allocator.ioBuffer(RedisConstants.TYPE_LENGTH + RedisConstants.NULL_LENGTH +
188 RedisConstants.EOL_LENGTH);
189 RedisMessageType.ARRAY_HEADER.writeTo(buf);
190 buf.writeShort(RedisConstants.NULL_SHORT);
191 buf.writeShort(RedisConstants.EOL_SHORT);
192 out.add(buf);
193 } else {
194 final ByteBuf buf = allocator.ioBuffer(RedisConstants.TYPE_LENGTH + RedisConstants.LONG_MAX_LENGTH +
195 RedisConstants.EOL_LENGTH);
196 RedisMessageType.ARRAY_HEADER.writeTo(buf);
197 buf.writeBytes(numberToBytes(length));
198 buf.writeShort(RedisConstants.EOL_SHORT);
199 out.add(buf);
200 }
201 }
202
203 private byte[] numberToBytes(long value) {
204 byte[] bytes = messagePool.getByteBufOfInteger(value);
205 return bytes != null ? bytes : RedisCodecUtil.longToAsciiBytes(value);
206 }
207 }
以上源码来自： 即时通讯网 - 即时通讯开发者社区！