001/*
002 * ====================================================================
003 * Licensed to the Apache Software Foundation (ASF) under one
004 * or more contributor license agreements.  See the NOTICE file
005 * distributed with this work for additional information
006 * regarding copyright ownership.  The ASF licenses this file
007 * to you under the Apache License, Version 2.0 (the
008 * "License"); you may not use this file except in compliance
009 * with the License.  You may obtain a copy of the License at
010 *
011 *   http://www.apache.org/licenses/LICENSE-2.0
012 *
013 * Unless required by applicable law or agreed to in writing,
014 * software distributed under the License is distributed on an
015 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
016 * KIND, either express or implied.  See the License for the
017 * specific language governing permissions and limitations
018 * under the License.
019 * ====================================================================
020 *
021 * This software consists of voluntary contributions made by many
022 * individuals on behalf of the Apache Software Foundation.  For more
023 * information on the Apache Software Foundation, please see
024 * <http://www.apache.org/>.
025 *
026 */
027
028package org.apache.http.impl;
029
030import java.io.IOException;
031import java.io.InputStream;
032import java.io.OutputStream;
033import java.net.InetAddress;
034import java.net.Socket;
035import java.net.SocketAddress;
036import java.net.SocketException;
037import java.net.SocketTimeoutException;
038import java.nio.charset.CharsetDecoder;
039import java.nio.charset.CharsetEncoder;
040import java.util.concurrent.atomic.AtomicReference;
041
042import org.apache.http.ConnectionClosedException;
043import org.apache.http.Header;
044import org.apache.http.HttpConnection;
045import org.apache.http.HttpConnectionMetrics;
046import org.apache.http.HttpEntity;
047import org.apache.http.HttpException;
048import org.apache.http.HttpInetConnection;
049import org.apache.http.HttpMessage;
050import org.apache.http.config.MessageConstraints;
051import org.apache.http.entity.BasicHttpEntity;
052import org.apache.http.entity.ContentLengthStrategy;
053import org.apache.http.impl.entity.LaxContentLengthStrategy;
054import org.apache.http.impl.entity.StrictContentLengthStrategy;
055import org.apache.http.impl.io.ChunkedInputStream;
056import org.apache.http.impl.io.ChunkedOutputStream;
057import org.apache.http.impl.io.ContentLengthInputStream;
058import org.apache.http.impl.io.ContentLengthOutputStream;
059import org.apache.http.impl.io.EmptyInputStream;
060import org.apache.http.impl.io.HttpTransportMetricsImpl;
061import org.apache.http.impl.io.IdentityInputStream;
062import org.apache.http.impl.io.IdentityOutputStream;
063import org.apache.http.impl.io.SessionInputBufferImpl;
064import org.apache.http.impl.io.SessionOutputBufferImpl;
065import org.apache.http.io.SessionInputBuffer;
066import org.apache.http.io.SessionOutputBuffer;
067import org.apache.http.protocol.HTTP;
068import org.apache.http.util.Args;
069import org.apache.http.util.NetUtils;
070
071/**
072 * This class serves as a base for all {@link HttpConnection} implementations and provides
073 * functionality common to both client and server HTTP connections.
074 *
075 * @since 4.0
076 */
077public class BHttpConnectionBase implements HttpConnection, HttpInetConnection {
078
079    private final SessionInputBufferImpl inbuffer;
080    private final SessionOutputBufferImpl outbuffer;
081    private final MessageConstraints messageConstraints;
082    private final HttpConnectionMetricsImpl connMetrics;
083    private final ContentLengthStrategy incomingContentStrategy;
084    private final ContentLengthStrategy outgoingContentStrategy;
085    private final AtomicReference<Socket> socketHolder;
086
087    /**
088     * Creates new instance of BHttpConnectionBase.
089     *
090     * @param buffersize buffer size. Must be a positive number.
091     * @param fragmentSizeHint fragment size hint.
092     * @param chardecoder decoder to be used for decoding HTTP protocol elements.
093     *   If {@code null} simple type cast will be used for byte to char conversion.
094     * @param charencoder encoder to be used for encoding HTTP protocol elements.
095     *   If {@code null} simple type cast will be used for char to byte conversion.
096     * @param messageConstraints Message constraints. If {@code null}
097     *   {@link MessageConstraints#DEFAULT} will be used.
098     * @param incomingContentStrategy incoming content length strategy. If {@code null}
099     *   {@link LaxContentLengthStrategy#INSTANCE} will be used.
100     * @param outgoingContentStrategy outgoing content length strategy. If {@code null}
101     *   {@link StrictContentLengthStrategy#INSTANCE} will be used.
102     */
103    protected BHttpConnectionBase(
104            final int buffersize,
105            final int fragmentSizeHint,
106            final CharsetDecoder chardecoder,
107            final CharsetEncoder charencoder,
108            final MessageConstraints messageConstraints,
109            final ContentLengthStrategy incomingContentStrategy,
110            final ContentLengthStrategy outgoingContentStrategy) {
111        super();
112        Args.positive(buffersize, "Buffer size");
113        final HttpTransportMetricsImpl inTransportMetrics = new HttpTransportMetricsImpl();
114        final HttpTransportMetricsImpl outTransportMetrics = new HttpTransportMetricsImpl();
115        this.inbuffer = new SessionInputBufferImpl(inTransportMetrics, buffersize, -1,
116                messageConstraints != null ? messageConstraints : MessageConstraints.DEFAULT, chardecoder);
117        this.outbuffer = new SessionOutputBufferImpl(outTransportMetrics, buffersize, fragmentSizeHint,
118                charencoder);
119        this.messageConstraints = messageConstraints;
120        this.connMetrics = new HttpConnectionMetricsImpl(inTransportMetrics, outTransportMetrics);
121        this.incomingContentStrategy = incomingContentStrategy != null ? incomingContentStrategy :
122            LaxContentLengthStrategy.INSTANCE;
123        this.outgoingContentStrategy = outgoingContentStrategy != null ? outgoingContentStrategy :
124            StrictContentLengthStrategy.INSTANCE;
125        this.socketHolder = new AtomicReference<Socket>();
126    }
127
128    protected void ensureOpen() throws IOException {
129        final Socket socket = this.socketHolder.get();
130        if (socket == null) {
131            throw new ConnectionClosedException("Connection is closed");
132        }
133        if (!this.inbuffer.isBound()) {
134            this.inbuffer.bind(getSocketInputStream(socket));
135        }
136        if (!this.outbuffer.isBound()) {
137            this.outbuffer.bind(getSocketOutputStream(socket));
138        }
139    }
140
141    protected InputStream getSocketInputStream(final Socket socket) throws IOException {
142        return socket.getInputStream();
143    }
144
145    protected OutputStream getSocketOutputStream(final Socket socket) throws IOException {
146        return socket.getOutputStream();
147    }
148
149    /**
150     * Binds this connection to the given {@link Socket}. This socket will be
151     * used by the connection to send and receive data.
152     * <p>
153     * After this method's execution the connection status will be reported
154     * as open and the {@link #isOpen()} will return {@code true}.
155     *
156     * @param socket the socket.
157     * @throws IOException in case of an I/O error.
158     */
159    protected void bind(final Socket socket) throws IOException {
160        Args.notNull(socket, "Socket");
161        this.socketHolder.set(socket);
162        this.inbuffer.bind(null);
163        this.outbuffer.bind(null);
164    }
165
166    protected SessionInputBuffer getSessionInputBuffer() {
167        return this.inbuffer;
168    }
169
170    protected SessionOutputBuffer getSessionOutputBuffer() {
171        return this.outbuffer;
172    }
173
174    protected void doFlush() throws IOException {
175        this.outbuffer.flush();
176    }
177
178    @Override
179    public boolean isOpen() {
180        return this.socketHolder.get() != null;
181    }
182
183    protected Socket getSocket() {
184        return this.socketHolder.get();
185    }
186
187    protected OutputStream createOutputStream(
188            final long len,
189            final SessionOutputBuffer outbuffer) {
190        if (len == ContentLengthStrategy.CHUNKED) {
191            return new ChunkedOutputStream(2048, outbuffer);
192        } else if (len == ContentLengthStrategy.IDENTITY) {
193            return new IdentityOutputStream(outbuffer);
194        } else {
195            return new ContentLengthOutputStream(outbuffer, len);
196        }
197    }
198
199    protected OutputStream prepareOutput(final HttpMessage message) throws HttpException {
200        final long len = this.outgoingContentStrategy.determineLength(message);
201        return createOutputStream(len, this.outbuffer);
202    }
203
204    protected InputStream createInputStream(
205            final long len,
206            final SessionInputBuffer inbuffer) {
207        if (len == ContentLengthStrategy.CHUNKED) {
208            return new ChunkedInputStream(inbuffer, this.messageConstraints);
209        } else if (len == ContentLengthStrategy.IDENTITY) {
210            return new IdentityInputStream(inbuffer);
211        } else if (len == 0L) {
212            return EmptyInputStream.INSTANCE;
213        } else {
214            return new ContentLengthInputStream(inbuffer, len);
215        }
216    }
217
218    protected HttpEntity prepareInput(final HttpMessage message) throws HttpException {
219        final BasicHttpEntity entity = new BasicHttpEntity();
220
221        final long len = this.incomingContentStrategy.determineLength(message);
222        final InputStream instream = createInputStream(len, this.inbuffer);
223        if (len == ContentLengthStrategy.CHUNKED) {
224            entity.setChunked(true);
225            entity.setContentLength(-1);
226            entity.setContent(instream);
227        } else if (len == ContentLengthStrategy.IDENTITY) {
228            entity.setChunked(false);
229            entity.setContentLength(-1);
230            entity.setContent(instream);
231        } else {
232            entity.setChunked(false);
233            entity.setContentLength(len);
234            entity.setContent(instream);
235        }
236
237        final Header contentTypeHeader = message.getFirstHeader(HTTP.CONTENT_TYPE);
238        if (contentTypeHeader != null) {
239            entity.setContentType(contentTypeHeader);
240        }
241        final Header contentEncodingHeader = message.getFirstHeader(HTTP.CONTENT_ENCODING);
242        if (contentEncodingHeader != null) {
243            entity.setContentEncoding(contentEncodingHeader);
244        }
245        return entity;
246    }
247
248    @Override
249    public InetAddress getLocalAddress() {
250        final Socket socket = this.socketHolder.get();
251        return socket != null ? socket.getLocalAddress() : null;
252    }
253
254    @Override
255    public int getLocalPort() {
256        final Socket socket = this.socketHolder.get();
257        return socket != null ? socket.getLocalPort() : -1;
258    }
259
260    @Override
261    public InetAddress getRemoteAddress() {
262        final Socket socket = this.socketHolder.get();
263        return socket != null ? socket.getInetAddress() : null;
264    }
265
266    @Override
267    public int getRemotePort() {
268        final Socket socket = this.socketHolder.get();
269        return socket != null ? socket.getPort() : -1;
270    }
271
272    @Override
273    public void setSocketTimeout(final int timeout) {
274        final Socket socket = this.socketHolder.get();
275        if (socket != null) {
276            try {
277                socket.setSoTimeout(timeout);
278            } catch (final SocketException ignore) {
279                // It is not quite clear from the Sun's documentation if there are any
280                // other legitimate cases for a socket exception to be thrown when setting
281                // SO_TIMEOUT besides the socket being already closed
282            }
283        }
284    }
285
286    @Override
287    public int getSocketTimeout() {
288        final Socket socket = this.socketHolder.get();
289        if (socket != null) {
290            try {
291                return socket.getSoTimeout();
292            } catch (final SocketException ignore) {
293                return -1;
294            }
295        } else {
296            return -1;
297        }
298    }
299
300    @Override
301    public void shutdown() throws IOException {
302        final Socket socket = this.socketHolder.getAndSet(null);
303        if (socket != null) {
304            // force abortive close (RST)
305            try {
306                socket.setSoLinger(true, 0);
307            } catch (final IOException ex) {
308            } finally {
309                socket.close();
310            }
311        }
312    }
313
314    @Override
315    public void close() throws IOException {
316        final Socket socket = this.socketHolder.getAndSet(null);
317        if (socket != null) {
318            try {
319                this.inbuffer.clear();
320                this.outbuffer.flush();
321                try {
322                    try {
323                        socket.shutdownOutput();
324                    } catch (final IOException ignore) {
325                    }
326                    try {
327                        socket.shutdownInput();
328                    } catch (final IOException ignore) {
329                    }
330                } catch (final UnsupportedOperationException ignore) {
331                    // if one isn't supported, the other one isn't either
332                }
333            } finally {
334                socket.close();
335            }
336        }
337    }
338
339    private int fillInputBuffer(final int timeout) throws IOException {
340        final Socket socket = this.socketHolder.get();
341        final int oldtimeout = socket.getSoTimeout();
342        try {
343            socket.setSoTimeout(timeout);
344            return this.inbuffer.fillBuffer();
345        } finally {
346            socket.setSoTimeout(oldtimeout);
347        }
348    }
349
350    protected boolean awaitInput(final int timeout) throws IOException {
351        if (this.inbuffer.hasBufferedData()) {
352            return true;
353        }
354        fillInputBuffer(timeout);
355        return this.inbuffer.hasBufferedData();
356    }
357
358    @Override
359    public boolean isStale() {
360        if (!isOpen()) {
361            return true;
362        }
363        try {
364            final int bytesRead = fillInputBuffer(1);
365            return bytesRead < 0;
366        } catch (final SocketTimeoutException ex) {
367            return false;
368        } catch (final IOException ex) {
369            return true;
370        }
371    }
372
373    protected void incrementRequestCount() {
374        this.connMetrics.incrementRequestCount();
375    }
376
377    protected void incrementResponseCount() {
378        this.connMetrics.incrementResponseCount();
379    }
380
381    @Override
382    public HttpConnectionMetrics getMetrics() {
383        return this.connMetrics;
384    }
385
386    @Override
387    public String toString() {
388        final Socket socket = this.socketHolder.get();
389        if (socket != null) {
390            final StringBuilder buffer = new StringBuilder();
391            final SocketAddress remoteAddress = socket.getRemoteSocketAddress();
392            final SocketAddress localAddress = socket.getLocalSocketAddress();
393            if (remoteAddress != null && localAddress != null) {
394                NetUtils.formatAddress(buffer, localAddress);
395                buffer.append("<->");
396                NetUtils.formatAddress(buffer, remoteAddress);
397            }
398            return buffer.toString();
399        } else {
400            return "[Not bound]";
401        }
402    }
403
404}