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 protected abstract void completed() throws IOException;
90
91 protected final void setCharset(final Charset charset) {
92 this.charset = charset != null ? charset : charCodingConfig.getCharset();
93 this.charsetDecoder = null;
94 }
95
96 @Override
97 public final void updateCapacity(final CapacityChannel capacityChannel) throws IOException {
98 capacityChannel.update(capacityIncrement());
99 }
100
101 private void checkResult(final CoderResult result) throws IOException {
102 if (result.isError()) {
103 result.throwException();
104 }
105 }
106
107 private void doDecode(final boolean endOfStream) throws IOException {
108 charBuffer.flip();
109 data(charBuffer, endOfStream);
110 charBuffer.clear();
111 }
112
113 private CharsetDecoder getCharsetDecoder() {
114 CharsetDecoder charsetDecoder = this.charsetDecoder;
115 if (charsetDecoder == null) {
116 Charset charset = this.charset;
117 if (charset == null) {
118 charset = charCodingConfig.getCharset();
119 }
120 if (charset == null) {
121 charset = StandardCharsets.UTF_8;
122 }
123 charsetDecoder = charset.newDecoder();
124 this.charsetDecoder = charsetDecoder;
125 if (charCodingConfig.getMalformedInputAction() != null) {
126 charsetDecoder.onMalformedInput(charCodingConfig.getMalformedInputAction());
127 }
128 if (charCodingConfig.getUnmappableInputAction() != null) {
129 charsetDecoder.onUnmappableCharacter(charCodingConfig.getUnmappableInputAction());
130 }
131 }
132 return charsetDecoder;
133 }
134
135 @Override
136 public final void consume(final ByteBuffer src) throws IOException {
137 final CharsetDecoder charsetDecoder = getCharsetDecoder();
138 while (src.hasRemaining()) {
139 ByteBuffer byteBuffer = this.byteBuffer;
140 if (byteBuffer != null && byteBuffer.position() > 0) {
141
142 final int n = byteBuffer.remaining();
143 if (n < src.remaining()) {
144 final int oldLimit = src.limit();
145 src.limit(src.position() + n);
146 byteBuffer.put(src);
147 src.limit(oldLimit);
148 } else {
149 byteBuffer.put(src);
150 }
151 byteBuffer.flip();
152 final CoderResult r = charsetDecoder.decode(byteBuffer, charBuffer, false);
153 checkResult(r);
154 doDecode(false);
155 byteBuffer.compact();
156 }
157 if (byteBuffer == null || byteBuffer.position() == 0) {
158 final CoderResult r = charsetDecoder.decode(src, charBuffer, false);
159 checkResult(r);
160 doDecode(false);
161 if (r.isUnderflow() && src.hasRemaining()) {
162
163 if (byteBuffer == null) {
164 byteBuffer = ByteBuffer.allocate(Math.max(src.remaining(), 1024));
165 this.byteBuffer = byteBuffer;
166 }
167 byteBuffer.put(src);
168 }
169 }
170 }
171 }
172
173 @Override
174 public final void streamEnd(final List<? extends Header> trailers) throws HttpException, IOException {
175 final CharsetDecoder charsetDecoder = getCharsetDecoder();
176 checkResult(charsetDecoder.decode(EMPTY_BIN, charBuffer, true));
177 doDecode(false);
178 checkResult(charsetDecoder.flush(charBuffer));
179 doDecode(true);
180 completed();
181 }
182
183 }