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 if (type.isInline()) {
106
107 if (content.indexOf('\r') != -1 || content.indexOf('\n') != -1) {
108 throw new CodecException("Line breaks are not permitted in 'simple' messages");
109 }
110 }
111 ByteBuf buf = allocator.ioBuffer(type.length() + ByteBufUtil.utf8MaxBytes(content) +
112 RedisConstants.EOL_LENGTH);
113 type.writeTo(buf);
114 ByteBufUtil.writeUtf8(buf, content);
115 buf.writeShort(RedisConstants.EOL_SHORT);
116 out.add(buf);
117 }
118
119 private void writeIntegerMessage(ByteBufAllocator allocator, IntegerRedisMessage msg, List<Object> out) {
120 ByteBuf buf = allocator.ioBuffer(RedisConstants.TYPE_LENGTH + RedisConstants.LONG_MAX_LENGTH +
121 RedisConstants.EOL_LENGTH);
122 RedisMessageType.INTEGER.writeTo(buf);
123 buf.writeBytes(numberToBytes(msg.value()));
124 buf.writeShort(RedisConstants.EOL_SHORT);
125 out.add(buf);
126 }
127
128 private void writeBulkStringHeader(ByteBufAllocator allocator, BulkStringHeaderRedisMessage msg, List<Object> out) {
129 final ByteBuf buf = allocator.ioBuffer(RedisConstants.TYPE_LENGTH +
130 (msg.isNull() ? RedisConstants.NULL_LENGTH :
131 RedisConstants.LONG_MAX_LENGTH + RedisConstants.EOL_LENGTH));
132 RedisMessageType.BULK_STRING.writeTo(buf);
133 if (msg.isNull()) {
134 buf.writeShort(RedisConstants.NULL_SHORT);
135 } else {
136 buf.writeBytes(numberToBytes(msg.bulkStringLength()));
137 buf.writeShort(RedisConstants.EOL_SHORT);
138 }
139 out.add(buf);
140 }
141
142 private static void writeBulkStringContent(ByteBufAllocator allocator, BulkStringRedisContent msg,
143 List<Object> out) {
144 out.add(msg.content().retain());
145 if (msg instanceof LastBulkStringRedisContent) {
146 out.add(allocator.ioBuffer(RedisConstants.EOL_LENGTH).writeShort(RedisConstants.EOL_SHORT));
147 }
148 }
149
150 private void writeFullBulkStringMessage(ByteBufAllocator allocator, FullBulkStringRedisMessage msg,
151 List<Object> out) {
152 if (msg.isNull()) {
153 ByteBuf buf = allocator.ioBuffer(RedisConstants.TYPE_LENGTH + RedisConstants.NULL_LENGTH +
154 RedisConstants.EOL_LENGTH);
155 RedisMessageType.BULK_STRING.writeTo(buf);
156 buf.writeShort(RedisConstants.NULL_SHORT);
157 buf.writeShort(RedisConstants.EOL_SHORT);
158 out.add(buf);
159 } else {
160 ByteBuf headerBuf = allocator.ioBuffer(RedisConstants.TYPE_LENGTH + RedisConstants.LONG_MAX_LENGTH +
161 RedisConstants.EOL_LENGTH);
162 RedisMessageType.BULK_STRING.writeTo(headerBuf);
163 headerBuf.writeBytes(numberToBytes(msg.content().readableBytes()));
164 headerBuf.writeShort(RedisConstants.EOL_SHORT);
165 out.add(headerBuf);
166 out.add(msg.content().retain());
167 out.add(allocator.ioBuffer(RedisConstants.EOL_LENGTH).writeShort(RedisConstants.EOL_SHORT));
168 }
169 }
170
171
172
173
174 private void writeArrayHeader(ByteBufAllocator allocator, ArrayHeaderRedisMessage msg, List<Object> out) {
175 writeArrayHeader(allocator, msg.isNull(), msg.length(), out);
176 }
177
178
179
180
181 private void writeArrayMessage(ByteBufAllocator allocator, ArrayRedisMessage msg, List<Object> out) {
182 if (msg.isNull()) {
183 writeArrayHeader(allocator, msg.isNull(), RedisConstants.NULL_VALUE, out);
184 } else {
185 writeArrayHeader(allocator, msg.isNull(), msg.children().size(), out);
186 for (RedisMessage child : msg.children()) {
187 writeRedisMessage(allocator, child, out);
188 }
189 }
190 }
191
192 private void writeArrayHeader(ByteBufAllocator allocator, boolean isNull, long length, List<Object> out) {
193 if (isNull) {
194 final ByteBuf buf = allocator.ioBuffer(RedisConstants.TYPE_LENGTH + RedisConstants.NULL_LENGTH +
195 RedisConstants.EOL_LENGTH);
196 RedisMessageType.ARRAY_HEADER.writeTo(buf);
197 buf.writeShort(RedisConstants.NULL_SHORT);
198 buf.writeShort(RedisConstants.EOL_SHORT);
199 out.add(buf);
200 } else {
201 final ByteBuf buf = allocator.ioBuffer(RedisConstants.TYPE_LENGTH + RedisConstants.LONG_MAX_LENGTH +
202 RedisConstants.EOL_LENGTH);
203 RedisMessageType.ARRAY_HEADER.writeTo(buf);
204 buf.writeBytes(numberToBytes(length));
205 buf.writeShort(RedisConstants.EOL_SHORT);
206 out.add(buf);
207 }
208 }
209
210 private byte[] numberToBytes(long value) {
211 byte[] bytes = messagePool.getByteBufOfInteger(value);
212 return bytes != null ? bytes : RedisCodecUtil.longToAsciiBytes(value);
213 }
214 }