1 /*
2 * Copyright 2009 Red Hat, Inc.
3 *
4 * Red Hat licenses this file to you under the Apache License, version 2.0
5 * (the "License"); you may not use this file except in compliance with the
6 * 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 * @author <a href="http://www.jboss.org/netty/">The Netty Project</a>
66 * @author <a href="http://gleamynode.net/">Trustin Lee</a>
67 *
68 * @version $Rev:231 $, $Date:2008-06-12 16:44:50 +0900 (목, 12 6월 2008) $
69 *
70 * @apiviz.landmark
71 */
72 @Sharable
73 public class ProtobufDecoder extends OneToOneDecoder {
74
75 private final MessageLite prototype;
76 private final ExtensionRegistry extensionRegistry;
77
78 /**
79 * Creates a new instance.
80 */
81 public ProtobufDecoder(MessageLite prototype) {
82 this(prototype, null);
83 }
84
85 public ProtobufDecoder(MessageLite prototype, ExtensionRegistry extensionRegistry) {
86 if (prototype == null) {
87 throw new NullPointerException("prototype");
88 }
89 this.prototype = prototype.getDefaultInstanceForType();
90 this.extensionRegistry = extensionRegistry;
91 }
92
93 @Override
94 protected Object decode(
95 ChannelHandlerContext ctx, Channel channel, Object msg) throws Exception {
96 if (!(msg instanceof ChannelBuffer)) {
97 return msg;
98 }
99
100 ChannelBuffer buf = (ChannelBuffer) msg;
101 if (buf.hasArray()) {
102 final int offset = buf.readerIndex();
103 if(extensionRegistry == null) {
104 return prototype.newBuilderForType().mergeFrom(
105 buf.array(), buf.arrayOffset() + offset, buf.readableBytes()).build();
106 } else {
107 return prototype.newBuilderForType().mergeFrom(
108 buf.array(), buf.arrayOffset() + offset, buf.readableBytes(), extensionRegistry).build();
109 }
110 } else {
111 if (extensionRegistry == null) {
112 return prototype.newBuilderForType().mergeFrom(
113 new ChannelBufferInputStream((ChannelBuffer) msg)).build();
114 } else {
115 return prototype.newBuilderForType().mergeFrom(
116 new ChannelBufferInputStream((ChannelBuffer) msg), extensionRegistry).build();
117 }
118 }
119 }
120 }