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.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
31
32
33 @UnstableApi
34 public class RedisEncoder extends MessageToMessageEncoder<RedisMessage> {
35
36 private final RedisMessagePool messagePool;
37
38
39
40
41 public RedisEncoder() {
42 this(FixedRedisMessagePool.INSTANCE);
43 }
44
45
46
47
48
49 public RedisEncoder(RedisMessagePool messagePool) {
50 super(RedisMessage.class);
51 this.messagePool = ObjectUtil.checkNotNull(messagePool, "messagePool");
52 }
53
54 @Override
55 protected void encode(ChannelHandlerContext ctx, RedisMessage msg, List<Object> out) throws Exception {
56 try {
57 writeRedisMessage(ctx.alloc(), msg, out);
58 } catch (CodecException e) {
59 throw e;
60 } catch (Exception e) {
61 throw new CodecException(e);
62 }
63 }
64
65 private void writeRedisMessage(ByteBufAllocator allocator, RedisMessage msg, List<Object> out) {
66 if (msg instanceof InlineCommandRedisMessage) {
67 writeInlineCommandMessage(allocator, (InlineCommandRedisMessage) msg, out);
68 } else if (msg instanceof SimpleStringRedisMessage) {
69 writeSimpleStringMessage(allocator, (SimpleStringRedisMessage) msg, out);
70 } else if (msg instanceof ErrorRedisMessage) {
71 writeErrorMessage(allocator, (ErrorRedisMessage) msg, out);
72 } else if (msg instanceof IntegerRedisMessage) {
73 writeIntegerMessage(allocator, (IntegerRedisMessage) msg, out);
74 } else if (msg instanceof FullBulkStringRedisMessage) {
75 writeFullBulkStringMessage(allocator, (FullBulkStringRedisMessage) msg, out);
76 } else if (msg instanceof BulkStringRedisContent) {
77 writeBulkStringContent(allocator, (BulkStringRedisContent) msg, out);
78 } else if (msg instanceof BulkStringHeaderRedisMessage) {
79 writeBulkStringHeader(allocator, (BulkStringHeaderRedisMessage) msg, out);
80 } else if (msg instanceof ArrayHeaderRedisMessage) {
81 writeArrayHeader(allocator, (ArrayHeaderRedisMessage) msg, out);
82 } else if (msg instanceof ArrayRedisMessage) {
83 writeArrayMessage(allocator, (ArrayRedisMessage) msg, out);
84 } else {
85 throw new CodecException("unknown message type: " + msg);
86 }
87 }
88
89 private static void writeInlineCommandMessage(ByteBufAllocator allocator, InlineCommandRedisMessage msg,
90 List<Object> out) {
91 writeString(allocator, RedisMessageType.INLINE_COMMAND, msg.content(), out);
92 }
93
94 private static void writeSimpleStringMessage(ByteBufAllocator allocator, SimpleStringRedisMessage msg,
95 List<Object> out) {
96 writeString(allocator, RedisMessageType.SIMPLE_STRING, msg.content(), out);
97 }
98
99 private static void writeErrorMessage(ByteBufAllocator allocator, ErrorRedisMessage msg, List<Object> out) {
100 writeString(allocator, RedisMessageType.ERROR, msg.content(), out);
101 }
102
103 private static void writeString(ByteBufAllocator allocator, RedisMessageType type, String content,
104 List<Object> out) {
105 ByteBuf buf = allocator.ioBuffer(type.length() + ByteBufUtil.utf8MaxBytes(content) +
106 RedisConstants.EOL_LENGTH);
107 type.writeTo(buf);
108 ByteBufUtil.writeUtf8(buf, content);
109 buf.writeShort(RedisConstants.EOL_SHORT);
110 out.add(buf);
111 }
112
113 private void writeIntegerMessage(ByteBufAllocator allocator, IntegerRedisMessage msg, List<Object> out) {
114 ByteBuf buf = allocator.ioBuffer(RedisConstants.TYPE_LENGTH + RedisConstants.LONG_MAX_LENGTH +
115 RedisConstants.EOL_LENGTH);
116 RedisMessageType.INTEGER.writeTo(buf);
117 buf.writeBytes(numberToBytes(msg.value()));
118 buf.writeShort(RedisConstants.EOL_SHORT);
119 out.add(buf);
120 }
121
122 private void writeBulkStringHeader(ByteBufAllocator allocator, BulkStringHeaderRedisMessage msg, List<Object> out) {
123 final ByteBuf buf = allocator.ioBuffer(RedisConstants.TYPE_LENGTH +
124 (msg.isNull() ? RedisConstants.NULL_LENGTH :
125 RedisConstants.LONG_MAX_LENGTH + RedisConstants.EOL_LENGTH));
126 RedisMessageType.BULK_STRING.writeTo(buf);
127 if (msg.isNull()) {
128 buf.writeShort(RedisConstants.NULL_SHORT);
129 } else {
130 buf.writeBytes(numberToBytes(msg.bulkStringLength()));
131 buf.writeShort(RedisConstants.EOL_SHORT);
132 }
133 out.add(buf);
134 }
135
136 private static void writeBulkStringContent(ByteBufAllocator allocator, BulkStringRedisContent msg,
137 List<Object> out) {
138 out.add(msg.content().retain());
139 if (msg instanceof LastBulkStringRedisContent) {
140 out.add(allocator.ioBuffer(RedisConstants.EOL_LENGTH).writeShort(RedisConstants.EOL_SHORT));
141 }
142 }
143
144 private void writeFullBulkStringMessage(ByteBufAllocator allocator, FullBulkStringRedisMessage msg,
145 List<Object> out) {
146 if (msg.isNull()) {
147 ByteBuf buf = allocator.ioBuffer(RedisConstants.TYPE_LENGTH + RedisConstants.NULL_LENGTH +
148 RedisConstants.EOL_LENGTH);
149 RedisMessageType.BULK_STRING.writeTo(buf);
150 buf.writeShort(RedisConstants.NULL_SHORT);
151 buf.writeShort(RedisConstants.EOL_SHORT);
152 out.add(buf);
153 } else {
154 ByteBuf headerBuf = allocator.ioBuffer(RedisConstants.TYPE_LENGTH + RedisConstants.LONG_MAX_LENGTH +
155 RedisConstants.EOL_LENGTH);
156 RedisMessageType.BULK_STRING.writeTo(headerBuf);
157 headerBuf.writeBytes(numberToBytes(msg.content().readableBytes()));
158 headerBuf.writeShort(RedisConstants.EOL_SHORT);
159 out.add(headerBuf);
160 out.add(msg.content().retain());
161 out.add(allocator.ioBuffer(RedisConstants.EOL_LENGTH).writeShort(RedisConstants.EOL_SHORT));
162 }
163 }
164
165
166
167
168 private void writeArrayHeader(ByteBufAllocator allocator, ArrayHeaderRedisMessage msg, List<Object> out) {
169 writeArrayHeader(allocator, msg.isNull(), msg.length(), out);
170 }
171
172
173
174
175 private void writeArrayMessage(ByteBufAllocator allocator, ArrayRedisMessage msg, List<Object> out) {
176 if (msg.isNull()) {
177 writeArrayHeader(allocator, msg.isNull(), RedisConstants.NULL_VALUE, out);
178 } else {
179 writeArrayHeader(allocator, msg.isNull(), msg.children().size(), out);
180 for (RedisMessage child : msg.children()) {
181 writeRedisMessage(allocator, child, out);
182 }
183 }
184 }
185
186 private void writeArrayHeader(ByteBufAllocator allocator, boolean isNull, long length, List<Object> out) {
187 if (isNull) {
188 final ByteBuf buf = allocator.ioBuffer(RedisConstants.TYPE_LENGTH + RedisConstants.NULL_LENGTH +
189 RedisConstants.EOL_LENGTH);
190 RedisMessageType.ARRAY_HEADER.writeTo(buf);
191 buf.writeShort(RedisConstants.NULL_SHORT);
192 buf.writeShort(RedisConstants.EOL_SHORT);
193 out.add(buf);
194 } else {
195 final ByteBuf buf = allocator.ioBuffer(RedisConstants.TYPE_LENGTH + RedisConstants.LONG_MAX_LENGTH +
196 RedisConstants.EOL_LENGTH);
197 RedisMessageType.ARRAY_HEADER.writeTo(buf);
198 buf.writeBytes(numberToBytes(length));
199 buf.writeShort(RedisConstants.EOL_SHORT);
200 out.add(buf);
201 }
202 }
203
204 private byte[] numberToBytes(long value) {
205 byte[] bytes = messagePool.getByteBufOfInteger(value);
206 return bytes != null ? bytes : RedisCodecUtil.longToAsciiBytes(value);
207 }
208 }