View Javadoc
1   /*
2    * ====================================================================
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *   http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing,
14   * software distributed under the License is distributed on an
15   * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
16   * KIND, either express or implied.  See the License for the
17   * specific language governing permissions and limitations
18   * under the License.
19   * ====================================================================
20   *
21   * This software consists of voluntary contributions made by many
22   * individuals on behalf of the Apache Software Foundation.  For more
23   * information on the Apache Software Foundation, please see
24   * <http://www.apache.org/>.
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   * Abstract text data consumer.
47   *
48   * @since 5.0
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       * Triggered to obtain the capacity increment.
72       *
73       * @return the number of bytes this consumer is prepared to process.
74       */
75      protected abstract int capacityIncrement();
76  
77      /**
78       * Triggered to pass incoming data packet to the data consumer.
79       *
80       * @param src the data packet.
81       * @param endOfStream flag indicating whether this data packet is the last in the data stream.
82       * @throws IOException in case of an I/O error.
83       */
84      protected abstract void data(CharBuffer src, boolean endOfStream) throws IOException;
85  
86      /**
87       * Triggered to signal completion of data processing.
88       *
89       * @throws IOException in case of an I/O error.
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                 // There are some left-overs from the previous input operation
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                     // in case of input underflow src can be expected to be very small (one incomplete UTF8 char)
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 }