Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix for https://issues.apache.org/jira/browse/AMQ-3521 

Moves the HTTP client up to the newest v4.1.2 release. Update some additional dependencies as well.

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1179226 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Timothy A. Bish 2011-10-05 13:49:50 +00:00
parent 71d4e2a2aa
commit 082fdc5627
5 changed files with 268 additions and 255 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
activemq-optional/pom.xml
View File

@ -6,9 +6,9 @@
The ASF licenses this file to You under the Apache License, Version 2.0 The ASF licenses this file to You under the Apache License, Version 2.0
(the "License"); you may not use this file except in compliance with (the "License"); you may not use this file except in compliance with
the License. You may obtain a copy of the License at the License. You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0 http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS, distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@ -16,7 +16,7 @@
limitations under the License. limitations under the License.
--> -->
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd"> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
<modelVersion>4.0.0</modelVersion> <modelVersion>4.0.0</modelVersion>
<parent> <parent>
@ -44,7 +44,7 @@
</releases> </releases>
</repository> </repository>
</repositories> </repositories>
<dependencies> <dependencies>
<!-- activemq --> <!-- activemq -->
@ -60,12 +60,12 @@
<dependency> <dependency>
<groupId>${project.groupId}</groupId> <groupId>${project.groupId}</groupId>
<artifactId>activeio-core</artifactId> <artifactId>activeio-core</artifactId>
</dependency> </dependency>
<dependency> <dependency>
<groupId>${project.groupId}</groupId> <groupId>${project.groupId}</groupId>
<artifactId>activemq-console</artifactId> <artifactId>activemq-console</artifactId>
</dependency> </dependency>
<dependency> <dependency>
<groupId>org.springframework</groupId> <groupId>org.springframework</groupId>
<artifactId>spring-core</artifactId> <artifactId>spring-core</artifactId>
@ -80,7 +80,7 @@
<groupId>aopalliance</groupId> <groupId>aopalliance</groupId>
<artifactId>aopalliance</artifactId> <artifactId>aopalliance</artifactId>
</dependency> </dependency>
<dependency> <dependency>
<groupId>com.thoughtworks.xstream</groupId> <groupId>com.thoughtworks.xstream</groupId>
<artifactId>xstream</artifactId> <artifactId>xstream</artifactId>
@ -89,7 +89,7 @@
<groupId>xpp3</groupId> <groupId>xpp3</groupId>
<artifactId>xpp3</artifactId> <artifactId>xpp3</artifactId>
</dependency> </dependency>
<dependency> <dependency>
<groupId>org.eclipse.jetty.aggregate</groupId> <groupId>org.eclipse.jetty.aggregate</groupId>
<artifactId>jetty-all-server</artifactId> <artifactId>jetty-all-server</artifactId>
@ -104,22 +104,23 @@
<artifactId>jetty-websocket</artifactId> <artifactId>jetty-websocket</artifactId>
<version>${jetty-version}</version> <version>${jetty-version}</version>
</dependency> </dependency>
<dependency> <dependency>
<groupId>axis</groupId> <groupId>axis</groupId>
<artifactId>axis</artifactId> <artifactId>axis</artifactId>
</dependency> </dependency>
<dependency> <dependency>
<groupId>commons-httpclient</groupId> <groupId>org.apache.httpcomponents</groupId>
<artifactId>commons-httpclient</artifactId> <artifactId>httpclient</artifactId>
<version>${httpclient-version}</version>
</dependency> </dependency>
<dependency> <dependency>
<groupId>commons-pool</groupId> <groupId>commons-pool</groupId>
<artifactId>commons-pool</artifactId> <artifactId>commons-pool</artifactId>
</dependency> </dependency>
<dependency> <dependency>
<groupId>commons-collections</groupId> <groupId>commons-collections</groupId>
<artifactId>commons-collections</artifactId> <artifactId>commons-collections</artifactId>
@ -152,8 +153,8 @@
<groupId>xmlbeans</groupId> <groupId>xmlbeans</groupId>
<artifactId>xbean_xpath</artifactId> <artifactId>xbean_xpath</artifactId>
<optional>true</optional> <optional>true</optional>
</dependency> </dependency>
<dependency> <dependency>
<groupId>xalan</groupId> <groupId>xalan</groupId>
<artifactId>xalan</artifactId> <artifactId>xalan</artifactId>
@ -172,21 +173,21 @@
<scope>test</scope> <scope>test</scope>
</dependency> </dependency>
<dependency> <dependency>
<groupId>org.springframework.ws</groupId> <groupId>org.springframework.ws</groupId>
<artifactId>spring-oxm-tiger</artifactId> <artifactId>spring-oxm-tiger</artifactId>
<optional>true</optional> <optional>true</optional>
</dependency> </dependency>
<dependency> <dependency>
<groupId>org.apache.xbean</groupId> <groupId>org.apache.xbean</groupId>
<artifactId>xbean-spring</artifactId> <artifactId>xbean-spring</artifactId>
<scope>test</scope> <scope>test</scope>
</dependency> </dependency>
<dependency> <dependency>
<groupId>org.codehaus.jettison</groupId> <groupId>org.codehaus.jettison</groupId>
<artifactId>jettison</artifactId> <artifactId>jettison</artifactId>
<scope>test</scope> <scope>test</scope>
</dependency> </dependency>
<dependency> <dependency>
<groupId>org.apache.velocity</groupId> <groupId>org.apache.velocity</groupId>
<artifactId>velocity</artifactId> <artifactId>velocity</artifactId>
@ -199,8 +200,8 @@
<groupId>net.sf.josql</groupId> <groupId>net.sf.josql</groupId>
<artifactId>gentlyweb-utils</artifactId> <artifactId>gentlyweb-utils</artifactId>
</dependency> </dependency>
</dependencies> </dependencies>
<build> <build>

308
activemq-optional/src/main/java/org/apache/activemq/transport/discovery/http/HTTPDiscoveryAgent.java
View File

@ -31,39 +31,42 @@ import org.apache.activemq.command.DiscoveryEvent;
import org.apache.activemq.transport.discovery.DiscoveryAgent; import org.apache.activemq.transport.discovery.DiscoveryAgent;
import org.apache.activemq.transport.discovery.DiscoveryListener; import org.apache.activemq.transport.discovery.DiscoveryListener;
import org.apache.activemq.util.IntrospectionSupport; import org.apache.activemq.util.IntrospectionSupport;
import org.apache.commons.httpclient.HttpClient; import org.apache.http.client.HttpClient;
import org.apache.commons.httpclient.methods.DeleteMethod; import org.apache.http.client.ResponseHandler;
import org.apache.commons.httpclient.methods.GetMethod; import org.apache.http.client.methods.HttpDelete;
import org.apache.commons.httpclient.methods.PutMethod; import org.apache.http.client.methods.HttpGet;
import org.apache.http.client.methods.HttpPut;
import org.apache.http.impl.client.BasicResponseHandler;
import org.apache.http.impl.client.DefaultHttpClient;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
public class HTTPDiscoveryAgent implements DiscoveryAgent { public class HTTPDiscoveryAgent implements DiscoveryAgent {
private static final Logger LOG = LoggerFactory.getLogger(HTTPDiscoveryAgent.class); private static final Logger LOG = LoggerFactory.getLogger(HTTPDiscoveryAgent.class);
private String registryURL = "http://localhost:8080/discovery-registry/default"; private String registryURL = "http://localhost:8080/discovery-registry/default";
private HttpClient httpClient = new HttpClient(); private HttpClient httpClient = new DefaultHttpClient();
private AtomicBoolean running=new AtomicBoolean(); private AtomicBoolean running = new AtomicBoolean();
private final AtomicReference<DiscoveryListener> discoveryListener = new AtomicReference<DiscoveryListener>(); private final AtomicReference<DiscoveryListener> discoveryListener = new AtomicReference<DiscoveryListener>();
private final HashSet<String> registeredServices = new HashSet<String>(); private final HashSet<String> registeredServices = new HashSet<String>();
private final HashMap<String, SimpleDiscoveryEvent> discoveredServices = new HashMap<String, SimpleDiscoveryEvent>(); private final HashMap<String, SimpleDiscoveryEvent> discoveredServices = new HashMap<String, SimpleDiscoveryEvent>();
private Thread thread; private Thread thread;
private long updateInterval = 1000*10; private long updateInterval = 1000 * 10;
@SuppressWarnings("unused")
private String brokerName; private String brokerName;
private boolean startEmbeddRegistry=false; private boolean startEmbeddRegistry = false;
private Service jetty; private Service jetty;
private AtomicInteger startCounter=new AtomicInteger(0); private AtomicInteger startCounter = new AtomicInteger(0);
private long initialReconnectDelay = 1000; private long initialReconnectDelay = 1000;
private long maxReconnectDelay = 1000 * 30; private long maxReconnectDelay = 1000 * 30;
private long backOffMultiplier = 2; private long backOffMultiplier = 2;
private boolean useExponentialBackOff=true; private boolean useExponentialBackOff = true;
private int maxReconnectAttempts; private int maxReconnectAttempts;
private final Object sleepMutex = new Object(); private final Object sleepMutex = new Object();
private long minConnectTime = 5000; private long minConnectTime = 5000;
class SimpleDiscoveryEvent extends DiscoveryEvent { class SimpleDiscoveryEvent extends DiscoveryEvent {
private int connectFailures; private int connectFailures;
@ -75,16 +78,14 @@ public class HTTPDiscoveryAgent implements DiscoveryAgent {
public SimpleDiscoveryEvent(String service) { public SimpleDiscoveryEvent(String service) {
super(service); super(service);
} }
} }
public String getGroup() { public String getGroup() {
return null; return null;
} }
public void registerService(String service) throws IOException { public void registerService(String service) throws IOException {
synchronized(registeredServices) { synchronized (registeredServices) {
registeredServices.add(service); registeredServices.add(service);
} }
doRegister(service); doRegister(service);
@ -93,128 +94,121 @@ public class HTTPDiscoveryAgent implements DiscoveryAgent {
synchronized private void doRegister(String service) { synchronized private void doRegister(String service) {
String url = registryURL; String url = registryURL;
try { try {
PutMethod method = new PutMethod(url); HttpPut method = new HttpPut(url);
// method.setParams(createParams()); method.addHeader("service", service);
method.setRequestHeader("service", service); ResponseHandler<String> handler = new BasicResponseHandler();
int responseCode = httpClient.executeMethod(method); String responseBody = httpClient.execute(method, handler);
LOG.debug("PUT to "+url+" got a "+responseCode); LOG.debug("PUT to " + url + " got a " + responseBody);
} catch (Exception e) { } catch (Exception e) {
LOG.debug("PUT to "+url+" failed with: "+e); LOG.debug("PUT to " + url + " failed with: " + e);
}
}
synchronized private void doUnRegister(String service) {
String url = registryURL;
try {
DeleteMethod method = new DeleteMethod(url);
// method.setParams(createParams());
method.setRequestHeader("service", service);
int responseCode = httpClient.executeMethod(method);
LOG.debug("DELETE to "+url+" got a "+responseCode);
} catch (Exception e) {
LOG.debug("DELETE to "+url+" failed with: "+e);
} }
} }
// private HttpMethodParams createParams() { @SuppressWarnings("unused")
// HttpMethodParams params = new HttpMethodParams(); synchronized private void doUnRegister(String service) {
// params.setParameter(HttpMethodParams.RETRY_HANDLER, new DefaultHttpMethodRetryHandler(0,false)); String url = registryURL;
// return params;
// }
synchronized private Set<String> doLookup(long freshness) {
String url = registryURL+"?freshness="+freshness;
try { try {
GetMethod method = new GetMethod(url); HttpDelete method = new HttpDelete(url);
// method.setParams(createParams()); method.addHeader("service", service);
int responseCode = httpClient.executeMethod(method); ResponseHandler<String> handler = new BasicResponseHandler();
LOG.debug("GET to "+url+" got a "+responseCode); String responseBody = httpClient.execute(method, handler);
if( responseCode == 200 ) { LOG.debug("DELETE to " + url + " got a " + responseBody);
Set<String> rc = new HashSet<String>();
Scanner scanner = new Scanner(method.getResponseBodyAsStream());
while( scanner.hasNextLine() ) {
String service = scanner.nextLine();
if( service.trim().length() != 0 ) {
rc.add(service);
}
}
return rc;
} else {
LOG.debug("GET to "+url+" failed with response code: "+responseCode);
return null;
}
} catch (Exception e) { } catch (Exception e) {
LOG.debug("GET to "+url+" failed with: "+e); LOG.debug("DELETE to " + url + " failed with: " + e);
}
}
synchronized private Set<String> doLookup(long freshness) {
String url = registryURL + "?freshness=" + freshness;
try {
HttpGet method = new HttpGet(url);
ResponseHandler<String> handler = new BasicResponseHandler();
String response = httpClient.execute(method, handler);
LOG.debug("GET to " + url + " got a " + response);
Set<String> rc = new HashSet<String>();
Scanner scanner = new Scanner(response);
while (scanner.hasNextLine()) {
String service = scanner.nextLine();
if (service.trim().length() != 0) {
rc.add(service);
}
}
return rc;
} catch (Exception e) {
LOG.debug("GET to " + url + " failed with: " + e);
return null; return null;
} }
} }
public void serviceFailed(DiscoveryEvent devent) throws IOException { public void serviceFailed(DiscoveryEvent devent) throws IOException {
final SimpleDiscoveryEvent event = (SimpleDiscoveryEvent)devent; final SimpleDiscoveryEvent event = (SimpleDiscoveryEvent) devent;
if (event.failed.compareAndSet(false, true)) { if (event.failed.compareAndSet(false, true)) {
discoveryListener.get().onServiceRemove(event); discoveryListener.get().onServiceRemove(event);
if(!event.removed.get()) { if (!event.removed.get()) {
// Setup a thread to re-raise the event... // Setup a thread to re-raise the event...
Thread thread = new Thread() { Thread thread = new Thread() {
public void run() { public void run() {
// We detect a failed connection attempt because the service // We detect a failed connection attempt because the
// fails right away. // service
if (event.connectTime + minConnectTime > System.currentTimeMillis()) { // fails right away.
LOG.debug("Failure occured soon after the discovery event was generated. It will be clasified as a connection failure: "+event); if (event.connectTime + minConnectTime > System.currentTimeMillis()) {
LOG.debug("Failure occured soon after the discovery event was generated. " +
event.connectFailures++; "It will be clasified as a connection failure: " + event);
if (maxReconnectAttempts > 0 && event.connectFailures >= maxReconnectAttempts) { event.connectFailures++;
LOG.debug("Reconnect attempts exceeded "+maxReconnectAttempts+" tries. Reconnecting has been disabled.");
return; if (maxReconnectAttempts > 0 && event.connectFailures >= maxReconnectAttempts) {
} LOG.debug("Reconnect attempts exceeded " + maxReconnectAttempts +
" tries. Reconnecting has been disabled.");
synchronized (sleepMutex) { return;
try { }
if (!running.get() || event.removed.get()) {
return; synchronized (sleepMutex) {
} try {
LOG.debug("Waiting "+event.reconnectDelay+" ms before attepting to reconnect."); if (!running.get() || event.removed.get()) {
sleepMutex.wait(event.reconnectDelay); return;
} catch (InterruptedException ie) { }
Thread.currentThread().interrupt(); LOG.debug("Waiting " + event.reconnectDelay +
return; " ms before attepting to reconnect.");
} sleepMutex.wait(event.reconnectDelay);
} } catch (InterruptedException ie) {
Thread.currentThread().interrupt();
if (!useExponentialBackOff) { return;
event.reconnectDelay = initialReconnectDelay; }
} else { }
// Exponential increment of reconnect delay.
event.reconnectDelay *= backOffMultiplier; if (!useExponentialBackOff) {
if (event.reconnectDelay > maxReconnectDelay) { event.reconnectDelay = initialReconnectDelay;
event.reconnectDelay = maxReconnectDelay; } else {
} // Exponential increment of reconnect delay.
} event.reconnectDelay *= backOffMultiplier;
if (event.reconnectDelay > maxReconnectDelay) {
} else { event.reconnectDelay = maxReconnectDelay;
event.connectFailures = 0; }
event.reconnectDelay = initialReconnectDelay; }
}
} else {
if (!running.get() || event.removed.get()) { event.connectFailures = 0;
return; event.reconnectDelay = initialReconnectDelay;
} }
event.connectTime = System.currentTimeMillis(); if (!running.get() || event.removed.get()) {
event.failed.set(false); return;
discoveryListener.get().onServiceAdd(event); }
}
}; event.connectTime = System.currentTimeMillis();
thread.setDaemon(true); event.failed.set(false);
thread.start(); discoveryListener.get().onServiceAdd(event);
} }
};
thread.setDaemon(true);
thread.start();
}
} }
} }
public void setBrokerName(String brokerName) { public void setBrokerName(String brokerName) {
this.brokerName = brokerName; this.brokerName = brokerName;
} }
@ -227,20 +221,20 @@ public class HTTPDiscoveryAgent implements DiscoveryAgent {
} }
public void start() throws Exception { public void start() throws Exception {
if( startCounter.addAndGet(1)==1 ) { if (startCounter.addAndGet(1) == 1) {
if( startEmbeddRegistry ) { if (startEmbeddRegistry) {
jetty = createEmbeddedJettyServer(); jetty = createEmbeddedJettyServer();
Map props = new HashMap(); Map<String, Object> props = new HashMap<String, Object>();
props.put("agent", this); props.put("agent", this);
IntrospectionSupport.setProperties(jetty, props); IntrospectionSupport.setProperties(jetty, props);
jetty.start(); jetty.start();
} }
running.set(true); running.set(true);
thread = new Thread("HTTPDiscovery Agent") { thread = new Thread("HTTPDiscovery Agent") {
@Override @Override
public void run() { public void run() {
while(running.get()) { while (running.get()) {
try { try {
update(); update();
Thread.sleep(updateInterval); Thread.sleep(updateInterval);
@ -256,51 +250,52 @@ public class HTTPDiscoveryAgent implements DiscoveryAgent {
} }
/** /**
* Create the EmbeddedJettyServer instance via reflection so that we can avoid a hard runtime dependency on * Create the EmbeddedJettyServer instance via reflection so that we can
* jetty. * avoid a hard runtime dependency on jetty.
* *
* @return * @return
* @throws Exception * @throws Exception
*/ */
private Service createEmbeddedJettyServer() throws Exception { private Service createEmbeddedJettyServer() throws Exception {
Class clazz = HTTPDiscoveryAgent.class.getClassLoader().loadClass("org.apache.activemq.transport.discovery.http.EmbeddedJettyServer"); Class<?> clazz = HTTPDiscoveryAgent.class.getClassLoader().loadClass("org.apache.activemq.transport.discovery.http.EmbeddedJettyServer");
return (Service)clazz.newInstance(); return (Service) clazz.newInstance();
} }
private void update() { private void update() {
// Register all our services... // Register all our services...
synchronized(registeredServices) { synchronized (registeredServices) {
for (String service : registeredServices) { for (String service : registeredServices) {
doRegister(service); doRegister(service);
} }
} }
// Find new registered services... // Find new registered services...
DiscoveryListener discoveryListener = this.discoveryListener.get(); DiscoveryListener discoveryListener = this.discoveryListener.get();
if(discoveryListener!=null) { if (discoveryListener != null) {
Set<String> activeServices = doLookup(updateInterval*3); Set<String> activeServices = doLookup(updateInterval * 3);
// If there is error talking the the central server, then activeServices == null // If there is error talking the the central server, then
if( activeServices !=null ) { // activeServices == null
synchronized(discoveredServices) { if (activeServices != null) {
synchronized (discoveredServices) {
HashSet<String> removedServices = new HashSet<String>(discoveredServices.keySet()); HashSet<String> removedServices = new HashSet<String>(discoveredServices.keySet());
removedServices.removeAll(activeServices); removedServices.removeAll(activeServices);
HashSet<String> addedServices = new HashSet<String>(activeServices); HashSet<String> addedServices = new HashSet<String>(activeServices);
addedServices.removeAll(discoveredServices.keySet()); addedServices.removeAll(discoveredServices.keySet());
addedServices.removeAll(removedServices); addedServices.removeAll(removedServices);
for (String service : addedServices) { for (String service : addedServices) {
SimpleDiscoveryEvent e = new SimpleDiscoveryEvent(service); SimpleDiscoveryEvent e = new SimpleDiscoveryEvent(service);
discoveredServices.put(service, e); discoveredServices.put(service, e);
discoveryListener.onServiceAdd(e); discoveryListener.onServiceAdd(e);
} }
for (String service : removedServices) { for (String service : removedServices) {
SimpleDiscoveryEvent e = discoveredServices.remove(service); SimpleDiscoveryEvent e = discoveredServices.remove(service);
if( e !=null ) { if (e != null) {
e.removed.set(true); e.removed.set(true);
} }
discoveryListener.onServiceRemove(e); discoveryListener.onServiceRemove(e);
} }
} }
@ -309,13 +304,13 @@ public class HTTPDiscoveryAgent implements DiscoveryAgent {
} }
public void stop() throws Exception { public void stop() throws Exception {
if( startCounter.decrementAndGet()==0 ) { if (startCounter.decrementAndGet() == 0) {
running.set(false); running.set(false);
if( thread!=null ) { if (thread != null) {
thread.join(updateInterval*3); thread.join(updateInterval * 3);
thread=null; thread = null;
} }
if( jetty!=null ) { if (jetty != null) {
jetty.stop(); jetty.stop();
jetty = null; jetty = null;
} }
@ -345,5 +340,4 @@ public class HTTPDiscoveryAgent implements DiscoveryAgent {
public void setStartEmbeddRegistry(boolean startEmbeddRegistry) { public void setStartEmbeddRegistry(boolean startEmbeddRegistry) {
this.startEmbeddRegistry = startEmbeddRegistry; this.startEmbeddRegistry = startEmbeddRegistry;
} }
} }

107
activemq-optional/src/main/java/org/apache/activemq/transport/http/HttpClientTransport.java
View File

@ -24,18 +24,25 @@ import java.net.URI;
import org.apache.activemq.command.ShutdownInfo; import org.apache.activemq.command.ShutdownInfo;
import org.apache.activemq.transport.FutureResponse; import org.apache.activemq.transport.FutureResponse;
import org.apache.activemq.transport.util.TextWireFormat; import org.apache.activemq.transport.util.TextWireFormat;
import org.apache.activemq.util.ByteArrayInputStream;
import org.apache.activemq.util.IOExceptionSupport; import org.apache.activemq.util.IOExceptionSupport;
import org.apache.activemq.util.IdGenerator; import org.apache.activemq.util.IdGenerator;
import org.apache.activemq.util.ServiceStopper; import org.apache.activemq.util.ServiceStopper;
import org.apache.commons.httpclient.HttpClient; import org.apache.http.HttpHost;
import org.apache.commons.httpclient.HttpMethod; import org.apache.http.HttpResponse;
import org.apache.commons.httpclient.HttpStatus; import org.apache.http.HttpStatus;
import org.apache.commons.httpclient.methods.GetMethod; import org.apache.http.client.HttpClient;
import org.apache.commons.httpclient.methods.HeadMethod; import org.apache.http.client.ResponseHandler;
import org.apache.commons.httpclient.methods.InputStreamRequestEntity; import org.apache.http.client.methods.HttpGet;
import org.apache.commons.httpclient.methods.PostMethod; import org.apache.http.client.methods.HttpHead;
import org.apache.commons.httpclient.params.HttpClientParams; import org.apache.http.client.methods.HttpPost;
import org.apache.http.conn.params.ConnRoutePNames;
import org.apache.http.entity.ByteArrayEntity;
import org.apache.http.impl.client.BasicResponseHandler;
import org.apache.http.impl.client.DefaultHttpClient;
import org.apache.http.message.AbstractHttpMessage;
import org.apache.http.params.HttpConnectionParams;
import org.apache.http.params.HttpParams;
import org.apache.http.util.EntityUtils;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
@ -43,8 +50,6 @@ import org.slf4j.LoggerFactory;
* A HTTP {@link org.apache.activemq.transport.TransportChannel} which uses the * A HTTP {@link org.apache.activemq.transport.TransportChannel} which uses the
* <a href="http://jakarta.apache.org/commons/httpclient/">commons-httpclient</a> * <a href="http://jakarta.apache.org/commons/httpclient/">commons-httpclient</a>
* library * library
*
*
*/ */
public class HttpClientTransport extends HttpTransportSupport { public class HttpClientTransport extends HttpTransportSupport {
@ -57,11 +62,11 @@ public class HttpClientTransport extends HttpTransportSupport {
private final String clientID = CLIENT_ID_GENERATOR.generateId(); private final String clientID = CLIENT_ID_GENERATOR.generateId();
private boolean trace; private boolean trace;
private GetMethod httpMethod; private HttpGet httpMethod;
private volatile int receiveCounter; private volatile int receiveCounter;
private int soTimeout = MAX_CLIENT_TIMEOUT; private int soTimeout = MAX_CLIENT_TIMEOUT;
public HttpClientTransport(TextWireFormat wireFormat, URI remoteUrl) { public HttpClientTransport(TextWireFormat wireFormat, URI remoteUrl) {
super(wireFormat, remoteUrl); super(wireFormat, remoteUrl);
} }
@ -75,35 +80,37 @@ public class HttpClientTransport extends HttpTransportSupport {
if (isStopped()) { if (isStopped()) {
throw new IOException("stopped."); throw new IOException("stopped.");
} }
PostMethod httpMethod = new PostMethod(getRemoteUrl().toString()); HttpPost httpMethod = new HttpPost(getRemoteUrl().toString());
configureMethod(httpMethod); configureMethod(httpMethod);
String data = getTextWireFormat().marshalText(command); String data = getTextWireFormat().marshalText(command);
byte[] bytes = data.getBytes("UTF-8"); byte[] bytes = data.getBytes("UTF-8");
InputStreamRequestEntity entity = new InputStreamRequestEntity(new ByteArrayInputStream(bytes)); ByteArrayEntity entity = new ByteArrayEntity(bytes);
httpMethod.setRequestEntity(entity); httpMethod.setEntity(entity);
HttpClient client = null;
HttpResponse answer = null;
try { try {
client = getSendHttpClient();
HttpClient client = getSendHttpClient(); HttpParams params = client.getParams();
HttpClientParams params = new HttpClientParams(); HttpConnectionParams.setSoTimeout(params, soTimeout);
params.setSoTimeout(soTimeout); answer = client.execute(httpMethod);
client.setParams(params); int status = answer.getStatusLine().getStatusCode();
int answer = client.executeMethod(httpMethod); if (status != HttpStatus.SC_OK) {
if (answer != HttpStatus.SC_OK) {
throw new IOException("Failed to post command: " + command + " as response was: " + answer); throw new IOException("Failed to post command: " + command + " as response was: " + answer);
} }
if (command instanceof ShutdownInfo) { if (command instanceof ShutdownInfo) {
try { try {
stop(); stop();
} catch (Exception e) { } catch (Exception e) {
LOG.warn("Error trying to stop HTTP client: "+ e, e); LOG.warn("Error trying to stop HTTP client: "+ e, e);
} }
} }
} catch (IOException e) { } catch (IOException e) {
throw IOExceptionSupport.create("Could not post command: " + command + " due to: " + e, e); throw IOExceptionSupport.create("Could not post command: " + command + " due to: " + e, e);
} finally { } finally {
httpMethod.getResponseBody(); if (answer != null) {
httpMethod.releaseConnection(); EntityUtils.consume(answer.getEntity());
}
} }
} }
@ -119,13 +126,15 @@ public class HttpClientTransport extends HttpTransportSupport {
while (!isStopped() && !isStopping()) { while (!isStopped() && !isStopping()) {
httpMethod = new GetMethod(remoteUrl.toString()); httpMethod = new HttpGet(remoteUrl.toString());
configureMethod(httpMethod); configureMethod(httpMethod);
HttpResponse answer = null;
try { try {
int answer = httpClient.executeMethod(httpMethod); answer = httpClient.execute(httpMethod);
if (answer != HttpStatus.SC_OK) { int status = answer.getStatusLine().getStatusCode();
if (answer == HttpStatus.SC_REQUEST_TIMEOUT) { if (status != HttpStatus.SC_OK) {
if (status == HttpStatus.SC_REQUEST_TIMEOUT) {
LOG.debug("GET timed out"); LOG.debug("GET timed out");
try { try {
Thread.sleep(1000); Thread.sleep(1000);
@ -139,19 +148,25 @@ public class HttpClientTransport extends HttpTransportSupport {
} }
} else { } else {
receiveCounter++; receiveCounter++;
DataInputStream stream = new DataInputStream(httpMethod.getResponseBodyAsStream()); DataInputStream stream = new DataInputStream(answer.getEntity().getContent());
Object command = (Object)getTextWireFormat().unmarshal(stream); Object command = (Object)getTextWireFormat().unmarshal(stream);
if (command == null) { if (command == null) {
LOG.debug("Received null command from url: " + remoteUrl); LOG.debug("Received null command from url: " + remoteUrl);
} else { } else {
doConsume(command); doConsume(command);
} }
stream.close();
} }
} catch (IOException e) { } catch (IOException e) {
onException(IOExceptionSupport.create("Failed to perform GET on: " + remoteUrl + " Reason: " + e.getMessage(), e)); onException(IOExceptionSupport.create("Failed to perform GET on: " + remoteUrl + " Reason: " + e.getMessage(), e));
break; break;
} finally { } finally {
httpMethod.releaseConnection(); if (answer != null) {
try {
EntityUtils.consume(answer.getEntity());
} catch (IOException e) {
}
}
} }
} }
} }
@ -188,12 +203,13 @@ public class HttpClientTransport extends HttpTransportSupport {
HttpClient httpClient = getReceiveHttpClient(); HttpClient httpClient = getReceiveHttpClient();
URI remoteUrl = getRemoteUrl(); URI remoteUrl = getRemoteUrl();
HeadMethod httpMethod = new HeadMethod(remoteUrl.toString()); HttpHead httpMethod = new HttpHead(remoteUrl.toString());
configureMethod(httpMethod); configureMethod(httpMethod);
ResponseHandler<String> handler = new BasicResponseHandler();
int answer = httpClient.executeMethod(httpMethod); try {
if (answer != HttpStatus.SC_OK) { httpClient.execute(httpMethod, handler);
throw new IOException("Failed to perform GET on: " + remoteUrl + " as response was: " + answer); } catch(Exception e) {
throw new IOException("Failed to perform GET on: " + remoteUrl + " as response was: " + e.getMessage());
} }
super.doStart(); super.doStart();
@ -206,15 +222,16 @@ public class HttpClientTransport extends HttpTransportSupport {
} }
protected HttpClient createHttpClient() { protected HttpClient createHttpClient() {
HttpClient client = new HttpClient(); HttpClient client = new DefaultHttpClient();
if (getProxyHost() != null) { if (getProxyHost() != null) {
client.getHostConfiguration().setProxy(getProxyHost(), getProxyPort()); HttpHost proxy = new HttpHost(getProxyHost(), getProxyPort());
client.getParams().setParameter(ConnRoutePNames.DEFAULT_PROXY, proxy);
} }
return client; return client;
} }
protected void configureMethod(HttpMethod method) { protected void configureMethod(AbstractHttpMessage method) {
method.setRequestHeader("clientID", clientID); method.setHeader("clientID", clientID);
} }
public boolean isTrace() { public boolean isTrace() {

44
assembly/pom.xml
View File

@ -6,9 +6,9 @@
The ASF licenses this file to You under the Apache License, Version 2.0 The ASF licenses this file to You under the Apache License, Version 2.0
(the "License"); you may not use this file except in compliance with (the "License"); you may not use this file except in compliance with
the License. You may obtain a copy of the License at the License. You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0 http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS, distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@ -16,7 +16,7 @@
limitations under the License. limitations under the License.
--> -->
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd"> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
<modelVersion>4.0.0</modelVersion> <modelVersion>4.0.0</modelVersion>
<parent> <parent>
@ -38,9 +38,9 @@
<url>http://download.java.net/maven/2</url> <url>http://download.java.net/maven/2</url>
</repository> </repository>
</repositories> </repositories>
<dependencies> <dependencies>
<!-- activemq --> <!-- activemq -->
<dependency> <dependency>
<groupId>${project.groupId}</groupId> <groupId>${project.groupId}</groupId>
@ -95,7 +95,7 @@
<dependency> <dependency>
<groupId>${project.groupId}</groupId> <groupId>${project.groupId}</groupId>
<artifactId>activemq-spring</artifactId> <artifactId>activemq-spring</artifactId>
</dependency> </dependency>
<dependency> <dependency>
<groupId>${project.groupId}</groupId> <groupId>${project.groupId}</groupId>
<artifactId>activemq-web-demo</artifactId> <artifactId>activemq-web-demo</artifactId>
@ -120,7 +120,7 @@
<groupId>${project.groupId}</groupId> <groupId>${project.groupId}</groupId>
<artifactId>kahadb</artifactId> <artifactId>kahadb</artifactId>
</dependency> </dependency>
<!-- camel --> <!-- camel -->
<dependency> <dependency>
<groupId>org.apache.camel</groupId> <groupId>org.apache.camel</groupId>
@ -146,8 +146,8 @@
<!-- commons --> <!-- commons -->
<dependency> <dependency>
<groupId>commons-httpclient</groupId> <groupId>org.apache.httpcomponents</groupId>
<artifactId>commons-httpclient</artifactId> <artifactId>httpclient</artifactId>
</dependency> </dependency>
<dependency> <dependency>
<groupId>commons-beanutils</groupId> <groupId>commons-beanutils</groupId>
@ -163,7 +163,7 @@
<artifactId>commons-pool</artifactId> <artifactId>commons-pool</artifactId>
<optional>true</optional> <optional>true</optional>
</dependency> </dependency>
<dependency> <dependency>
<groupId>activesoap</groupId> <groupId>activesoap</groupId>
<artifactId>jaxp-api</artifactId> <artifactId>jaxp-api</artifactId>
@ -185,7 +185,7 @@
<groupId>log4j</groupId> <groupId>log4j</groupId>
<artifactId>log4j</artifactId> <artifactId>log4j</artifactId>
</dependency> </dependency>
<dependency> <dependency>
<groupId>org.apache.geronimo.specs</groupId> <groupId>org.apache.geronimo.specs</groupId>
<artifactId>geronimo-jms_1.1_spec</artifactId> <artifactId>geronimo-jms_1.1_spec</artifactId>
@ -222,7 +222,7 @@
<dependency> <dependency>
<groupId>org.springframework</groupId> <groupId>org.springframework</groupId>
<artifactId>spring-aop</artifactId> <artifactId>spring-aop</artifactId>
</dependency> </dependency>
<dependency> <dependency>
<groupId>org.springframework</groupId> <groupId>org.springframework</groupId>
<artifactId>spring-jms</artifactId> <artifactId>spring-jms</artifactId>
@ -247,7 +247,7 @@
<dependency> <dependency>
<groupId>org.codehaus.jettison</groupId> <groupId>org.codehaus.jettison</groupId>
<artifactId>jettison</artifactId> <artifactId>jettison</artifactId>
</dependency> </dependency>
<dependency> <dependency>
<groupId>org.eclipse.jetty.aggregate</groupId> <groupId>org.eclipse.jetty.aggregate</groupId>
@ -268,7 +268,7 @@
<artifactId>derbynet</artifactId> <artifactId>derbynet</artifactId>
<optional>true</optional> <optional>true</optional>
</dependency> </dependency>
<!-- copied dependencies from activemq-web-console --> <!-- copied dependencies from activemq-web-console -->
<!-- enable commons-logging when inside jetty6:run --> <!-- enable commons-logging when inside jetty6:run -->
<dependency> <dependency>
@ -309,7 +309,7 @@
<artifactId>junit</artifactId> <artifactId>junit</artifactId>
<scope>test</scope> <scope>test</scope>
</dependency> </dependency>
<!-- dependencies specific to this module --> <!-- dependencies specific to this module -->
<dependency> <dependency>
<groupId>jrms</groupId> <groupId>jrms</groupId>
@ -326,11 +326,11 @@
<artifactId>xmlParserAPIs</artifactId> <artifactId>xmlParserAPIs</artifactId>
<version>2.6.2</version> <version>2.6.2</version>
</dependency> </dependency>
</dependencies> </dependencies>
<build> <build>
<plugins> <plugins>
<plugin> <plugin>
<groupId>org.apache.maven.plugins</groupId> <groupId>org.apache.maven.plugins</groupId>
@ -379,7 +379,7 @@
</goals> </goals>
<configuration> <configuration>
<descriptors> <descriptors>
<descriptor>src/main/descriptors/unix-bin.xml</descriptor> <descriptor>src/main/descriptors/unix-bin.xml</descriptor>
</descriptors> </descriptors>
<!--<finalName>${pom.artifactId}-${project.version}</finalName>--> <!--<finalName>${pom.artifactId}-${project.version}</finalName>-->
<!--<appendAssemblyId>false</appendAssemblyId>--> <!--<appendAssemblyId>false</appendAssemblyId>-->
@ -393,7 +393,7 @@
</goals> </goals>
<configuration> <configuration>
<descriptors> <descriptors>
<descriptor>src/main/descriptors/windows-bin.xml</descriptor> <descriptor>src/main/descriptors/windows-bin.xml</descriptor>
</descriptors> </descriptors>
<!--<finalName>${pom.artifactId}-${project.version}</finalName>--> <!--<finalName>${pom.artifactId}-${project.version}</finalName>-->
<!--<appendAssemblyId>false</appendAssemblyId>--> <!--<appendAssemblyId>false</appendAssemblyId>-->
@ -422,15 +422,15 @@
</goals> </goals>
<configuration> <configuration>
<outputDirectory>target/conf</outputDirectory> <outputDirectory>target/conf</outputDirectory>
<resources> <resources>
<resource> <resource>
<directory>src/release/conf</directory> <directory>src/release/conf</directory>
</resource> </resource>
<resource> <resource>
<directory>src/sample-conf</directory> <directory>src/sample-conf</directory>
</resource> </resource>
</resources> </resources>
</configuration> </configuration>
</execution> </execution>
</executions> </executions>
</plugin> </plugin>

17
pom.xml
View File

@ -45,16 +45,16 @@
<aopalliance-version>1.0</aopalliance-version> <aopalliance-version>1.0</aopalliance-version>
<aries-version>0.2-incubating</aries-version> <aries-version>0.2-incubating</aries-version>
<axion-version>1.0-M3-dev</axion-version> <axion-version>1.0-M3-dev</axion-version>
<axis-version>1.4</axis-version> <axis-version>1.4.2</axis-version>
<camel-version>2.8.1</camel-version> <camel-version>2.8.1</camel-version>
<camel-web-version>2.4.0</camel-web-version> <camel-web-version>2.4.0</camel-web-version>
<cglib-version>2.0</cglib-version> <cglib-version>2.0</cglib-version>
<openjpa-version>1.2.0</openjpa-version> <openjpa-version>1.2.0</openjpa-version>
<commons-beanutils-version>1.8.3</commons-beanutils-version> <commons-beanutils-version>1.8.3</commons-beanutils-version>
<commons-collections-version>3.2.1</commons-collections-version> <commons-collections-version>3.2.1</commons-collections-version>
<commons-daemon-version>1.0.5</commons-daemon-version> <commons-daemon-version>1.0.7</commons-daemon-version>
<commons-dbcp-version>1.4</commons-dbcp-version> <commons-dbcp-version>1.4</commons-dbcp-version>
<commons-httpclient-version>3.1</commons-httpclient-version> <httpclient-version>4.1.2</httpclient-version>
<commons-io-version>1.4</commons-io-version> <commons-io-version>1.4</commons-io-version>
<commons-lang-version>2.6</commons-lang-version> <commons-lang-version>2.6</commons-lang-version>
<commons-logging-version>1.1.1</commons-logging-version> <commons-logging-version>1.1.1</commons-logging-version>
@ -64,6 +64,7 @@
<directory-version>1.5.7</directory-version> <directory-version>1.5.7</directory-version>
<geronimo-version>1.0</geronimo-version> <geronimo-version>1.0</geronimo-version>
<howl-version>0.1.8</howl-version> <howl-version>0.1.8</howl-version>
<hawtbuf-version>1.6</hawtbuf-version>
<hsqldb-version>1.7.2.2</hsqldb-version> <hsqldb-version>1.7.2.2</hsqldb-version>
<jasypt-version>1.8</jasypt-version> <jasypt-version>1.8</jasypt-version>
<jdom-version>1.0</jdom-version> <jdom-version>1.0</jdom-version>
@ -73,7 +74,7 @@
<jettison-version>1.3</jettison-version> <jettison-version>1.3</jettison-version>
<jmock-version>2.5.1</jmock-version> <jmock-version>2.5.1</jmock-version>
<josql-version>1.5</josql-version> <josql-version>1.5</josql-version>
<junit-version>4.8.2</junit-version> <junit-version>4.10</junit-version>
<jxta-version>2.0</jxta-version> <jxta-version>2.0</jxta-version>
<karaf-version>2.2.0</karaf-version> <karaf-version>2.2.0</karaf-version>
<log4j-version>1.2.16</log4j-version> <log4j-version>1.2.16</log4j-version>
@ -87,7 +88,7 @@
<xmlbeans-version>2.2.0</xmlbeans-version> <xmlbeans-version>2.2.0</xmlbeans-version>
<xpp3-version>1.1.4c</xpp3-version> <xpp3-version>1.1.4c</xpp3-version>
<xstream-version>1.4.1</xstream-version> <xstream-version>1.4.1</xstream-version>
<xbean-version>3.7</xbean-version> <xbean-version>3.8</xbean-version>
<velocity-version>1.6.4</velocity-version> <velocity-version>1.6.4</velocity-version>
<ftpserver-version>1.0.0</ftpserver-version> <ftpserver-version>1.0.0</ftpserver-version>
<activemq-protobuf-version>1.1</activemq-protobuf-version> <activemq-protobuf-version>1.1</activemq-protobuf-version>
@ -811,9 +812,9 @@
</dependency> </dependency>
<dependency> <dependency>
<groupId>commons-httpclient</groupId> <groupId>org.apache.httpcomponents</groupId>
<artifactId>commons-httpclient</artifactId> <artifactId>httpclient</artifactId>
<version>${commons-httpclient-version}</version> <version>${httpclient-version}</version>
</dependency> </dependency>
<dependency> <dependency>