View Javadoc
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          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      * Write array header only without body. Use this if you want to write arrays as streaming.
167      */
168     private void writeArrayHeader(ByteBufAllocator allocator, ArrayHeaderRedisMessage msg, List<Object> out) {
169         writeArrayHeader(allocator, msg.isNull(), msg.length(), out);
170     }
171 
172     /**
173      * Write full constructed array message.
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 }