CAMEL-6197: Added disconnectLocalSession option to disconnect local sessions to avoid OOME. Thanks to Seb C for contribution.
Project: http://git-wip-us.apache.org/repos/asf/camel/repo Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/f3de135d Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/f3de135d Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/f3de135d Branch: refs/heads/camel-2.10.x Commit: f3de135d8aa7bf8db09dfd0010b84cd274c0ba1a Parents: e10eecf Author: Claus Ibsen <davscl...@apache.org> Authored: Sun Jun 2 17:50:50 2013 +0200 Committer: Claus Ibsen <davscl...@apache.org> Committed: Sun Jun 2 17:51:39 2013 +0200 ---------------------------------------------------------------------- .../camel/component/cometd/CometdEndpoint.java | 9 ++++++ .../camel/component/cometd/CometdProducer.java | 23 ++++++++++---- 2 files changed, 25 insertions(+), 7 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/camel/blob/f3de135d/components/camel-cometd/src/main/java/org/apache/camel/component/cometd/CometdEndpoint.java ---------------------------------------------------------------------- diff --git a/components/camel-cometd/src/main/java/org/apache/camel/component/cometd/CometdEndpoint.java b/components/camel-cometd/src/main/java/org/apache/camel/component/cometd/CometdEndpoint.java index be9ec9d..4a5a342 100644 --- a/components/camel-cometd/src/main/java/org/apache/camel/component/cometd/CometdEndpoint.java +++ b/components/camel-cometd/src/main/java/org/apache/camel/component/cometd/CometdEndpoint.java @@ -44,6 +44,7 @@ public class CometdEndpoint extends DefaultEndpoint { private boolean crossOriginFilterOn; private String allowedOrigins; private String filterPath; + private boolean disconnectLocalSession = true; public CometdEndpoint(CometdComponent component, String uri, String remaining, Map<String, Object> parameters) { super(uri, component); @@ -193,4 +194,12 @@ public class CometdEndpoint extends DefaultEndpoint { public void setFilterPath(String filterPath) { this.filterPath = filterPath; } + + public boolean isDisconnectLocalSession() { + return disconnectLocalSession; + } + + public void setDisconnectLocalSession(boolean disconnectLocalSession) { + this.disconnectLocalSession = disconnectLocalSession; + } } http://git-wip-us.apache.org/repos/asf/camel/blob/f3de135d/components/camel-cometd/src/main/java/org/apache/camel/component/cometd/CometdProducer.java ---------------------------------------------------------------------- diff --git a/components/camel-cometd/src/main/java/org/apache/camel/component/cometd/CometdProducer.java b/components/camel-cometd/src/main/java/org/apache/camel/component/cometd/CometdProducer.java index a077db0..a55bf81 100644 --- a/components/camel-cometd/src/main/java/org/apache/camel/component/cometd/CometdProducer.java +++ b/components/camel-cometd/src/main/java/org/apache/camel/component/cometd/CometdProducer.java @@ -50,7 +50,7 @@ public class CometdProducer extends DefaultProducer implements CometdProducerCon endpoint.connect(this); // should probably look into synchronization for this. if (service == null) { - service = new ProducerService(getBayeux(), new CometdBinding(bayeux), endpoint.getPath(), this); + service = new ProducerService(getBayeux(), new CometdBinding(bayeux), endpoint.getPath(), this, getEndpoint().isDisconnectLocalSession()); } } @@ -84,12 +84,14 @@ public class CometdProducer extends DefaultProducer implements CometdProducerCon private final CometdProducer producer; private final CometdBinding binding; + private final boolean disconnectLocalSession; public ProducerService(BayeuxServer bayeux, CometdBinding cometdBinding, String channel, - CometdProducer producer) { + CometdProducer producer, boolean disconnectLocalSession) { super(bayeux, channel); this.producer = producer; this.binding = cometdBinding; + this.disconnectLocalSession = disconnectLocalSession; } public void process(final Exchange exchange) { @@ -98,11 +100,18 @@ public class CometdProducer extends DefaultProducer implements CometdProducerCon ServerChannel channel = bayeux.getChannel(channelName); ServerSession serverSession = getServerSession(); - if (channel != null) { - logDelivery(exchange, channel); - ServerMessage.Mutable mutable = binding.createCometdMessage(channel, serverSession, - exchange.getIn()); - channel.publish(serverSession, mutable); + try { + if (channel != null) { + logDelivery(exchange, channel); + ServerMessage.Mutable mutable = binding.createCometdMessage(channel, serverSession, + exchange.getIn()); + channel.publish(serverSession, mutable); + } + } finally { + if (disconnectLocalSession && serverSession.isLocalSession()) { + LOG.trace("Disconnection local session {}", serverSession); + serverSession.disconnect(); + } } }
