You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ni...@apache.org on 2014/10/17 05:40:21 UTC
[1/8] git commit: CAMEL-7909 camel-netty-http consumer need to close
the connection if the response connection header is close
Repository: camel
Updated Branches:
refs/heads/camel-2.13.x f0e3e2015 -> f0ddb5ef7
refs/heads/camel-2.14.x 6eb1275ab -> 7f730a86b
CAMEL-7909 camel-netty-http consumer need to close the connection if the response connection header is close
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/14ba1b38
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/14ba1b38
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/14ba1b38
Branch: refs/heads/camel-2.14.x
Commit: 14ba1b38f468d12d2d6425b3e92421b47df08155
Parents: 6eb1275
Author: Willem Jiang <wi...@gmail.com>
Authored: Tue Oct 14 10:14:12 2014 +0800
Committer: Willem Jiang <wi...@gmail.com>
Committed: Fri Oct 17 11:25:22 2014 +0800
----------------------------------------------------------------------
.../netty/http/DefaultNettyHttpBinding.java | 5 +++++
.../http/handlers/HttpServerChannelHandler.java | 18 ++----------------
.../apache/camel/component/netty/NettyHelper.java | 2 +-
.../camel/component/netty/NettyProducer.java | 13 ++++++++++---
.../handlers/ServerResponseFutureListener.java | 5 +++++
5 files changed, 23 insertions(+), 20 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/14ba1b38/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/DefaultNettyHttpBinding.java
----------------------------------------------------------------------
diff --git a/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/DefaultNettyHttpBinding.java b/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/DefaultNettyHttpBinding.java
index eed39b0..102fb91 100644
--- a/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/DefaultNettyHttpBinding.java
+++ b/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/DefaultNettyHttpBinding.java
@@ -34,6 +34,7 @@ import org.apache.camel.Message;
import org.apache.camel.NoTypeConversionAvailableException;
import org.apache.camel.RuntimeCamelException;
import org.apache.camel.TypeConverter;
+import org.apache.camel.component.netty.NettyConstants;
import org.apache.camel.spi.HeaderFilterStrategy;
import org.apache.camel.util.ExchangeHelper;
import org.apache.camel.util.IOHelper;
@@ -411,6 +412,10 @@ public class DefaultNettyHttpBinding implements NettyHttpBinding, Cloneable {
}
}
response.headers().set(HttpHeaders.Names.CONNECTION, connection);
+ // Just make sure we close the channel when the connection value is close
+ if (connection.equalsIgnoreCase(HttpHeaders.Values.CLOSE)) {
+ message.setHeader(NettyConstants.NETTY_CLOSE_CHANNEL_WHEN_COMPLETE, true);
+ }
LOG.trace("Connection: {}", connection);
return response;
http://git-wip-us.apache.org/repos/asf/camel/blob/14ba1b38/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/handlers/HttpServerChannelHandler.java
----------------------------------------------------------------------
diff --git a/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/handlers/HttpServerChannelHandler.java b/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/handlers/HttpServerChannelHandler.java
index da17d25..cb92d44 100644
--- a/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/handlers/HttpServerChannelHandler.java
+++ b/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/handlers/HttpServerChannelHandler.java
@@ -16,7 +16,6 @@
*/
package org.apache.camel.component.netty.http.handlers;
-import java.net.SocketAddress;
import java.net.URI;
import java.nio.channels.ClosedChannelException;
import java.nio.charset.Charset;
@@ -27,7 +26,6 @@ import javax.security.auth.login.LoginException;
import org.apache.camel.Exchange;
import org.apache.camel.LoggingLevel;
-import org.apache.camel.component.netty.NettyConsumer;
import org.apache.camel.component.netty.NettyHelper;
import org.apache.camel.component.netty.handlers.ServerChannelHandler;
import org.apache.camel.component.netty.http.HttpPrincipal;
@@ -38,7 +36,6 @@ import org.apache.camel.util.CamelLogger;
import org.apache.camel.util.ObjectHelper;
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBuffers;
-import org.jboss.netty.channel.ChannelFutureListener;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
@@ -50,7 +47,6 @@ import org.jboss.netty.handler.codec.http.HttpResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import static org.jboss.netty.handler.codec.http.HttpHeaders.isKeepAlive;
import static org.jboss.netty.handler.codec.http.HttpResponseStatus.BAD_REQUEST;
import static org.jboss.netty.handler.codec.http.HttpResponseStatus.METHOD_NOT_ALLOWED;
import static org.jboss.netty.handler.codec.http.HttpResponseStatus.OK;
@@ -58,6 +54,7 @@ import static org.jboss.netty.handler.codec.http.HttpResponseStatus.SERVICE_UNAV
import static org.jboss.netty.handler.codec.http.HttpResponseStatus.UNAUTHORIZED;
import static org.jboss.netty.handler.codec.http.HttpVersion.HTTP_1_1;
+
/**
* Netty HTTP {@link ServerChannelHandler} that handles the incoming HTTP requests and routes
* the received message in Camel.
@@ -303,18 +300,7 @@ public class HttpServerChannelHandler extends ServerChannelHandler {
}
}
}
-
- @Override
- protected ChannelFutureListener createResponseFutureListener(NettyConsumer consumer, Exchange exchange, SocketAddress remoteAddress) {
- // make sure to close channel if not keep-alive
- if (request != null && isKeepAlive(request)) {
- LOG.trace("Request has Connection: keep-alive so Channel is not being closed");
- return null;
- } else {
- LOG.trace("Request is not Connection: close so Channel is being closed");
- return ChannelFutureListener.CLOSE;
- }
- }
+
@Override
protected Object getResponseBody(Exchange exchange) throws Exception {
http://git-wip-us.apache.org/repos/asf/camel/blob/14ba1b38/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyHelper.java
----------------------------------------------------------------------
diff --git a/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyHelper.java b/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyHelper.java
index 05f3e4d..b9368fa 100644
--- a/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyHelper.java
+++ b/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyHelper.java
@@ -111,7 +111,7 @@ public final class NettyHelper {
public static void close(Channel channel) {
if (channel != null) {
LOG.trace("Closing channel: {}", channel);
- channel.close();
+ channel.close().syncUninterruptibly();
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/14ba1b38/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyProducer.java b/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyProducer.java
index 50de736..87ad2be 100644
--- a/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyProducer.java
+++ b/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyProducer.java
@@ -498,8 +498,11 @@ public class NettyProducer extends DefaultAsyncProducer {
public void done(boolean doneSync) {
// put back in pool
try {
- LOG.trace("Putting channel back to pool {}", channel);
- pool.returnObject(channel);
+ // Only put the connected channel back to the pool
+ if (channel.isConnected()) {
+ LOG.trace("Putting channel back to pool {}", channel);
+ pool.returnObject(channel);
+ }
} catch (Exception e) {
LOG.warn("Error returning channel to pool {}. This exception will be ignored.", channel);
} finally {
@@ -525,7 +528,9 @@ public class NettyProducer extends DefaultAsyncProducer {
@Override
public void destroyObject(Channel channel) throws Exception {
LOG.trace("Destroying channel: {}", channel);
- NettyHelper.close(channel);
+ if (channel.isOpen()) {
+ NettyHelper.close(channel);
+ }
allChannels.remove(channel);
}
@@ -540,11 +545,13 @@ public class NettyProducer extends DefaultAsyncProducer {
@Override
public void activateObject(Channel channel) throws Exception {
// noop
+ LOG.trace("activateObject channel: {} -> {}", channel);
}
@Override
public void passivateObject(Channel channel) throws Exception {
// noop
+ LOG.trace("passivateObject channel: {} -> {}", channel);
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/14ba1b38/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ServerResponseFutureListener.java
----------------------------------------------------------------------
diff --git a/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ServerResponseFutureListener.java b/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ServerResponseFutureListener.java
index 619a62e..90dc9e6 100644
--- a/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ServerResponseFutureListener.java
+++ b/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ServerResponseFutureListener.java
@@ -61,6 +61,11 @@ public class ServerResponseFutureListener implements ChannelFutureListener {
} else {
close = exchange.getIn().getHeader(NettyConstants.NETTY_CLOSE_CHANNEL_WHEN_COMPLETE, Boolean.class);
}
+
+ // check the setting on the exchange property
+ if (close == null) {
+ close = exchange.getProperty(NettyConstants.NETTY_CLOSE_CHANNEL_WHEN_COMPLETE, Boolean.class);
+ }
// should we disconnect, the header can override the configuration
boolean disconnect = consumer.getConfiguration().isDisconnect();
[4/8] git commit: Removed the fork per test setting of camel-netty4
and camel-netty4-http
Posted by ni...@apache.org.
Removed the fork per test setting of camel-netty4 and camel-netty4-http
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/48823ddf
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/48823ddf
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/48823ddf
Branch: refs/heads/camel-2.14.x
Commit: 48823ddf1ab4f47a04fb461a2bb70978ac6efaef
Parents: 64389b1
Author: Willem Jiang <wi...@gmail.com>
Authored: Tue Oct 14 10:41:09 2014 +0800
Committer: Willem Jiang <wi...@gmail.com>
Committed: Fri Oct 17 11:26:25 2014 +0800
----------------------------------------------------------------------
components/camel-netty4-http/pom.xml | 12 +-----------
components/camel-netty4/pom.xml | 12 ------------
2 files changed, 1 insertion(+), 23 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/48823ddf/components/camel-netty4-http/pom.xml
----------------------------------------------------------------------
diff --git a/components/camel-netty4-http/pom.xml b/components/camel-netty4-http/pom.xml
index 58074e8..bd44ef0 100644
--- a/components/camel-netty4-http/pom.xml
+++ b/components/camel-netty4-http/pom.xml
@@ -81,15 +81,5 @@
<scope>test</scope>
</dependency>
</dependencies>
- <build>
- <plugins>
- <plugin>
- <artifactId>maven-surefire-plugin</artifactId>
- <configuration>
- <forkMode>perTest</forkMode>
- </configuration>
- </plugin>
- </plugins>
- </build>
-
+
</project>
http://git-wip-us.apache.org/repos/asf/camel/blob/48823ddf/components/camel-netty4/pom.xml
----------------------------------------------------------------------
diff --git a/components/camel-netty4/pom.xml b/components/camel-netty4/pom.xml
index 39dd5a0..179866b 100644
--- a/components/camel-netty4/pom.xml
+++ b/components/camel-netty4/pom.xml
@@ -90,17 +90,5 @@
<scope>test</scope>
</dependency>
</dependencies>
-
- <build>
- <plugins>
- <plugin>
- <artifactId>maven-surefire-plugin</artifactId>
- <configuration>
- <forkMode>perTest</forkMode>
- <enableAssertions>false</enableAssertions>
- </configuration>
- </plugin>
- </plugins>
- </build>
</project>
[7/8] git commit: CAMEL-7920 Make sure the JAXB fallback converter is
the last one loaded
Posted by ni...@apache.org.
CAMEL-7920 Make sure the JAXB fallback converter is the last one loaded
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/7f730a86
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/7f730a86
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/7f730a86
Branch: refs/heads/camel-2.14.x
Commit: 7f730a86bf4f471d32b403aef87c1e8737096ca8
Parents: 48823dd
Author: Willem Jiang <wi...@gmail.com>
Authored: Fri Oct 17 11:07:26 2014 +0800
Committer: Willem Jiang <wi...@gmail.com>
Committed: Fri Oct 17 11:39:01 2014 +0800
----------------------------------------------------------------------
.../converter/BaseTypeConverterRegistry.java | 1 +
.../org/apache/camel/impl/osgi/Activator.java | 18 ++++++++---
.../camel/core/osgi/OsgiTypeConverter.java | 33 ++++++++++++++------
3 files changed, 38 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/7f730a86/camel-core/src/main/java/org/apache/camel/impl/converter/BaseTypeConverterRegistry.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/impl/converter/BaseTypeConverterRegistry.java b/camel-core/src/main/java/org/apache/camel/impl/converter/BaseTypeConverterRegistry.java
index af4e1b3..962c415 100644
--- a/camel-core/src/main/java/org/apache/camel/impl/converter/BaseTypeConverterRegistry.java
+++ b/camel-core/src/main/java/org/apache/camel/impl/converter/BaseTypeConverterRegistry.java
@@ -390,6 +390,7 @@ public abstract class BaseTypeConverterRegistry extends ServiceSupport implement
log.trace("Adding fallback type converter: {} which can promote: {}", typeConverter, canPromote);
// add in top of fallback as the toString() fallback will nearly always be able to convert
+ // the last one which is add to the FallbackTypeConverter will be called at the first place
fallbackConverters.add(0, new FallbackTypeConverter(typeConverter, canPromote));
if (typeConverter instanceof TypeConverterAware) {
TypeConverterAware typeConverterAware = (TypeConverterAware) typeConverter;
http://git-wip-us.apache.org/repos/asf/camel/blob/7f730a86/camel-core/src/main/java/org/apache/camel/impl/osgi/Activator.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/impl/osgi/Activator.java b/camel-core/src/main/java/org/apache/camel/impl/osgi/Activator.java
index 9502a1d..9900207 100644
--- a/camel-core/src/main/java/org/apache/camel/impl/osgi/Activator.java
+++ b/camel-core/src/main/java/org/apache/camel/impl/osgi/Activator.java
@@ -61,8 +61,8 @@ import org.osgi.framework.Bundle;
import org.osgi.framework.BundleActivator;
import org.osgi.framework.BundleContext;
import org.osgi.framework.BundleEvent;
+import org.osgi.framework.Constants;
import org.osgi.framework.ServiceRegistration;
-
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -176,7 +176,8 @@ public class Activator implements BundleActivator, BundleTrackerCustomizer {
URL url1 = bundle.getEntry(META_INF_TYPE_CONVERTER);
URL url2 = bundle.getEntry(META_INF_FALLBACK_TYPE_CONVERTER);
if (url1 != null || url2 != null) {
- resolvers.add(new BundleTypeConverterLoader(bundle));
+ LOG.debug("Found TypeConverter in bundle {}", bundle.getSymbolicName());
+ resolvers.add(new BundleTypeConverterLoader(bundle, url2 != null));
}
}
}
@@ -264,11 +265,13 @@ public class Activator implements BundleActivator, BundleTrackerCustomizer {
private final AnnotationTypeConverterLoader loader = new Loader();
private final Bundle bundle;
+ private final boolean hasFallbackTypeConverter;
- public BundleTypeConverterLoader(Bundle bundle) {
+ public BundleTypeConverterLoader(Bundle bundle, boolean hasFallbackTypeConverter) {
super(bundle, TypeConverter.class);
ObjectHelper.notNull(bundle, "bundle");
this.bundle = bundle;
+ this.hasFallbackTypeConverter = hasFallbackTypeConverter;
}
public synchronized void load(TypeConverterRegistry registry) throws TypeConverterLoaderException {
@@ -282,7 +285,13 @@ public class Activator implements BundleActivator, BundleTrackerCustomizer {
}
public void register() {
- doRegister(TypeConverterLoader.class);
+ if (hasFallbackTypeConverter) {
+ // The FallbackTypeConverter should have a higher ranking
+ doRegister(TypeConverterLoader.class, Constants.SERVICE_RANKING, new Integer(100));
+ } else {
+ // The default service ranking is Integer(0);
+ doRegister(TypeConverterLoader.class);
+ }
}
class Loader extends AnnotationTypeConverterLoader {
@@ -360,6 +369,7 @@ public class Activator implements BundleActivator, BundleTrackerCustomizer {
// register fallback converters
URL fallbackUrl = bundle.getEntry(META_INF_FALLBACK_TYPE_CONVERTER);
if (fallbackUrl != null) {
+ LOG.debug("Found {} to load the FallbackTypeConverter", META_INF_FALLBACK_TYPE_CONVERTER);
TypeConverter tc = createInstance("FallbackTypeConverter", fallbackUrl, registry.getInjector());
registry.addFallbackTypeConverter(tc, false);
}
http://git-wip-us.apache.org/repos/asf/camel/blob/7f730a86/components/camel-core-osgi/src/main/java/org/apache/camel/core/osgi/OsgiTypeConverter.java
----------------------------------------------------------------------
diff --git a/components/camel-core-osgi/src/main/java/org/apache/camel/core/osgi/OsgiTypeConverter.java b/components/camel-core-osgi/src/main/java/org/apache/camel/core/osgi/OsgiTypeConverter.java
index 8bd1c00..0c3ab78 100644
--- a/components/camel-core-osgi/src/main/java/org/apache/camel/core/osgi/OsgiTypeConverter.java
+++ b/components/camel-core-osgi/src/main/java/org/apache/camel/core/osgi/OsgiTypeConverter.java
@@ -16,6 +16,8 @@
*/
package org.apache.camel.core.osgi;
+import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Set;
@@ -60,12 +62,16 @@ public class OsgiTypeConverter extends ServiceSupport implements TypeConverter,
public Object addingService(ServiceReference<TypeConverterLoader> serviceReference) {
LOG.trace("AddingService: {}", serviceReference);
TypeConverterLoader loader = bundleContext.getService(serviceReference);
- if (loader != null) {
+ // just make sure we don't load the bundle converter this time
+ if (delegate != null) {
try {
- loader.load(getDelegate());
- } catch (Throwable t) {
- throw ObjectHelper.wrapRuntimeCamelException(t);
+ ServiceHelper.stopService(this.delegate);
+ } catch (Exception e) {
+ // ignore
+ LOG.debug("Error stopping service due: " + e.getMessage() + ". This exception will be ignored.", e);
}
+ // It can force camel to reload the type converter again
+ this.delegate = null;
}
return loader;
}
@@ -81,6 +87,7 @@ public class OsgiTypeConverter extends ServiceSupport implements TypeConverter,
// ignore
LOG.debug("Error stopping service due: " + e.getMessage() + ". This exception will be ignored.", e);
}
+ // It can force camel to reload the type converter again
this.delegate = null;
}
@@ -185,14 +192,20 @@ public class OsgiTypeConverter extends ServiceSupport implements TypeConverter,
throw new RuntimeCamelException("Error loading CoreTypeConverter due: " + e.getMessage(), e);
}
- // load the type converters the tracker has been tracking
- Object[] services = this.tracker.getServices();
- if (services != null) {
- for (Object o : services) {
+ // Load the type converters the tracker has been tracking
+ // Here we need to use the ServiceReference to check the ranking
+ ServiceReference<TypeConverterLoader>[] serviceReferences = this.tracker.getServiceReferences();
+ if (serviceReferences != null) {
+ ArrayList<ServiceReference<TypeConverterLoader>> servicesList =
+ new ArrayList<ServiceReference<TypeConverterLoader>>(Arrays.asList(serviceReferences));
+ // Just make sure we install the high ranking fallback converter at last
+ Collections.sort(servicesList);
+ for (ServiceReference<TypeConverterLoader> sr : servicesList) {
try {
- ((TypeConverterLoader) o).load(answer);
+ LOG.debug("loading the type converter from bundle{} ", sr.getBundle().getSymbolicName());
+ ((TypeConverterLoader)this.tracker.getService(sr)).load(answer);
} catch (Throwable t) {
- throw new RuntimeCamelException("Error loading type converters from service: " + o + " due: " + t.getMessage(), t);
+ throw new RuntimeCamelException("Error loading type converters from service: " + sr + " due: " + t.getMessage(), t);
}
}
}
[3/8] git commit: CAMEL-7909 Merged the patch into camel-netty4-http
module
Posted by ni...@apache.org.
CAMEL-7909 Merged the patch into camel-netty4-http module
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/64389b1e
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/64389b1e
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/64389b1e
Branch: refs/heads/camel-2.14.x
Commit: 64389b1eaae59c9c1365b83452bf09f42a900744
Parents: a733210
Author: Willem Jiang <wi...@gmail.com>
Authored: Tue Oct 14 10:39:33 2014 +0800
Committer: Willem Jiang <wi...@gmail.com>
Committed: Fri Oct 17 11:25:57 2014 +0800
----------------------------------------------------------------------
.../component/netty4/http/DefaultNettyHttpBinding.java | 5 +++++
.../netty4/http/handlers/HttpServerChannelHandler.java | 12 ------------
.../org/apache/camel/component/netty4/NettyHelper.java | 2 +-
.../apache/camel/component/netty4/NettyProducer.java | 13 ++++++++++---
.../netty4/handlers/ServerResponseFutureListener.java | 5 +++++
5 files changed, 21 insertions(+), 16 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/64389b1e/components/camel-netty4-http/src/main/java/org/apache/camel/component/netty4/http/DefaultNettyHttpBinding.java
----------------------------------------------------------------------
diff --git a/components/camel-netty4-http/src/main/java/org/apache/camel/component/netty4/http/DefaultNettyHttpBinding.java b/components/camel-netty4-http/src/main/java/org/apache/camel/component/netty4/http/DefaultNettyHttpBinding.java
index dc1c8d3..ceb64c4 100644
--- a/components/camel-netty4-http/src/main/java/org/apache/camel/component/netty4/http/DefaultNettyHttpBinding.java
+++ b/components/camel-netty4-http/src/main/java/org/apache/camel/component/netty4/http/DefaultNettyHttpBinding.java
@@ -45,6 +45,7 @@ import org.apache.camel.Message;
import org.apache.camel.NoTypeConversionAvailableException;
import org.apache.camel.RuntimeCamelException;
import org.apache.camel.TypeConverter;
+import org.apache.camel.component.netty4.NettyConstants;
import org.apache.camel.component.netty4.NettyConverter;
import org.apache.camel.spi.HeaderFilterStrategy;
import org.apache.camel.util.ExchangeHelper;
@@ -418,6 +419,10 @@ public class DefaultNettyHttpBinding implements NettyHttpBinding, Cloneable {
}
}
response.headers().set(HttpHeaders.Names.CONNECTION, connection);
+ // Just make sure we close the channel when the connection value is close
+ if (connection.equalsIgnoreCase(HttpHeaders.Values.CLOSE)) {
+ message.setHeader(NettyConstants.NETTY_CLOSE_CHANNEL_WHEN_COMPLETE, true);
+ }
LOG.trace("Connection: {}", connection);
return response;
http://git-wip-us.apache.org/repos/asf/camel/blob/64389b1e/components/camel-netty4-http/src/main/java/org/apache/camel/component/netty4/http/handlers/HttpServerChannelHandler.java
----------------------------------------------------------------------
diff --git a/components/camel-netty4-http/src/main/java/org/apache/camel/component/netty4/http/handlers/HttpServerChannelHandler.java b/components/camel-netty4-http/src/main/java/org/apache/camel/component/netty4/http/handlers/HttpServerChannelHandler.java
index eb33c4c..7e72a86 100644
--- a/components/camel-netty4-http/src/main/java/org/apache/camel/component/netty4/http/handlers/HttpServerChannelHandler.java
+++ b/components/camel-netty4-http/src/main/java/org/apache/camel/component/netty4/http/handlers/HttpServerChannelHandler.java
@@ -295,18 +295,6 @@ public class HttpServerChannelHandler extends ServerChannelHandler {
}
@Override
- protected ChannelFutureListener createResponseFutureListener(NettyConsumer consumer, Exchange exchange, SocketAddress remoteAddress) {
- // make sure to close channel if not keep-alive
- if (request != null && isKeepAlive(request)) {
- LOG.trace("Request has Connection: keep-alive so Channel is not being closed");
- return null;
- } else {
- LOG.trace("Request is not Connection: close so Channel is being closed");
- return ChannelFutureListener.CLOSE;
- }
- }
-
- @Override
protected Object getResponseBody(Exchange exchange) throws Exception {
// use the binding
if (exchange.hasOut()) {
http://git-wip-us.apache.org/repos/asf/camel/blob/64389b1e/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyHelper.java
----------------------------------------------------------------------
diff --git a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyHelper.java b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyHelper.java
index b345cca..5a25928 100644
--- a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyHelper.java
+++ b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyHelper.java
@@ -118,7 +118,7 @@ public final class NettyHelper {
public static void close(Channel channel) {
if (channel != null) {
LOG.trace("Closing channel: {}", channel);
- channel.close();
+ channel.close().syncUninterruptibly();
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/64389b1e/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyProducer.java b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyProducer.java
index 7f54990..11f8b55 100644
--- a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyProducer.java
+++ b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyProducer.java
@@ -464,8 +464,11 @@ public class NettyProducer extends DefaultAsyncProducer {
public void done(boolean doneSync) {
// put back in pool
try {
- LOG.trace("Putting channel back to pool {}", channel);
- pool.returnObject(channel);
+ // Only put the connected channel back to the pool
+ if (channel.isActive()) {
+ LOG.trace("Putting channel back to pool {}", channel);
+ pool.returnObject(channel);
+ }
} catch (Exception e) {
LOG.warn("Error returning channel to pool {}. This exception will be ignored.", channel);
} finally {
@@ -491,7 +494,9 @@ public class NettyProducer extends DefaultAsyncProducer {
@Override
public void destroyObject(Channel channel) throws Exception {
LOG.trace("Destroying channel: {}", channel);
- NettyHelper.close(channel);
+ if (channel.isOpen()) {
+ NettyHelper.close(channel);
+ }
allChannels.remove(channel);
}
@@ -506,11 +511,13 @@ public class NettyProducer extends DefaultAsyncProducer {
@Override
public void activateObject(Channel channel) throws Exception {
// noop
+ LOG.trace("activateObject channel: {} -> {}", channel);
}
@Override
public void passivateObject(Channel channel) throws Exception {
// noop
+ LOG.trace("passivateObject channel: {} -> {}", channel);
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/64389b1e/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/handlers/ServerResponseFutureListener.java
----------------------------------------------------------------------
diff --git a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/handlers/ServerResponseFutureListener.java b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/handlers/ServerResponseFutureListener.java
index 759b69d..3d33e62 100644
--- a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/handlers/ServerResponseFutureListener.java
+++ b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/handlers/ServerResponseFutureListener.java
@@ -62,6 +62,11 @@ public class ServerResponseFutureListener implements ChannelFutureListener {
} else {
close = exchange.getIn().getHeader(NettyConstants.NETTY_CLOSE_CHANNEL_WHEN_COMPLETE, Boolean.class);
}
+
+ // check the setting on the exchange property
+ if (close == null) {
+ close = exchange.getProperty(NettyConstants.NETTY_CLOSE_CHANNEL_WHEN_COMPLETE, Boolean.class);
+ }
// should we disconnect, the header can override the configuration
boolean disconnect = consumer.getConfiguration().isDisconnect();
[2/8] git commit: CAMEL-7910 Netty {Client|Server}ChannelHandler need
to pass the close and open event around
Posted by ni...@apache.org.
CAMEL-7910 Netty {Client|Server}ChannelHandler need to pass the close and open event around
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/a7332105
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/a7332105
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/a7332105
Branch: refs/heads/camel-2.14.x
Commit: a73321054389e43014bcf9c559a99693b5e41ad3
Parents: 14ba1b3
Author: Willem Jiang <wi...@gmail.com>
Authored: Tue Oct 14 10:24:46 2014 +0800
Committer: Willem Jiang <wi...@gmail.com>
Committed: Fri Oct 17 11:25:34 2014 +0800
----------------------------------------------------------------------
.../camel/component/netty/handlers/ClientChannelHandler.java | 4 ++++
.../camel/component/netty/handlers/ServerChannelHandler.java | 4 ++++
2 files changed, 8 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/a7332105/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ClientChannelHandler.java
----------------------------------------------------------------------
diff --git a/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ClientChannelHandler.java b/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ClientChannelHandler.java
index ea9d578..7a6d31f 100644
--- a/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ClientChannelHandler.java
+++ b/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ClientChannelHandler.java
@@ -56,6 +56,8 @@ public class ClientChannelHandler extends SimpleChannelUpstreamHandler {
}
// to keep track of open sockets
producer.getAllChannels().add(channelStateEvent.getChannel());
+ // make sure the event can be processed by other handlers
+ super.channelOpen(ctx, channelStateEvent);
}
@Override
@@ -119,6 +121,8 @@ public class ClientChannelHandler extends SimpleChannelUpstreamHandler {
// signal callback
callback.done(false);
}
+ // make sure the event can be processed by other handlers
+ super.channelClosed(ctx, e);
}
@Override
http://git-wip-us.apache.org/repos/asf/camel/blob/a7332105/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ServerChannelHandler.java
----------------------------------------------------------------------
diff --git a/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ServerChannelHandler.java b/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ServerChannelHandler.java
index 4554730..afab4c9 100644
--- a/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ServerChannelHandler.java
+++ b/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ServerChannelHandler.java
@@ -56,6 +56,8 @@ public class ServerChannelHandler extends SimpleChannelUpstreamHandler {
}
// to keep track of open sockets
consumer.getNettyServerBootstrapFactory().addChannel(e.getChannel());
+ // make sure the event can be processed by other handlers
+ super.channelOpen(ctx, e);
}
@Override
@@ -65,6 +67,8 @@ public class ServerChannelHandler extends SimpleChannelUpstreamHandler {
}
// to keep track of open sockets
consumer.getNettyServerBootstrapFactory().removeChannel(e.getChannel());
+ // make sure the event can be processed by other handlers
+ super.channelClosed(ctx, e);
}
@Override
[6/8] git commit: CAMEL-7910 Netty {Client|Server}ChannelHandler need
to pass the close and open event around
Posted by ni...@apache.org.
CAMEL-7910 Netty {Client|Server}ChannelHandler need to pass the close and open event around
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/b41b8d05
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/b41b8d05
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/b41b8d05
Branch: refs/heads/camel-2.13.x
Commit: b41b8d05adcb3bd3bac8fd8948bef128f646d42a
Parents: 9cf11bd
Author: Willem Jiang <wi...@gmail.com>
Authored: Tue Oct 14 10:24:46 2014 +0800
Committer: Willem Jiang <wi...@gmail.com>
Committed: Fri Oct 17 11:36:33 2014 +0800
----------------------------------------------------------------------
.../camel/component/netty/handlers/ClientChannelHandler.java | 4 ++++
.../camel/component/netty/handlers/ServerChannelHandler.java | 4 ++++
2 files changed, 8 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/b41b8d05/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ClientChannelHandler.java
----------------------------------------------------------------------
diff --git a/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ClientChannelHandler.java b/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ClientChannelHandler.java
index ea9d578..7a6d31f 100644
--- a/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ClientChannelHandler.java
+++ b/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ClientChannelHandler.java
@@ -56,6 +56,8 @@ public class ClientChannelHandler extends SimpleChannelUpstreamHandler {
}
// to keep track of open sockets
producer.getAllChannels().add(channelStateEvent.getChannel());
+ // make sure the event can be processed by other handlers
+ super.channelOpen(ctx, channelStateEvent);
}
@Override
@@ -119,6 +121,8 @@ public class ClientChannelHandler extends SimpleChannelUpstreamHandler {
// signal callback
callback.done(false);
}
+ // make sure the event can be processed by other handlers
+ super.channelClosed(ctx, e);
}
@Override
http://git-wip-us.apache.org/repos/asf/camel/blob/b41b8d05/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ServerChannelHandler.java
----------------------------------------------------------------------
diff --git a/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ServerChannelHandler.java b/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ServerChannelHandler.java
index 6bafe7b..f966dee 100644
--- a/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ServerChannelHandler.java
+++ b/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ServerChannelHandler.java
@@ -56,6 +56,8 @@ public class ServerChannelHandler extends SimpleChannelUpstreamHandler {
}
// to keep track of open sockets
consumer.getNettyServerBootstrapFactory().addChannel(e.getChannel());
+ // make sure the event can be processed by other handlers
+ super.channelOpen(ctx, e);
}
@Override
@@ -65,6 +67,8 @@ public class ServerChannelHandler extends SimpleChannelUpstreamHandler {
}
// to keep track of open sockets
consumer.getNettyServerBootstrapFactory().removeChannel(e.getChannel());
+ // make sure the event can be processed by other handlers
+ super.channelClosed(ctx, e);
}
@Override
[8/8] git commit: CAMEL-7920 Make sure the JAXB fallback converter is
the last one loaded
Posted by ni...@apache.org.
CAMEL-7920 Make sure the JAXB fallback converter is the last one loaded
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/f0ddb5ef
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/f0ddb5ef
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/f0ddb5ef
Branch: refs/heads/camel-2.13.x
Commit: f0ddb5ef76a37bd0b7715ed6ee28165d814f2d95
Parents: b41b8d0
Author: Willem Jiang <wi...@gmail.com>
Authored: Fri Oct 17 11:07:26 2014 +0800
Committer: Willem Jiang <wi...@gmail.com>
Committed: Fri Oct 17 11:39:26 2014 +0800
----------------------------------------------------------------------
.../converter/BaseTypeConverterRegistry.java | 1 +
.../org/apache/camel/impl/osgi/Activator.java | 18 ++++++++---
.../camel/core/osgi/OsgiTypeConverter.java | 33 ++++++++++++++------
3 files changed, 38 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/f0ddb5ef/camel-core/src/main/java/org/apache/camel/impl/converter/BaseTypeConverterRegistry.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/impl/converter/BaseTypeConverterRegistry.java b/camel-core/src/main/java/org/apache/camel/impl/converter/BaseTypeConverterRegistry.java
index 5443e75..a71d46e 100644
--- a/camel-core/src/main/java/org/apache/camel/impl/converter/BaseTypeConverterRegistry.java
+++ b/camel-core/src/main/java/org/apache/camel/impl/converter/BaseTypeConverterRegistry.java
@@ -389,6 +389,7 @@ public abstract class BaseTypeConverterRegistry extends ServiceSupport implement
log.trace("Adding fallback type converter: {} which can promote: {}", typeConverter, canPromote);
// add in top of fallback as the toString() fallback will nearly always be able to convert
+ // the last one which is add to the FallbackTypeConverter will be called at the first place
fallbackConverters.add(0, new FallbackTypeConverter(typeConverter, canPromote));
if (typeConverter instanceof TypeConverterAware) {
TypeConverterAware typeConverterAware = (TypeConverterAware) typeConverter;
http://git-wip-us.apache.org/repos/asf/camel/blob/f0ddb5ef/camel-core/src/main/java/org/apache/camel/impl/osgi/Activator.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/impl/osgi/Activator.java b/camel-core/src/main/java/org/apache/camel/impl/osgi/Activator.java
index 9be601b..36d8249 100644
--- a/camel-core/src/main/java/org/apache/camel/impl/osgi/Activator.java
+++ b/camel-core/src/main/java/org/apache/camel/impl/osgi/Activator.java
@@ -61,8 +61,8 @@ import org.osgi.framework.Bundle;
import org.osgi.framework.BundleActivator;
import org.osgi.framework.BundleContext;
import org.osgi.framework.BundleEvent;
+import org.osgi.framework.Constants;
import org.osgi.framework.ServiceRegistration;
-
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -176,7 +176,8 @@ public class Activator implements BundleActivator, BundleTrackerCustomizer {
URL url1 = bundle.getEntry(META_INF_TYPE_CONVERTER);
URL url2 = bundle.getEntry(META_INF_FALLBACK_TYPE_CONVERTER);
if (url1 != null || url2 != null) {
- resolvers.add(new BundleTypeConverterLoader(bundle));
+ LOG.debug("Found TypeConverter in bundle {}", bundle.getSymbolicName());
+ resolvers.add(new BundleTypeConverterLoader(bundle, url2 != null));
}
}
}
@@ -264,11 +265,13 @@ public class Activator implements BundleActivator, BundleTrackerCustomizer {
private final AnnotationTypeConverterLoader loader = new Loader();
private final Bundle bundle;
+ private final boolean hasFallbackTypeConverter;
- public BundleTypeConverterLoader(Bundle bundle) {
+ public BundleTypeConverterLoader(Bundle bundle, boolean hasFallbackTypeConverter) {
super(bundle, TypeConverter.class);
ObjectHelper.notNull(bundle, "bundle");
this.bundle = bundle;
+ this.hasFallbackTypeConverter = hasFallbackTypeConverter;
}
public synchronized void load(TypeConverterRegistry registry) throws TypeConverterLoaderException {
@@ -282,7 +285,13 @@ public class Activator implements BundleActivator, BundleTrackerCustomizer {
}
public void register() {
- doRegister(TypeConverterLoader.class);
+ if (hasFallbackTypeConverter) {
+ // The FallbackTypeConverter should have a higher ranking
+ doRegister(TypeConverterLoader.class, Constants.SERVICE_RANKING, new Integer(100));
+ } else {
+ // The default service ranking is Integer(0);
+ doRegister(TypeConverterLoader.class);
+ }
}
class Loader extends AnnotationTypeConverterLoader {
@@ -360,6 +369,7 @@ public class Activator implements BundleActivator, BundleTrackerCustomizer {
// register fallback converters
URL fallbackUrl = bundle.getEntry(META_INF_FALLBACK_TYPE_CONVERTER);
if (fallbackUrl != null) {
+ LOG.debug("Found {} to load the FallbackTypeConverter", META_INF_FALLBACK_TYPE_CONVERTER);
TypeConverter tc = createInstance("FallbackTypeConverter", fallbackUrl, registry.getInjector());
registry.addFallbackTypeConverter(tc, false);
}
http://git-wip-us.apache.org/repos/asf/camel/blob/f0ddb5ef/components/camel-core-osgi/src/main/java/org/apache/camel/core/osgi/OsgiTypeConverter.java
----------------------------------------------------------------------
diff --git a/components/camel-core-osgi/src/main/java/org/apache/camel/core/osgi/OsgiTypeConverter.java b/components/camel-core-osgi/src/main/java/org/apache/camel/core/osgi/OsgiTypeConverter.java
index e5452fe..1ac5fa7 100644
--- a/components/camel-core-osgi/src/main/java/org/apache/camel/core/osgi/OsgiTypeConverter.java
+++ b/components/camel-core-osgi/src/main/java/org/apache/camel/core/osgi/OsgiTypeConverter.java
@@ -16,6 +16,8 @@
*/
package org.apache.camel.core.osgi;
+import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Set;
@@ -60,12 +62,16 @@ public class OsgiTypeConverter extends ServiceSupport implements TypeConverter,
public Object addingService(ServiceReference<TypeConverterLoader> serviceReference) {
LOG.trace("AddingService: {}", serviceReference);
TypeConverterLoader loader = bundleContext.getService(serviceReference);
- if (loader != null) {
+ // just make sure we don't load the bundle converter this time
+ if (delegate != null) {
try {
- loader.load(getDelegate());
- } catch (Throwable t) {
- throw ObjectHelper.wrapRuntimeCamelException(t);
+ ServiceHelper.stopService(this.delegate);
+ } catch (Exception e) {
+ // ignore
+ LOG.debug("Error stopping service due: " + e.getMessage() + ". This exception will be ignored.", e);
}
+ // It can force camel to reload the type converter again
+ this.delegate = null;
}
return loader;
}
@@ -81,6 +87,7 @@ public class OsgiTypeConverter extends ServiceSupport implements TypeConverter,
// ignore
LOG.debug("Error stopping service due: " + e.getMessage() + ". This exception will be ignored.", e);
}
+ // It can force camel to reload the type converter again
this.delegate = null;
}
@@ -185,14 +192,20 @@ public class OsgiTypeConverter extends ServiceSupport implements TypeConverter,
throw new RuntimeCamelException("Error loading CoreTypeConverter due: " + e.getMessage(), e);
}
- // load the type converters the tracker has been tracking
- Object[] services = this.tracker.getServices();
- if (services != null) {
- for (Object o : services) {
+ // Load the type converters the tracker has been tracking
+ // Here we need to use the ServiceReference to check the ranking
+ ServiceReference<TypeConverterLoader>[] serviceReferences = this.tracker.getServiceReferences();
+ if (serviceReferences != null) {
+ ArrayList<ServiceReference<TypeConverterLoader>> servicesList =
+ new ArrayList<ServiceReference<TypeConverterLoader>>(Arrays.asList(serviceReferences));
+ // Just make sure we install the high ranking fallback converter at last
+ Collections.sort(servicesList);
+ for (ServiceReference<TypeConverterLoader> sr : servicesList) {
try {
- ((TypeConverterLoader) o).load(answer);
+ LOG.debug("loading the type converter from bundle{} ", sr.getBundle().getSymbolicName());
+ ((TypeConverterLoader)this.tracker.getService(sr)).load(answer);
} catch (Throwable t) {
- throw new RuntimeCamelException("Error loading type converters from service: " + o + " due: " + t.getMessage(), t);
+ throw new RuntimeCamelException("Error loading type converters from service: " + sr + " due: " + t.getMessage(), t);
}
}
}
[5/8] git commit: CAMEL-7909 camel-netty-http consumer need to close
the connection if the response connection header is close
Posted by ni...@apache.org.
CAMEL-7909 camel-netty-http consumer need to close the connection if the response connection header is close
Conflicts:
components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/DefaultNettyHttpBinding.java
components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/handlers/HttpServerChannelHandler.java
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/9cf11bd5
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/9cf11bd5
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/9cf11bd5
Branch: refs/heads/camel-2.13.x
Commit: 9cf11bd58146abb4938d0b546d3308d6e086b6da
Parents: f0e3e20
Author: Willem Jiang <wi...@gmail.com>
Authored: Tue Oct 14 10:14:12 2014 +0800
Committer: Willem Jiang <wi...@gmail.com>
Committed: Fri Oct 17 11:35:34 2014 +0800
----------------------------------------------------------------------
.../netty/http/DefaultNettyHttpBinding.java | 9 +++++++--
.../http/handlers/HttpServerChannelHandler.java | 19 ++-----------------
.../camel/component/netty/NettyHelper.java | 2 +-
.../camel/component/netty/NettyProducer.java | 13 ++++++++++---
.../handlers/ServerResponseFutureListener.java | 5 +++++
5 files changed, 25 insertions(+), 23 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/9cf11bd5/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/DefaultNettyHttpBinding.java
----------------------------------------------------------------------
diff --git a/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/DefaultNettyHttpBinding.java b/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/DefaultNettyHttpBinding.java
index 23cae6c..fa8fa81 100644
--- a/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/DefaultNettyHttpBinding.java
+++ b/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/DefaultNettyHttpBinding.java
@@ -34,6 +34,7 @@ import org.apache.camel.Message;
import org.apache.camel.NoTypeConversionAvailableException;
import org.apache.camel.RuntimeCamelException;
import org.apache.camel.TypeConverter;
+import org.apache.camel.component.netty.NettyConstants;
import org.apache.camel.spi.HeaderFilterStrategy;
import org.apache.camel.util.ExchangeHelper;
import org.apache.camel.util.IOHelper;
@@ -67,7 +68,7 @@ public class DefaultNettyHttpBinding implements NettyHttpBinding, Cloneable {
public DefaultNettyHttpBinding(HeaderFilterStrategy headerFilterStrategy) {
this.headerFilterStrategy = headerFilterStrategy;
}
-
+
public DefaultNettyHttpBinding copy() {
try {
return (DefaultNettyHttpBinding)this.clone();
@@ -384,7 +385,7 @@ public class DefaultNettyHttpBinding implements NettyHttpBinding, Cloneable {
if (buffer.readerIndex() == buffer.writerIndex()) {
buffer.setIndex(0, buffer.writerIndex());
}
- // TODO How to enable the chunk transport
+ // TODO How to enable the chunk transport
int len = buffer.readableBytes();
// set content-length
response.setHeader(HttpHeaders.Names.CONTENT_LENGTH, len);
@@ -411,6 +412,10 @@ public class DefaultNettyHttpBinding implements NettyHttpBinding, Cloneable {
}
}
response.setHeader(HttpHeaders.Names.CONNECTION, connection);
+ // Just make sure we close the channel when the connection value is close
+ if (connection.equalsIgnoreCase(HttpHeaders.Values.CLOSE)) {
+ message.setHeader(NettyConstants.NETTY_CLOSE_CHANNEL_WHEN_COMPLETE, true);
+ }
LOG.trace("Connection: {}", connection);
return response;
http://git-wip-us.apache.org/repos/asf/camel/blob/9cf11bd5/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/handlers/HttpServerChannelHandler.java
----------------------------------------------------------------------
diff --git a/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/handlers/HttpServerChannelHandler.java b/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/handlers/HttpServerChannelHandler.java
index 8894f6a..63edef7 100644
--- a/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/handlers/HttpServerChannelHandler.java
+++ b/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/handlers/HttpServerChannelHandler.java
@@ -16,7 +16,6 @@
*/
package org.apache.camel.component.netty.http.handlers;
-import java.net.SocketAddress;
import java.net.URI;
import java.nio.channels.ClosedChannelException;
import java.nio.charset.Charset;
@@ -26,7 +25,6 @@ import javax.security.auth.login.LoginException;
import org.apache.camel.Exchange;
import org.apache.camel.LoggingLevel;
-import org.apache.camel.component.netty.NettyConsumer;
import org.apache.camel.component.netty.NettyHelper;
import org.apache.camel.component.netty.handlers.ServerChannelHandler;
import org.apache.camel.component.netty.http.HttpPrincipal;
@@ -37,7 +35,6 @@ import org.apache.camel.util.CamelLogger;
import org.apache.camel.util.ObjectHelper;
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBuffers;
-import org.jboss.netty.channel.ChannelFutureListener;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
@@ -49,8 +46,6 @@ import org.jboss.netty.handler.codec.http.HttpResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import static org.jboss.netty.handler.codec.http.HttpHeaders.is100ContinueExpected;
-import static org.jboss.netty.handler.codec.http.HttpHeaders.isKeepAlive;
import static org.jboss.netty.handler.codec.http.HttpResponseStatus.BAD_REQUEST;
import static org.jboss.netty.handler.codec.http.HttpResponseStatus.CONTINUE;
import static org.jboss.netty.handler.codec.http.HttpResponseStatus.METHOD_NOT_ALLOWED;
@@ -58,6 +53,7 @@ import static org.jboss.netty.handler.codec.http.HttpResponseStatus.SERVICE_UNAV
import static org.jboss.netty.handler.codec.http.HttpResponseStatus.UNAUTHORIZED;
import static org.jboss.netty.handler.codec.http.HttpVersion.HTTP_1_1;
+
/**
* Netty HTTP {@link ServerChannelHandler} that handles the incoming HTTP requests and routes
* the received message in Camel.
@@ -268,7 +264,7 @@ public class HttpServerChannelHandler extends ServerChannelHandler {
@Override
public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent exceptionEvent) throws Exception {
-
+
// only close if we are still allowed to run
if (consumer.isRunAllowed()) {
@@ -282,17 +278,6 @@ public class HttpServerChannelHandler extends ServerChannelHandler {
}
}
- @Override
- protected ChannelFutureListener createResponseFutureListener(NettyConsumer consumer, Exchange exchange, SocketAddress remoteAddress) {
- // make sure to close channel if not keep-alive
- if (request != null && isKeepAlive(request)) {
- LOG.trace("Request has Connection: keep-alive so Channel is not being closed");
- return null;
- } else {
- LOG.trace("Request is not Connection: close so Channel is being closed");
- return ChannelFutureListener.CLOSE;
- }
- }
@Override
protected Object getResponseBody(Exchange exchange) throws Exception {
http://git-wip-us.apache.org/repos/asf/camel/blob/9cf11bd5/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyHelper.java
----------------------------------------------------------------------
diff --git a/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyHelper.java b/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyHelper.java
index 05f3e4d..b9368fa 100644
--- a/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyHelper.java
+++ b/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyHelper.java
@@ -111,7 +111,7 @@ public final class NettyHelper {
public static void close(Channel channel) {
if (channel != null) {
LOG.trace("Closing channel: {}", channel);
- channel.close();
+ channel.close().syncUninterruptibly();
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/9cf11bd5/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyProducer.java b/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyProducer.java
index 50de736..87ad2be 100644
--- a/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyProducer.java
+++ b/components/camel-netty/src/main/java/org/apache/camel/component/netty/NettyProducer.java
@@ -498,8 +498,11 @@ public class NettyProducer extends DefaultAsyncProducer {
public void done(boolean doneSync) {
// put back in pool
try {
- LOG.trace("Putting channel back to pool {}", channel);
- pool.returnObject(channel);
+ // Only put the connected channel back to the pool
+ if (channel.isConnected()) {
+ LOG.trace("Putting channel back to pool {}", channel);
+ pool.returnObject(channel);
+ }
} catch (Exception e) {
LOG.warn("Error returning channel to pool {}. This exception will be ignored.", channel);
} finally {
@@ -525,7 +528,9 @@ public class NettyProducer extends DefaultAsyncProducer {
@Override
public void destroyObject(Channel channel) throws Exception {
LOG.trace("Destroying channel: {}", channel);
- NettyHelper.close(channel);
+ if (channel.isOpen()) {
+ NettyHelper.close(channel);
+ }
allChannels.remove(channel);
}
@@ -540,11 +545,13 @@ public class NettyProducer extends DefaultAsyncProducer {
@Override
public void activateObject(Channel channel) throws Exception {
// noop
+ LOG.trace("activateObject channel: {} -> {}", channel);
}
@Override
public void passivateObject(Channel channel) throws Exception {
// noop
+ LOG.trace("passivateObject channel: {} -> {}", channel);
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/9cf11bd5/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ServerResponseFutureListener.java
----------------------------------------------------------------------
diff --git a/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ServerResponseFutureListener.java b/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ServerResponseFutureListener.java
index 619a62e..90dc9e6 100644
--- a/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ServerResponseFutureListener.java
+++ b/components/camel-netty/src/main/java/org/apache/camel/component/netty/handlers/ServerResponseFutureListener.java
@@ -61,6 +61,11 @@ public class ServerResponseFutureListener implements ChannelFutureListener {
} else {
close = exchange.getIn().getHeader(NettyConstants.NETTY_CLOSE_CHANNEL_WHEN_COMPLETE, Boolean.class);
}
+
+ // check the setting on the exchange property
+ if (close == null) {
+ close = exchange.getProperty(NettyConstants.NETTY_CLOSE_CHANNEL_WHEN_COMPLETE, Boolean.class);
+ }
// should we disconnect, the header can override the configuration
boolean disconnect = consumer.getConfiguration().isDisconnect();