001/**
002 * Licensed to the Apache Software Foundation (ASF) under one or more
003 * contributor license agreements.  See the NOTICE file distributed with
004 * this work for additional information regarding copyright ownership.
005 * The ASF licenses this file to You under the Apache License, Version 2.0
006 * (the "License"); you may not use this file except in compliance with
007 * the License.  You may obtain a copy of the License at
008 *
009 *      http://www.apache.org/licenses/LICENSE-2.0
010 *
011 * Unless required by applicable law or agreed to in writing, software
012 * distributed under the License is distributed on an "AS IS" BASIS,
013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014 * See the License for the specific language governing permissions and
015 * limitations under the License.
016 */
017package org.apache.activemq.transport.http;
018
019import java.io.DataInputStream;
020import java.io.IOException;
021import java.io.InterruptedIOException;
022import java.net.URI;
023import java.util.zip.GZIPInputStream;
024import java.util.zip.GZIPOutputStream;
025
026import org.apache.activemq.command.ShutdownInfo;
027import org.apache.activemq.transport.FutureResponse;
028import org.apache.activemq.transport.util.TextWireFormat;
029import org.apache.activemq.util.ByteArrayOutputStream;
030import org.apache.activemq.util.IOExceptionSupport;
031import org.apache.activemq.util.IdGenerator;
032import org.apache.activemq.util.ServiceStopper;
033import org.apache.http.Header;
034import org.apache.http.HttpHost;
035import org.apache.http.HttpRequest;
036import org.apache.http.HttpRequestInterceptor;
037import org.apache.http.HttpResponse;
038import org.apache.http.HttpStatus;
039import org.apache.http.auth.AuthScope;
040import org.apache.http.auth.UsernamePasswordCredentials;
041import org.apache.http.client.HttpClient;
042import org.apache.http.client.HttpResponseException;
043import org.apache.http.client.ResponseHandler;
044import org.apache.http.client.methods.HttpGet;
045import org.apache.http.client.methods.HttpHead;
046import org.apache.http.client.methods.HttpOptions;
047import org.apache.http.client.methods.HttpPost;
048import org.apache.http.conn.ClientConnectionManager;
049import org.apache.http.conn.params.ConnRoutePNames;
050import org.apache.http.conn.scheme.PlainSocketFactory;
051import org.apache.http.conn.scheme.Scheme;
052import org.apache.http.entity.ByteArrayEntity;
053import org.apache.http.impl.client.BasicResponseHandler;
054import org.apache.http.impl.client.DefaultHttpClient;
055import org.apache.http.impl.conn.PoolingClientConnectionManager;
056import org.apache.http.message.AbstractHttpMessage;
057import org.apache.http.params.HttpConnectionParams;
058import org.apache.http.params.HttpParams;
059import org.apache.http.protocol.HttpContext;
060import org.apache.http.util.EntityUtils;
061import org.slf4j.Logger;
062import org.slf4j.LoggerFactory;
063
064/**
065 * A HTTP {@link org.apache.activemq.transport.Transport} which uses the
066 * <a href="http://hc.apache.org/index.html">Apache HTTP Client</a>
067 * library
068 */
069public class HttpClientTransport extends HttpTransportSupport {
070
071    public static final int MAX_CLIENT_TIMEOUT = 30000;
072    private static final Logger LOG = LoggerFactory.getLogger(HttpClientTransport.class);
073    private static final IdGenerator CLIENT_ID_GENERATOR = new IdGenerator();
074
075    private HttpClient sendHttpClient;
076    private HttpClient receiveHttpClient;
077
078    private final String clientID = CLIENT_ID_GENERATOR.generateId();
079    private boolean trace;
080    private HttpGet httpMethod;
081    private volatile int receiveCounter;
082
083    private int soTimeout = MAX_CLIENT_TIMEOUT;
084
085    private boolean useCompression = false;
086    protected boolean canSendCompressed = false;
087    private int minSendAsCompressedSize = 0;
088
089    public HttpClientTransport(TextWireFormat wireFormat, URI remoteUrl) {
090        super(wireFormat, remoteUrl);
091    }
092
093    public FutureResponse asyncRequest(Object command) throws IOException {
094        return null;
095    }
096
097    @Override
098    public void oneway(Object command) throws IOException {
099
100        if (isStopped()) {
101            throw new IOException("stopped.");
102        }
103        HttpPost httpMethod = new HttpPost(getRemoteUrl().toString());
104        configureMethod(httpMethod);
105        String data = getTextWireFormat().marshalText(command);
106        byte[] bytes = data.getBytes("UTF-8");
107        if (useCompression && canSendCompressed && bytes.length > minSendAsCompressedSize) {
108            ByteArrayOutputStream bytesOut = new ByteArrayOutputStream();
109            GZIPOutputStream stream = new GZIPOutputStream(bytesOut);
110            stream.write(bytes);
111            stream.close();
112            httpMethod.addHeader("Content-Type", "application/x-gzip");
113            if (LOG.isTraceEnabled()) {
114                LOG.trace("Sending compressed, size = " + bytes.length + ", compressed size = " + bytesOut.size());
115            }
116            bytes = bytesOut.toByteArray();
117        }
118        ByteArrayEntity entity = new ByteArrayEntity(bytes);
119        httpMethod.setEntity(entity);
120
121        HttpClient client = null;
122        HttpResponse answer = null;
123        try {
124            client = getSendHttpClient();
125            HttpParams params = client.getParams();
126            HttpConnectionParams.setSoTimeout(params, soTimeout);
127            answer = client.execute(httpMethod);
128            int status = answer.getStatusLine().getStatusCode();
129            if (status != HttpStatus.SC_OK) {
130                throw new IOException("Failed to post command: " + command + " as response was: " + answer);
131            }
132            if (command instanceof ShutdownInfo) {
133                try {
134                    stop();
135                } catch (Exception e) {
136                    LOG.warn("Error trying to stop HTTP client: "+ e, e);
137                }
138            }
139        } catch (IOException e) {
140            throw IOExceptionSupport.create("Could not post command: " + command + " due to: " + e, e);
141        } finally {
142            if (answer != null) {
143                EntityUtils.consume(answer.getEntity());
144            }
145        }
146    }
147
148    @Override
149    public Object request(Object command) throws IOException {
150        return null;
151    }
152
153    private DataInputStream createDataInputStream(HttpResponse answer) throws IOException {
154        Header encoding = answer.getEntity().getContentEncoding();
155        if (encoding != null && "gzip".equalsIgnoreCase(encoding.getValue())) {
156            return new DataInputStream(new GZIPInputStream(answer.getEntity().getContent()));
157        } else {
158            return new DataInputStream(answer.getEntity().getContent());
159        }
160    }
161
162    @Override
163    public void run() {
164
165        if (LOG.isTraceEnabled()) {
166            LOG.trace("HTTP GET consumer thread starting: " + this);
167        }
168        HttpClient httpClient = getReceiveHttpClient();
169        URI remoteUrl = getRemoteUrl();
170
171        while (!isStopped() && !isStopping()) {
172
173            httpMethod = new HttpGet(remoteUrl.toString());
174            configureMethod(httpMethod);
175            HttpResponse answer = null;
176
177            try {
178                answer = httpClient.execute(httpMethod);
179                int status = answer.getStatusLine().getStatusCode();
180                if (status != HttpStatus.SC_OK) {
181                    if (status == HttpStatus.SC_REQUEST_TIMEOUT) {
182                        LOG.debug("GET timed out");
183                        try {
184                            Thread.sleep(1000);
185                        } catch (InterruptedException e) {
186                            onException(new InterruptedIOException());
187                            Thread.currentThread().interrupt();
188                            break;
189                        }
190                    } else {
191                        onException(new IOException("Failed to perform GET on: " + remoteUrl + " as response was: " + answer));
192                        break;
193                    }
194                } else {
195                    receiveCounter++;
196                    DataInputStream stream = createDataInputStream(answer);
197                    Object command = getTextWireFormat().unmarshal(stream);
198                    if (command == null) {
199                        LOG.debug("Received null command from url: " + remoteUrl);
200                    } else {
201                        doConsume(command);
202                    }
203                    stream.close();
204                }
205            } catch (IOException e) {
206                onException(IOExceptionSupport.create("Failed to perform GET on: " + remoteUrl + " Reason: " + e.getMessage(), e));
207                break;
208            } finally {
209                if (answer != null) {
210                    try {
211                        EntityUtils.consume(answer.getEntity());
212                    } catch (IOException e) {
213                    }
214                }
215            }
216        }
217    }
218
219    // Properties
220    // -------------------------------------------------------------------------
221    public HttpClient getSendHttpClient() {
222        if (sendHttpClient == null) {
223            sendHttpClient = createHttpClient();
224        }
225        return sendHttpClient;
226    }
227
228    public void setSendHttpClient(HttpClient sendHttpClient) {
229        this.sendHttpClient = sendHttpClient;
230    }
231
232    public HttpClient getReceiveHttpClient() {
233        if (receiveHttpClient == null) {
234            receiveHttpClient = createHttpClient();
235        }
236        return receiveHttpClient;
237    }
238
239    public void setReceiveHttpClient(HttpClient receiveHttpClient) {
240        this.receiveHttpClient = receiveHttpClient;
241    }
242
243    // Implementation methods
244    // -------------------------------------------------------------------------
245    @Override
246    protected void doStart() throws Exception {
247
248        if (LOG.isTraceEnabled()) {
249            LOG.trace("HTTP GET consumer thread starting: " + this);
250        }
251        HttpClient httpClient = getReceiveHttpClient();
252        URI remoteUrl = getRemoteUrl();
253
254        HttpHead httpMethod = new HttpHead(remoteUrl.toString());
255        configureMethod(httpMethod);
256
257        // Request the options from the server so we can find out if the broker we are
258        // talking to supports GZip compressed content.  If so and useCompression is on
259        // then we can compress our POST data, otherwise we must send it uncompressed to
260        // ensure backwards compatibility.
261        HttpOptions optionsMethod = new HttpOptions(remoteUrl.toString());
262        ResponseHandler<String> handler = new BasicResponseHandler() {
263            @Override
264            public String handleResponse(HttpResponse response) throws HttpResponseException, IOException {
265
266                for(Header header : response.getAllHeaders()) {
267                    if (header.getName().equals("Accepts-Encoding") && header.getValue().contains("gzip")) {
268                        LOG.info("Broker Servlet supports GZip compression.");
269                        canSendCompressed = true;
270                        break;
271                    }
272                }
273
274                return super.handleResponse(response);
275            }
276        };
277
278        try {
279            httpClient.execute(httpMethod, new BasicResponseHandler());
280            httpClient.execute(optionsMethod, handler);
281        } catch(Exception e) {
282            throw new IOException("Failed to perform GET on: " + remoteUrl + " as response was: " + e.getMessage());
283        }
284
285        super.doStart();
286    }
287
288    @Override
289    protected void doStop(ServiceStopper stopper) throws Exception {
290        if (httpMethod != null) {
291            // In some versions of the JVM a race between the httpMethod and the completion
292            // of the method when using HTTPS can lead to a deadlock.  This hack attempts to
293            // detect that and interrupt the thread that's locked so that they can complete
294            // on another attempt.
295            for (int i = 0; i < 3; ++i) {
296                Thread abortThread = new Thread(new Runnable() {
297
298                    @Override
299                    public void run() {
300                        try {
301                            httpMethod.abort();
302                        } catch (Exception e) {
303                        }
304                    }
305                });
306
307                abortThread.start();
308                abortThread.join(2000);
309                if (abortThread.isAlive() && !httpMethod.isAborted()) {
310                    abortThread.interrupt();
311                }
312            }
313        }
314    }
315
316    protected HttpClient createHttpClient() {
317        DefaultHttpClient client = new DefaultHttpClient(createClientConnectionManager());
318        if (useCompression) {
319            client.addRequestInterceptor( new HttpRequestInterceptor() {
320                @Override
321                public void process(HttpRequest request, HttpContext context) {
322                    // We expect to received a compression response that we un-gzip
323                    request.addHeader("Accept-Encoding", "gzip");
324                }
325            });
326        }
327        if (getProxyHost() != null) {
328            HttpHost proxy = new HttpHost(getProxyHost(), getProxyPort());
329            client.getParams().setParameter(ConnRoutePNames.DEFAULT_PROXY, proxy);
330
331            if (client.getConnectionManager().getSchemeRegistry().get("http") == null) {
332                client.getConnectionManager().getSchemeRegistry().register(
333                    new Scheme("http", getProxyPort(), PlainSocketFactory.getSocketFactory()));
334            }
335
336            if(getProxyUser() != null && getProxyPassword() != null) {
337                client.getCredentialsProvider().setCredentials(
338                    new AuthScope(getProxyHost(), getProxyPort()),
339                    new UsernamePasswordCredentials(getProxyUser(), getProxyPassword()));
340            }
341        }
342        return client;
343    }
344
345    protected ClientConnectionManager createClientConnectionManager() {
346        return new PoolingClientConnectionManager();
347    }
348
349    protected void configureMethod(AbstractHttpMessage method) {
350        method.setHeader("clientID", clientID);
351    }
352
353    public boolean isTrace() {
354        return trace;
355    }
356
357    public void setTrace(boolean trace) {
358        this.trace = trace;
359    }
360
361    @Override
362    public int getReceiveCounter() {
363        return receiveCounter;
364    }
365
366    public int getSoTimeout() {
367        return soTimeout;
368    }
369
370    public void setSoTimeout(int soTimeout) {
371        this.soTimeout = soTimeout;
372    }
373
374    public void setUseCompression(boolean useCompression) {
375        this.useCompression = useCompression;
376    }
377
378    public boolean isUseCompression() {
379        return this.useCompression;
380    }
381
382    public int getMinSendAsCompressedSize() {
383        return minSendAsCompressedSize;
384    }
385
386    /**
387     * Sets the minimum size that must be exceeded on a send before compression is used if
388     * the useCompression option is specified.  For very small payloads compression can be
389     * inefficient compared to the transmission size savings.
390     *
391     * Default value is 0.
392     *
393     * @param minSendAsCompressedSize
394     */
395    public void setMinSendAsCompressedSize(int minSendAsCompressedSize) {
396        this.minSendAsCompressedSize = minSendAsCompressedSize;
397    }
398
399}