You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2013/11/08 15:51:51 UTC
[1/5] git commit: CAMEL-6941: Fixed concurrency issue when shutdown
on default executor service manager.
Updated Branches:
refs/heads/camel-2.11.x 31fddb73c -> 69ccc76bf
refs/heads/camel-2.12.x 8228cfee9 -> ada4295fa
refs/heads/master d389a6e44 -> 5c459fdc9
CAMEL-6941: Fixed concurrency issue when shutdown on default executor service manager.
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/5c459fdc
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/5c459fdc
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/5c459fdc
Branch: refs/heads/master
Commit: 5c459fdc97faa461f3491f63fbeb7d65c2ab4159
Parents: 0b4f88c
Author: Claus Ibsen <da...@apache.org>
Authored: Fri Nov 8 15:50:52 2013 +0100
Committer: Claus Ibsen <da...@apache.org>
Committed: Fri Nov 8 15:52:32 2013 +0100
----------------------------------------------------------------------
.../org/apache/camel/impl/DefaultExecutorServiceManager.java | 8 ++++----
1 file changed, 4 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/5c459fdc/camel-core/src/main/java/org/apache/camel/impl/DefaultExecutorServiceManager.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/impl/DefaultExecutorServiceManager.java b/camel-core/src/main/java/org/apache/camel/impl/DefaultExecutorServiceManager.java
index dd8797d..a673382 100644
--- a/camel-core/src/main/java/org/apache/camel/impl/DefaultExecutorServiceManager.java
+++ b/camel-core/src/main/java/org/apache/camel/impl/DefaultExecutorServiceManager.java
@@ -16,13 +16,13 @@
*/
package org.apache.camel.impl;
-import java.util.ArrayList;
-import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
@@ -60,11 +60,11 @@ public class DefaultExecutorServiceManager extends ServiceSupport implements Exe
private final CamelContext camelContext;
private ThreadPoolFactory threadPoolFactory = new DefaultThreadPoolFactory();
- private final List<ExecutorService> executorServices = new ArrayList<ExecutorService>();
+ private final List<ExecutorService> executorServices = new CopyOnWriteArrayList<ExecutorService>();
private String threadNamePattern;
private long shutdownAwaitTermination = 10000;
private String defaultThreadPoolProfileId = "defaultThreadPoolProfile";
- private final Map<String, ThreadPoolProfile> threadPoolProfiles = new HashMap<String, ThreadPoolProfile>();
+ private final Map<String, ThreadPoolProfile> threadPoolProfiles = new ConcurrentHashMap<String, ThreadPoolProfile>();
private ThreadPoolProfile defaultProfile;
public DefaultExecutorServiceManager(CamelContext camelContext) {
[2/5] git commit: CAMEL-6947: rabbitmq producer should start|stop
more cleanly, such as making sure to call close and shutdown the thread pool.
Posted by da...@apache.org.
CAMEL-6947: rabbitmq producer should start|stop more cleanly, such as making sure to call close and shutdown the thread pool.
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/0b4f88c5
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/0b4f88c5
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/0b4f88c5
Branch: refs/heads/master
Commit: 0b4f88c565bf8dfb5014d90429ff9918c47f9ec5
Parents: a16b907
Author: Claus Ibsen <da...@apache.org>
Authored: Fri Nov 8 15:41:44 2013 +0100
Committer: Claus Ibsen <da...@apache.org>
Committed: Fri Nov 8 15:52:32 2013 +0100
----------------------------------------------------------------------
.../component/rabbitmq/RabbitMQProducer.java | 49 +++++++++++++++++---
1 file changed, 42 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/0b4f88c5/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQProducer.java b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQProducer.java
index 3bebb3f..5bf4269 100644
--- a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQProducer.java
+++ b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQProducer.java
@@ -21,7 +21,7 @@ import java.math.BigDecimal;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
-import java.util.concurrent.Executors;
+import java.util.concurrent.ExecutorService;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
@@ -32,13 +32,13 @@ import org.apache.camel.util.ObjectHelper;
public class RabbitMQProducer extends DefaultProducer {
- private final Connection conn;
- private final Channel channel;
+ private int closeTimeout = 30 * 1000;
+ private Connection conn;
+ private Channel channel;
+ private ExecutorService executorService;
public RabbitMQProducer(RabbitMQEndpoint endpoint) throws IOException {
super(endpoint);
- this.conn = endpoint.connect(Executors.newSingleThreadExecutor());
- this.channel = conn.createChannel();
}
@Override
@@ -46,8 +46,35 @@ public class RabbitMQProducer extends DefaultProducer {
return (RabbitMQEndpoint) super.getEndpoint();
}
- public void shutdown() throws IOException {
- conn.close();
+ @Override
+ protected void doStart() throws Exception {
+ this.executorService = getEndpoint().getCamelContext().getExecutorServiceManager().newSingleThreadExecutor(this, "CamelRabbitMQProducer[" + getEndpoint().getQueue() + "]");
+
+ log.trace("Creating connection...");
+ this.conn = getEndpoint().connect(executorService);
+ log.debug("Created connection: {}", conn);
+
+ log.trace("Creating channel...");
+ this.channel = conn.createChannel();
+ log.debug("Created channel: {}", channel);
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ if (channel != null) {
+ log.debug("Closing channel: {}", channel);
+ channel.close();
+ channel = null;
+ }
+ if (conn != null) {
+ log.debug("Closing connection: {} with timeout: {} ms.", conn, closeTimeout);
+ conn.close(closeTimeout);
+ conn = null;
+ }
+ if (executorService != null) {
+ getEndpoint().getCamelContext().getExecutorServiceManager().shutdownNow(executorService);
+ executorService = null;
+ }
}
@Override
@@ -179,4 +206,12 @@ public class RabbitMQProducer extends DefaultProducer {
}
return null;
}
+
+ public int getCloseTimeout() {
+ return closeTimeout;
+ }
+
+ public void setCloseTimeout(int closeTimeout) {
+ this.closeTimeout = closeTimeout;
+ }
}
[3/5] git commit: CAMEL-6909: Added exchangeType option to
camel-rabbit. Thanks to Dmitry Buzdin for the patch.
Posted by da...@apache.org.
CAMEL-6909: Added exchangeType option to camel-rabbit. Thanks to Dmitry Buzdin for the patch.
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/a16b9072
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/a16b9072
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/a16b9072
Branch: refs/heads/master
Commit: a16b90726f42331194b87da55ced3fc6c6aa41f6
Parents: d389a6e
Author: Claus Ibsen <da...@apache.org>
Authored: Fri Nov 8 15:29:05 2013 +0100
Committer: Claus Ibsen <da...@apache.org>
Committed: Fri Nov 8 15:52:32 2013 +0100
----------------------------------------------------------------------
.../component/rabbitmq/RabbitMQConsumer.java | 6 +++--
.../component/rabbitmq/RabbitMQEndpoint.java | 9 +++++++
.../rabbitmq/RabbitMQComponentTest.java | 27 +++++++++++++++++---
3 files changed, 37 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/a16b9072/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQConsumer.java b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQConsumer.java
index f2fa128..df3f698 100644
--- a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQConsumer.java
+++ b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQConsumer.java
@@ -55,8 +55,10 @@ public class RabbitMQConsumer extends DefaultConsumer {
channel = conn.createChannel();
log.debug("Using channel {}", channel);
- channel.exchangeDeclare(endpoint.getExchangeName(), "direct",
- endpoint.isDurable(), endpoint.isAutoDelete(),
+ channel.exchangeDeclare(endpoint.getExchangeName(),
+ endpoint.getExchangeType(),
+ endpoint.isDurable(),
+ endpoint.isAutoDelete(),
new HashMap<String, Object>());
// need to make sure the queueDeclare is same with the exchange declare
http://git-wip-us.apache.org/repos/asf/camel/blob/a16b9072/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQEndpoint.java b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQEndpoint.java
index 4423721..b008646 100644
--- a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQEndpoint.java
+++ b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQEndpoint.java
@@ -51,6 +51,7 @@ public class RabbitMQEndpoint extends DefaultEndpoint {
private boolean durable = true;
private String queue = String.valueOf(UUID.randomUUID().toString().hashCode());
private String exchangeName;
+ private String exchangeType = "direct";
private String routingKey;
public RabbitMQEndpoint() {
@@ -214,6 +215,14 @@ public class RabbitMQEndpoint extends DefaultEndpoint {
this.exchangeName = exchangeName;
}
+ public String getExchangeType() {
+ return exchangeType;
+ }
+
+ public void setExchangeType(String exchangeType) {
+ this.exchangeType = exchangeType;
+ }
+
public String getRoutingKey() {
return routingKey;
}
http://git-wip-us.apache.org/repos/asf/camel/blob/a16b9072/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQComponentTest.java
----------------------------------------------------------------------
diff --git a/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQComponentTest.java b/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQComponentTest.java
index 244bc29..454a26d 100644
--- a/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQComponentTest.java
+++ b/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQComponentTest.java
@@ -30,6 +30,18 @@ public class RabbitMQComponentTest {
private CamelContext context = Mockito.mock(CamelContext.class);
@Test
+ public void testDefaultProperties() throws Exception {
+ RabbitMQEndpoint endpoint = createEndpoint(new HashMap<String, Object>());
+
+ assertEquals(14, endpoint.getPortNumber());
+ assertEquals(10, endpoint.getThreadPoolSize());
+ assertEquals(true, endpoint.isAutoAck());
+ assertEquals(true, endpoint.isAutoDelete());
+ assertEquals(true, endpoint.isDurable());
+ assertEquals("direct", endpoint.getExchangeType());
+ }
+
+ @Test
public void testPropertiesSet() throws Exception {
Map<String, Object> params = new HashMap<String, Object>();
params.put("username", "coldplay");
@@ -40,11 +52,10 @@ public class RabbitMQComponentTest {
params.put("portNumber", 14123);
params.put("hostname", "special.host");
params.put("queue", "queuey");
+ params.put("exchangeType", "topic");
- String uri = "rabbitmq:special.host:14/queuey";
- String remaining = "special.host:14/queuey";
+ RabbitMQEndpoint endpoint = createEndpoint(params);
- RabbitMQEndpoint endpoint = new RabbitMQComponent(context).createEndpoint(uri, remaining, params);
assertEquals("chrism", endpoint.getPassword());
assertEquals("coldplay", endpoint.getUsername());
assertEquals("queuey", endpoint.getQueue());
@@ -53,5 +64,15 @@ public class RabbitMQComponentTest {
assertEquals(14123, endpoint.getPortNumber());
assertEquals(515, endpoint.getThreadPoolSize());
assertEquals(true, endpoint.isAutoAck());
+ assertEquals(true, endpoint.isAutoDelete());
+ assertEquals(true, endpoint.isDurable());
+ assertEquals("topic", endpoint.getExchangeType());
+ }
+
+ private RabbitMQEndpoint createEndpoint(Map<String, Object> params) throws Exception {
+ String uri = "rabbitmq:special.host:14/queuey";
+ String remaining = "special.host:14/queuey";
+
+ return new RabbitMQComponent(context).createEndpoint(uri, remaining, params);
}
}
[4/5] git commit: CAMEL-6941: Fixed concurrency issue when shutdown
on default executor service manager.
Posted by da...@apache.org.
CAMEL-6941: Fixed concurrency issue when shutdown on default executor service manager.
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/ada4295f
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/ada4295f
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/ada4295f
Branch: refs/heads/camel-2.12.x
Commit: ada4295fa07140121747e197a5a7f728fcd4985c
Parents: 8228cfe
Author: Claus Ibsen <da...@apache.org>
Authored: Fri Nov 8 15:50:52 2013 +0100
Committer: Claus Ibsen <da...@apache.org>
Committed: Fri Nov 8 15:52:44 2013 +0100
----------------------------------------------------------------------
.../org/apache/camel/impl/DefaultExecutorServiceManager.java | 8 ++++----
1 file changed, 4 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/ada4295f/camel-core/src/main/java/org/apache/camel/impl/DefaultExecutorServiceManager.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/impl/DefaultExecutorServiceManager.java b/camel-core/src/main/java/org/apache/camel/impl/DefaultExecutorServiceManager.java
index dd8797d..a673382 100644
--- a/camel-core/src/main/java/org/apache/camel/impl/DefaultExecutorServiceManager.java
+++ b/camel-core/src/main/java/org/apache/camel/impl/DefaultExecutorServiceManager.java
@@ -16,13 +16,13 @@
*/
package org.apache.camel.impl;
-import java.util.ArrayList;
-import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
@@ -60,11 +60,11 @@ public class DefaultExecutorServiceManager extends ServiceSupport implements Exe
private final CamelContext camelContext;
private ThreadPoolFactory threadPoolFactory = new DefaultThreadPoolFactory();
- private final List<ExecutorService> executorServices = new ArrayList<ExecutorService>();
+ private final List<ExecutorService> executorServices = new CopyOnWriteArrayList<ExecutorService>();
private String threadNamePattern;
private long shutdownAwaitTermination = 10000;
private String defaultThreadPoolProfileId = "defaultThreadPoolProfile";
- private final Map<String, ThreadPoolProfile> threadPoolProfiles = new HashMap<String, ThreadPoolProfile>();
+ private final Map<String, ThreadPoolProfile> threadPoolProfiles = new ConcurrentHashMap<String, ThreadPoolProfile>();
private ThreadPoolProfile defaultProfile;
public DefaultExecutorServiceManager(CamelContext camelContext) {
[5/5] git commit: CAMEL-6941: Fixed concurrency issue when shutdown
on default executor service manager.
Posted by da...@apache.org.
CAMEL-6941: Fixed concurrency issue when shutdown on default executor service manager.
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/69ccc76b
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/69ccc76b
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/69ccc76b
Branch: refs/heads/camel-2.11.x
Commit: 69ccc76bf412dd4118b2f03b3972878b6eb0f6c6
Parents: 31fddb7
Author: Claus Ibsen <da...@apache.org>
Authored: Fri Nov 8 15:50:52 2013 +0100
Committer: Claus Ibsen <da...@apache.org>
Committed: Fri Nov 8 15:52:56 2013 +0100
----------------------------------------------------------------------
.../org/apache/camel/impl/DefaultExecutorServiceManager.java | 8 ++++----
1 file changed, 4 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/69ccc76b/camel-core/src/main/java/org/apache/camel/impl/DefaultExecutorServiceManager.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/impl/DefaultExecutorServiceManager.java b/camel-core/src/main/java/org/apache/camel/impl/DefaultExecutorServiceManager.java
index d8bd891..08d6bc2 100644
--- a/camel-core/src/main/java/org/apache/camel/impl/DefaultExecutorServiceManager.java
+++ b/camel-core/src/main/java/org/apache/camel/impl/DefaultExecutorServiceManager.java
@@ -16,13 +16,13 @@
*/
package org.apache.camel.impl;
-import java.util.ArrayList;
-import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
@@ -59,11 +59,11 @@ public class DefaultExecutorServiceManager extends ServiceSupport implements Exe
private final CamelContext camelContext;
private ThreadPoolFactory threadPoolFactory = new DefaultThreadPoolFactory();
- private final List<ExecutorService> executorServices = new ArrayList<ExecutorService>();
+ private final List<ExecutorService> executorServices = new CopyOnWriteArrayList<ExecutorService>();
private String threadNamePattern;
private long shutdownAwaitTermination = 10000;
private String defaultThreadPoolProfileId = "defaultThreadPoolProfile";
- private final Map<String, ThreadPoolProfile> threadPoolProfiles = new HashMap<String, ThreadPoolProfile>();
+ private final Map<String, ThreadPoolProfile> threadPoolProfiles = new ConcurrentHashMap<String, ThreadPoolProfile>();
private ThreadPoolProfile defaultProfile;
public DefaultExecutorServiceManager(CamelContext camelContext) {