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.compression;
17
18 import org.jboss.netty.buffer.ChannelBuffer;
19 import org.jboss.netty.buffer.ChannelBuffers;
20 import org.jboss.netty.channel.Channel;
21 import org.jboss.netty.channel.ChannelHandlerContext;
22 import org.jboss.netty.handler.codec.oneone.OneToOneDecoder;
23 import org.jboss.netty.util.internal.jzlib.JZlib;
24 import org.jboss.netty.util.internal.jzlib.ZStream;
25
26
27 /**
28 * Decompresses a {@link ChannelBuffer} using the deflate algorithm.
29 * @apiviz.landmark
30 * @apiviz.has org.jboss.netty.handler.codec.compression.ZlibWrapper
31 */
32 public class ZlibDecoder extends OneToOneDecoder {
33
34 private final ZStream z = new ZStream();
35 private byte[] dictionary;
36 private volatile boolean finished;
37
38 /**
39 * Creates a new instance with the default wrapper ({@link ZlibWrapper#ZLIB}).
40 *
41 * @throws CompressionException if failed to initialize zlib
42 */
43 public ZlibDecoder() {
44 this(ZlibWrapper.ZLIB);
45 }
46
47 /**
48 * Creates a new instance with the specified wrapper.
49 *
50 * @throws CompressionException if failed to initialize zlib
51 */
52 public ZlibDecoder(ZlibWrapper wrapper) {
53 if (wrapper == null) {
54 throw new NullPointerException("wrapper");
55 }
56
57 synchronized (z) {
58 int resultCode = z.inflateInit(ZlibUtil.convertWrapperType(wrapper));
59 if (resultCode != JZlib.Z_OK) {
60 ZlibUtil.fail(z, "initialization failure", resultCode);
61 }
62 }
63 }
64
65 /**
66 * Creates a new instance with the specified preset dictionary. The wrapper
67 * is always {@link ZlibWrapper#ZLIB} because it is the only format that
68 * supports the preset dictionary.
69 *
70 * @throws CompressionException if failed to initialize zlib
71 */
72 public ZlibDecoder(byte[] dictionary) {
73 if (dictionary == null) {
74 throw new NullPointerException("dictionary");
75 }
76 this.dictionary = dictionary;
77
78 synchronized (z) {
79 int resultCode;
80 resultCode = z.inflateInit(JZlib.W_ZLIB);
81 if (resultCode != JZlib.Z_OK) {
82 ZlibUtil.fail(z, "initialization failure", resultCode);
83 }
84 }
85 }
86
87 /**
88 * Returns {@code true} if and only if the end of the compressed stream
89 * has been reached.
90 */
91 public boolean isClosed() {
92 return finished;
93 }
94
95 @Override
96 protected Object decode(ChannelHandlerContext ctx, Channel channel, Object msg) throws Exception {
97 if (!(msg instanceof ChannelBuffer) || finished) {
98 return msg;
99 }
100
101 synchronized (z) {
102 try {
103 // Configure input.
104 ChannelBuffer compressed = (ChannelBuffer) msg;
105 byte[] in = new byte[compressed.readableBytes()];
106 compressed.readBytes(in);
107 z.next_in = in;
108 z.next_in_index = 0;
109 z.avail_in = in.length;
110
111 // Configure output.
112 byte[] out = new byte[in.length << 1];
113 ChannelBuffer decompressed = ChannelBuffers.dynamicBuffer(
114 compressed.order(), out.length,
115 ctx.getChannel().getConfig().getBufferFactory());
116 z.next_out = out;
117 z.next_out_index = 0;
118 z.avail_out = out.length;
119
120 loop: for (;;) {
121 // Decompress 'in' into 'out'
122 int resultCode = z.inflate(JZlib.Z_SYNC_FLUSH);
123 if (z.next_out_index > 0) {
124 decompressed.writeBytes(out, 0, z.next_out_index);
125 z.avail_out = out.length;
126 }
127 z.next_out_index = 0;
128
129 switch (resultCode) {
130 case JZlib.Z_NEED_DICT:
131 if (dictionary == null) {
132 ZlibUtil.fail(z, "decompression failure", resultCode);
133 } else {
134 resultCode = z.inflateSetDictionary(dictionary, dictionary.length);
135 if (resultCode != JZlib.Z_OK) {
136 ZlibUtil.fail(z, "failed to set the dictionary", resultCode);
137 }
138 }
139 break;
140 case JZlib.Z_STREAM_END:
141 finished = true; // Do not decode anymore.
142 z.inflateEnd();
143 break loop;
144 case JZlib.Z_OK:
145 break;
146 case JZlib.Z_BUF_ERROR:
147 if (z.avail_in <= 0) {
148 break loop;
149 }
150 break;
151 default:
152 ZlibUtil.fail(z, "decompression failure", resultCode);
153 }
154 }
155
156 if (decompressed.writerIndex() != 0) { // readerIndex is always 0
157 return decompressed;
158 } else {
159 return null;
160 }
161 } finally {
162 // Deference the external references explicitly to tell the VM that
163 // the allocated byte arrays are temporary so that the call stack
164 // can be utilized.
165 // I'm not sure if the modern VMs do this optimization though.
166 z.next_in = null;
167 z.next_out = null;
168 }
169 }
170 }
171 }
以上源码来自： 即时通讯网 - 即时通讯开发者社区！