1 /*
2 * Copyright 2012 The Netty Project
3 *
4 * The Netty Project licenses this file to you under the Apache License,
5 * version 2.0 (the "License"); you may not use this file except in compliance
6 * with the License. You may obtain a 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
11 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13 * License for the specific language governing permissions and limitations
14 * under the License.
15 */
16 package org.jboss.netty.handler.codec.protobuf;
17
18 import org.jboss.netty.buffer.ChannelBuffer;
19 import org.jboss.netty.buffer.ChannelBufferInputStream;
20 import org.jboss.netty.channel.Channel;
21 import org.jboss.netty.channel.ChannelHandler.Sharable;
22 import org.jboss.netty.channel.ChannelHandlerContext;
23 import org.jboss.netty.channel.ChannelPipeline;
24 import org.jboss.netty.channel.MessageEvent;
25 import org.jboss.netty.handler.codec.frame.FrameDecoder;
26 import org.jboss.netty.handler.codec.frame.LengthFieldBasedFrameDecoder;
27 import org.jboss.netty.handler.codec.frame.LengthFieldPrepender;
28 import org.jboss.netty.handler.codec.oneone.OneToOneDecoder;
29
30 import com.google.protobuf.ExtensionRegistry;
31 import com.google.protobuf.Message;
32 import com.google.protobuf.MessageLite;
33
34 /**
35 * Decodes a received {@link ChannelBuffer} into a
36 * <a href="http://code.google.com/p/protobuf/">Google Protocol Buffers</a>
37 * {@link Message} and {@link MessageLite}. Please note that this decoder must
38 * be used with a proper {@link FrameDecoder} such as {@link ProtobufVarint32FrameDecoder}
39 * or {@link LengthFieldBasedFrameDecoder} if you are using a stream-based
40 * transport such as TCP/IP. A typical setup for TCP/IP would be:
41 * <pre>
42 * {@link ChannelPipeline} pipeline = ...;
43 *
44 * // Decoders
45 * pipeline.addLast("frameDecoder",
46 * new {@link LengthFieldBasedFrameDecoder}(1048576, 0, 4, 0, 4));
47 * pipeline.addLast("protobufDecoder",
48 * new {@link ProtobufDecoder}(MyMessage.getDefaultInstance()));
49 *
50 * // Encoder
51 * pipeline.addLast("frameEncoder", new {@link LengthFieldPrepender}(4));
52 * pipeline.addLast("protobufEncoder", new {@link ProtobufEncoder}());
53 * </pre>
54 * and then you can use a {@code MyMessage} instead of a {@link ChannelBuffer}
55 * as a message:
56 * <pre>
57 * void messageReceived({@link ChannelHandlerContext} ctx, {@link MessageEvent} e) {
58 * MyMessage req = (MyMessage) e.getMessage();
59 * MyMessage res = MyMessage.newBuilder().setText(
60 * "Did you say '" + req.getText() + "'?").build();
61 * ch.write(res);
62 * }
63 * </pre>
64 *
65 * @apiviz.landmark
66 */
67 @Sharable
68 public class ProtobufDecoder extends OneToOneDecoder {
69
70 private final MessageLite prototype;
71 private final ExtensionRegistry extensionRegistry;
72
73 /**
74 * Creates a new instance.
75 */
76 public ProtobufDecoder(MessageLite prototype) {
77 this(prototype, null);
78 }
79
80 public ProtobufDecoder(MessageLite prototype, ExtensionRegistry extensionRegistry) {
81 if (prototype == null) {
82 throw new NullPointerException("prototype");
83 }
84 this.prototype = prototype.getDefaultInstanceForType();
85 this.extensionRegistry = extensionRegistry;
86 }
87
88 @Override
89 protected Object decode(
90 ChannelHandlerContext ctx, Channel channel, Object msg) throws Exception {
91 if (!(msg instanceof ChannelBuffer)) {
92 return msg;
93 }
94
95 ChannelBuffer buf = (ChannelBuffer) msg;
96 if (buf.hasArray()) {
97 final int offset = buf.readerIndex();
98 if (extensionRegistry == null) {
99 return prototype.newBuilderForType().mergeFrom(
100 buf.array(), buf.arrayOffset() + offset, buf.readableBytes()).build();
101 } else {
102 return prototype.newBuilderForType().mergeFrom(
103 buf.array(), buf.arrayOffset() + offset, buf.readableBytes(), extensionRegistry).build();
104 }
105 } else {
106 if (extensionRegistry == null) {
107 return prototype.newBuilderForType().mergeFrom(
108 new ChannelBufferInputStream((ChannelBuffer) msg)).build();
109 } else {
110 return prototype.newBuilderForType().mergeFrom(
111 new ChannelBufferInputStream((ChannelBuffer) msg), extensionRegistry).build();
112 }
113 }
114 }
115 }