1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27 package org.apache.hc.core5.http.nio.entity;
28
29 import java.io.IOException;
30 import java.nio.ByteBuffer;
31 import java.nio.CharBuffer;
32 import java.nio.charset.Charset;
33 import java.nio.charset.CharsetDecoder;
34 import java.nio.charset.CoderResult;
35 import java.nio.charset.StandardCharsets;
36 import java.util.List;
37
38 import org.apache.hc.core5.http.Header;
39 import org.apache.hc.core5.http.HttpException;
40 import org.apache.hc.core5.http.config.CharCodingConfig;
41 import org.apache.hc.core5.http.nio.AsyncDataConsumer;
42 import org.apache.hc.core5.http.nio.CapacityChannel;
43 import org.apache.hc.core5.util.Args;
44
45
46
47
48
49
50 public abstract class AbstractCharDataConsumer implements AsyncDataConsumer {
51
52 protected static final int DEF_BUF_SIZE = 8192;
53 private static final ByteBuffer EMPTY_BIN = ByteBuffer.wrap(new byte[0]);
54
55 private final CharBuffer charBuffer;
56 private final CharCodingConfig charCodingConfig;
57
58 private volatile Charset charset;
59 private volatile CharsetDecoder charsetDecoder;
60 private volatile ByteBuffer byteBuffer;
61
62 protected AbstractCharDataConsumer(final int bufSize, final CharCodingConfig charCodingConfig) {
63 this.charBuffer = CharBuffer.allocate(Args.positive(bufSize, "Buffer size"));
64 this.charCodingConfig = charCodingConfig != null ? charCodingConfig : CharCodingConfig.DEFAULT;
65 }
66
67 public AbstractCharDataConsumer() {
68 this(DEF_BUF_SIZE, CharCodingConfig.DEFAULT);
69 }
70
71
72
73
74
75 protected abstract int capacityIncrement();
76
77
78
79
80
81
82
83
84 protected abstract void data(CharBuffer src, boolean endOfStream) throws IOException;
85
86
87
88
89
90
91 protected abstract void completed() throws IOException;
92
93 protected final void setCharset(final Charset charset) {
94 this.charset = charset != null ? charset : charCodingConfig.getCharset();
95 this.charsetDecoder = null;
96 }
97
98 @Override
99 public final void updateCapacity(final CapacityChannel capacityChannel) throws IOException {
100 capacityChannel.update(capacityIncrement());
101 }
102
103 private void checkResult(final CoderResult result) throws IOException {
104 if (result.isError()) {
105 result.throwException();
106 }
107 }
108
109 private void doDecode(final boolean endOfStream) throws IOException {
110 charBuffer.flip();
111 data(charBuffer, endOfStream);
112 charBuffer.clear();
113 }
114
115 private CharsetDecoder getCharsetDecoder() {
116 CharsetDecoder charsetDecoder = this.charsetDecoder;
117 if (charsetDecoder == null) {
118 Charset charset = this.charset;
119 if (charset == null) {
120 charset = charCodingConfig.getCharset();
121 }
122 if (charset == null) {
123 charset = StandardCharsets.UTF_8;
124 }
125 charsetDecoder = charset.newDecoder();
126 this.charsetDecoder = charsetDecoder;
127 if (charCodingConfig.getMalformedInputAction() != null) {
128 charsetDecoder.onMalformedInput(charCodingConfig.getMalformedInputAction());
129 }
130 if (charCodingConfig.getUnmappableInputAction() != null) {
131 charsetDecoder.onUnmappableCharacter(charCodingConfig.getUnmappableInputAction());
132 }
133 }
134 return charsetDecoder;
135 }
136
137 @Override
138 public final void consume(final ByteBuffer src) throws IOException {
139 final CharsetDecoder charsetDecoder = getCharsetDecoder();
140 while (src.hasRemaining()) {
141 ByteBuffer byteBuffer = this.byteBuffer;
142 if (byteBuffer != null && byteBuffer.position() > 0) {
143
144 final int n = byteBuffer.remaining();
145 if (n < src.remaining()) {
146 final int oldLimit = src.limit();
147 src.limit(src.position() + n);
148 byteBuffer.put(src);
149 src.limit(oldLimit);
150 } else {
151 byteBuffer.put(src);
152 }
153 byteBuffer.flip();
154 final CoderResult r = charsetDecoder.decode(byteBuffer, charBuffer, false);
155 checkResult(r);
156 doDecode(false);
157 byteBuffer.compact();
158 }
159 if (byteBuffer == null || byteBuffer.position() == 0) {
160 final CoderResult r = charsetDecoder.decode(src, charBuffer, false);
161 checkResult(r);
162 doDecode(false);
163 if (r.isUnderflow() && src.hasRemaining()) {
164
165 if (byteBuffer == null) {
166 byteBuffer = ByteBuffer.allocate(Math.max(src.remaining(), 1024));
167 this.byteBuffer = byteBuffer;
168 }
169 byteBuffer.put(src);
170 }
171 }
172 }
173 }
174
175 @Override
176 public final void streamEnd(final List<? extends Header> trailers) throws HttpException, IOException {
177 final CharsetDecoder charsetDecoder = getCharsetDecoder();
178 checkResult(charsetDecoder.decode(EMPTY_BIN, charBuffer, true));
179 doDecode(false);
180 checkResult(charsetDecoder.flush(charBuffer));
181 doDecode(true);
182 completed();
183 }
184
185 }