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    * http://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="http://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 SimpleStringRedisMessage) {
66              writeSimpleStringMessage(allocator, (SimpleStringRedisMessage) msg, out);
67          } else if (msg instanceof ErrorRedisMessage) {
68              writeErrorMessage(allocator, (ErrorRedisMessage) msg, out);
69          } else if (msg instanceof IntegerRedisMessage) {
70              writeIntegerMessage(allocator, (IntegerRedisMessage) msg, out);
71          } else if (msg instanceof FullBulkStringRedisMessage) {
72              writeFullBulkStringMessage(allocator, (FullBulkStringRedisMessage) msg, out);
73          } else if (msg instanceof BulkStringRedisContent) {
74              writeBulkStringContent(allocator, (BulkStringRedisContent) msg, out);
75          } else if (msg instanceof BulkStringHeaderRedisMessage) {
76              writeBulkStringHeader(allocator, (BulkStringHeaderRedisMessage) msg, out);
77          } else if (msg instanceof ArrayHeaderRedisMessage) {
78              writeArrayHeader(allocator, (ArrayHeaderRedisMessage) msg, out);
79          } else if (msg instanceof ArrayRedisMessage) {
80              writeArrayMessage(allocator, (ArrayRedisMessage) msg, out);
81          } else {
82              throw new CodecException("unknown message type: " + msg);
83          }
84      }
85  
86      private static void writeSimpleStringMessage(ByteBufAllocator allocator, SimpleStringRedisMessage msg,
87                                                   List<Object> out) {
88          writeString(allocator, RedisMessageType.SIMPLE_STRING.value(), msg.content(), out);
89      }
90  
91      private static void writeErrorMessage(ByteBufAllocator allocator, ErrorRedisMessage msg, List<Object> out) {
92          writeString(allocator, RedisMessageType.ERROR.value(), msg.content(), out);
93      }
94  
95      private static void writeString(ByteBufAllocator allocator, byte type, String content, List<Object> out) {
96          ByteBuf buf = allocator.ioBuffer(RedisConstants.TYPE_LENGTH + ByteBufUtil.utf8MaxBytes(content) +
97                                           RedisConstants.EOL_LENGTH);
98          buf.writeByte(type);
99          ByteBufUtil.writeUtf8(buf, content);
100         buf.writeShort(RedisConstants.EOL_SHORT);
101         out.add(buf);
102     }
103 
104     private void writeIntegerMessage(ByteBufAllocator allocator, IntegerRedisMessage msg, List<Object> out) {
105         ByteBuf buf = allocator.ioBuffer(RedisConstants.TYPE_LENGTH + RedisConstants.LONG_MAX_LENGTH +
106                                          RedisConstants.EOL_LENGTH);
107         buf.writeByte(RedisMessageType.INTEGER.value());
108         buf.writeBytes(numberToBytes(msg.value()));
109         buf.writeShort(RedisConstants.EOL_SHORT);
110         out.add(buf);
111     }
112 
113     private void writeBulkStringHeader(ByteBufAllocator allocator, BulkStringHeaderRedisMessage msg, List<Object> out) {
114         final ByteBuf buf = allocator.ioBuffer(RedisConstants.TYPE_LENGTH +
115                                         (msg.isNull() ? RedisConstants.NULL_LENGTH :
116                                                         RedisConstants.LONG_MAX_LENGTH + RedisConstants.EOL_LENGTH));
117         buf.writeByte(RedisMessageType.BULK_STRING.value());
118         if (msg.isNull()) {
119             buf.writeShort(RedisConstants.NULL_SHORT);
120         } else {
121             buf.writeBytes(numberToBytes(msg.bulkStringLength()));
122             buf.writeShort(RedisConstants.EOL_SHORT);
123         }
124         out.add(buf);
125     }
126 
127     private static void writeBulkStringContent(ByteBufAllocator allocator, BulkStringRedisContent msg,
128                                                List<Object> out) {
129         out.add(msg.content().retain());
130         if (msg instanceof LastBulkStringRedisContent) {
131             out.add(allocator.ioBuffer(RedisConstants.EOL_LENGTH).writeShort(RedisConstants.EOL_SHORT));
132         }
133     }
134 
135     private void writeFullBulkStringMessage(ByteBufAllocator allocator, FullBulkStringRedisMessage msg,
136                                             List<Object> out) {
137         if (msg.isNull()) {
138             ByteBuf buf = allocator.ioBuffer(RedisConstants.TYPE_LENGTH + RedisConstants.NULL_LENGTH +
139                                              RedisConstants.EOL_LENGTH);
140             buf.writeByte(RedisMessageType.BULK_STRING.value());
141             buf.writeShort(RedisConstants.NULL_SHORT);
142             buf.writeShort(RedisConstants.EOL_SHORT);
143             out.add(buf);
144         } else {
145             ByteBuf headerBuf = allocator.ioBuffer(RedisConstants.TYPE_LENGTH + RedisConstants.LONG_MAX_LENGTH +
146                                                    RedisConstants.EOL_LENGTH);
147             headerBuf.writeByte(RedisMessageType.BULK_STRING.value());
148             headerBuf.writeBytes(numberToBytes(msg.content().readableBytes()));
149             headerBuf.writeShort(RedisConstants.EOL_SHORT);
150             out.add(headerBuf);
151             out.add(msg.content().retain());
152             out.add(allocator.ioBuffer(RedisConstants.EOL_LENGTH).writeShort(RedisConstants.EOL_SHORT));
153         }
154     }
155 
156     /**
157      * Write array header only without body. Use this if you want to write arrays as streaming.
158      */
159     private void writeArrayHeader(ByteBufAllocator allocator, ArrayHeaderRedisMessage msg, List<Object> out) {
160         writeArrayHeader(allocator, msg.isNull(), msg.length(), out);
161     }
162 
163     /**
164      * Write full constructed array message.
165      */
166     private void writeArrayMessage(ByteBufAllocator allocator, ArrayRedisMessage msg, List<Object> out) {
167         if (msg.isNull()) {
168             writeArrayHeader(allocator, msg.isNull(), RedisConstants.NULL_VALUE, out);
169         } else {
170             writeArrayHeader(allocator, msg.isNull(), msg.children().size(), out);
171             for (RedisMessage child : msg.children()) {
172                 writeRedisMessage(allocator, child, out);
173             }
174         }
175     }
176 
177     private void writeArrayHeader(ByteBufAllocator allocator, boolean isNull, long length, List<Object> out) {
178         if (isNull) {
179             final ByteBuf buf = allocator.ioBuffer(RedisConstants.TYPE_LENGTH + RedisConstants.NULL_LENGTH +
180                                                    RedisConstants.EOL_LENGTH);
181             buf.writeByte(RedisMessageType.ARRAY_HEADER.value());
182             buf.writeShort(RedisConstants.NULL_SHORT);
183             buf.writeShort(RedisConstants.EOL_SHORT);
184             out.add(buf);
185         } else {
186             final ByteBuf buf = allocator.ioBuffer(RedisConstants.TYPE_LENGTH + RedisConstants.LONG_MAX_LENGTH +
187                                                    RedisConstants.EOL_LENGTH);
188             buf.writeByte(RedisMessageType.ARRAY_HEADER.value());
189             buf.writeBytes(numberToBytes(length));
190             buf.writeShort(RedisConstants.EOL_SHORT);
191             out.add(buf);
192         }
193     }
194 
195     private byte[] numberToBytes(long value) {
196         byte[] bytes = messagePool.getByteBufOfInteger(value);
197         return bytes != null ? bytes : RedisCodecUtil.longToAsciiBytes(value);
198     }
199 }