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 * https://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 io.netty.handler.codec;
17
18 import io.netty.buffer.ByteBuf;
19 import io.netty.buffer.Unpooled;
20 import io.netty.channel.ChannelHandlerContext;
21 import io.netty.channel.ChannelOutboundHandler;
22 import io.netty.channel.ChannelOutboundHandlerAdapter;
23 import io.netty.channel.ChannelPipeline;
24 import io.netty.channel.ChannelPromise;
25 import io.netty.util.ReferenceCountUtil;
26 import io.netty.util.internal.TypeParameterMatcher;
27
28
29 /**
30 * {@link ChannelOutboundHandlerAdapter} which encodes message in a stream-like fashion from one message to an
31 * {@link ByteBuf}.
32 *
33 *
34 * Example implementation which encodes {@link Integer}s to a {@link ByteBuf}.
35 *
36 * <pre>
37 * public class IntegerEncoder extends {@link MessageToByteEncoder}<{@link Integer}> {
38 * {@code @Override}
39 * public void encode({@link ChannelHandlerContext} ctx, {@link Integer} msg, {@link ByteBuf} out)
40 * throws {@link Exception} {
41 * out.writeInt(msg);
42 * }
43 * }
44 * </pre>
45 */
46 public abstract class MessageToByteEncoder<I> extends ChannelOutboundHandlerAdapter {
47
48 private final TypeParameterMatcher matcher;
49 private final boolean preferDirect;
50
51 /**
52 * see {@link #MessageToByteEncoder(boolean)} with {@code true} as boolean parameter.
53 */
54 protected MessageToByteEncoder() {
55 this(true);
56 }
57
58 /**
59 * see {@link #MessageToByteEncoder(Class, boolean)} with {@code true} as boolean value.
60 */
61 protected MessageToByteEncoder(Class<? extends I> outboundMessageType) {
62 this(outboundMessageType, true);
63 }
64
65 /**
66 * Create a new instance which will try to detect the types to match out of the type parameter of the class.
67 *
68 * @param preferDirect {@code true} if a direct {@link ByteBuf} should be tried to be used as target for
69 * the encoded messages. If {@code false} is used it will allocate a heap
70 * {@link ByteBuf}, which is backed by an byte array.
71 */
72 protected MessageToByteEncoder(boolean preferDirect) {
73 matcher = TypeParameterMatcher.find(this, MessageToByteEncoder.class, "I");
74 this.preferDirect = preferDirect;
75 }
76
77 /**
78 * Create a new instance
79 *
80 * @param outboundMessageType The type of messages to match
81 * @param preferDirect {@code true} if a direct {@link ByteBuf} should be tried to be used as target for
82 * the encoded messages. If {@code false} is used it will allocate a heap
83 * {@link ByteBuf}, which is backed by an byte array.
84 */
85 protected MessageToByteEncoder(Class<? extends I> outboundMessageType, boolean preferDirect) {
86 matcher = TypeParameterMatcher.get(outboundMessageType);
87 this.preferDirect = preferDirect;
88 }
89
90 /**
91 * Returns {@code true} if the given message should be handled. If {@code false} it will be passed to the next
92 * {@link ChannelOutboundHandler} in the {@link ChannelPipeline}.
93 */
94 public boolean acceptOutboundMessage(Object msg) throws Exception {
95 return matcher.match(msg);
96 }
97
98 @Override
99 public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
100 ByteBuf buf = null;
101 try {
102 if (acceptOutboundMessage(msg)) {
103 @SuppressWarnings("unchecked")
104 I cast = (I) msg;
105 buf = allocateBuffer(ctx, cast, preferDirect);
106 try {
107 encode(ctx, cast, buf);
108 } finally {
109 ReferenceCountUtil.release(cast);
110 }
111
112 if (buf.isReadable()) {
113 ctx.write(buf, promise);
114 } else {
115 buf.release();
116 ctx.write(Unpooled.EMPTY_BUFFER, promise);
117 }
118 buf = null;
119 } else {
120 ctx.write(msg, promise);
121 }
122 } catch (EncoderException e) {
123 throw e;
124 } catch (Throwable e) {
125 throw new EncoderException(e);
126 } finally {
127 if (buf != null) {
128 buf.release();
129 }
130 }
131 }
132
133 /**
134 * Allocate a {@link ByteBuf} which will be used as argument of {@link #encode(ChannelHandlerContext, I, ByteBuf)}.
135 * Sub-classes may override this method to return {@link ByteBuf} with a perfect matching {@code initialCapacity}.
136 */
137 protected ByteBuf allocateBuffer(ChannelHandlerContext ctx, @SuppressWarnings("unused") I msg,
138 boolean preferDirect) throws Exception {
139 if (preferDirect) {
140 return ctx.alloc().ioBuffer();
141 } else {
142 return ctx.alloc().heapBuffer();
143 }
144 }
145
146 /**
147 * Encode a message into a {@link ByteBuf}. This method will be called for each written message that can be handled
148 * by this encoder.
149 *
150 * @param ctx the {@link ChannelHandlerContext} which this {@link MessageToByteEncoder} belongs to
151 * @param msg the message to encode
152 * @param out the {@link ByteBuf} into which the encoded message will be written
153 * @throws Exception is thrown if an error occurs
154 */
155 protected abstract void encode(ChannelHandlerContext ctx, I msg, ByteBuf out) throws Exception;
156
157 protected boolean isPreferDirect() {
158 return preferDirect;
159 }
160 }
以上源码来自： 即时通讯网 - 即时通讯开发者社区！