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 2009/10/03 10:45:31 UTC
svn commit: r821272 - in /camel/trunk:
camel-core/src/main/java/org/apache/camel/component/bean/
camel-core/src/main/java/org/apache/camel/util/concurrent/
components/camel-mina/src/main/java/org/apache/camel/component/mina/
Author: davsclaus
Date: Sat Oct 3 08:45:31 2009
New Revision: 821272
URL: http://svn.apache.org/viewvc?rev=821272&view=rev
Log:
CAMEL-2049: camel-mina now using thread model recommended by Mina documentation. Thanks to Dmytro Rud for the patch.
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/component/bean/BeanProcessor.java
camel/trunk/camel-core/src/main/java/org/apache/camel/util/concurrent/ExecutorServiceHelper.java
camel/trunk/components/camel-mina/src/main/java/org/apache/camel/component/mina/MinaComponent.java
camel/trunk/components/camel-mina/src/main/java/org/apache/camel/component/mina/MinaConsumer.java
camel/trunk/components/camel-mina/src/main/java/org/apache/camel/component/mina/MinaProducer.java
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/component/bean/BeanProcessor.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/component/bean/BeanProcessor.java?rev=821272&r1=821271&r2=821272&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/component/bean/BeanProcessor.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/component/bean/BeanProcessor.java Sat Oct 3 08:45:31 2009
@@ -74,7 +74,7 @@
BeanInfo beanInfo = beanHolder.getBeanInfo();
// do we have a custom adapter for this POJO to a Processor
- // should not be invoced if an explicit method has been set
+ // should not be invoked if an explicit method has been set
Processor processor = getProcessor();
if (!isExplicitMethod && processor != null) {
if (LOG.isTraceEnabled()) {
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/util/concurrent/ExecutorServiceHelper.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/util/concurrent/ExecutorServiceHelper.java?rev=821272&r1=821271&r2=821272&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/util/concurrent/ExecutorServiceHelper.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/util/concurrent/ExecutorServiceHelper.java Sat Oct 3 08:45:31 2009
@@ -76,4 +76,14 @@
});
}
+ public static ExecutorService newCachedThreadPool(final String name, final boolean daemon) {
+ return Executors.newCachedThreadPool(new ThreadFactory() {
+ public Thread newThread(Runnable r) {
+ Thread answer = new Thread(r, getThreadName(name));
+ answer.setDaemon(daemon);
+ return answer;
+ }
+ });
+ }
+
}
Modified: camel/trunk/components/camel-mina/src/main/java/org/apache/camel/component/mina/MinaComponent.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-mina/src/main/java/org/apache/camel/component/mina/MinaComponent.java?rev=821272&r1=821271&r2=821272&view=diff
==============================================================================
--- camel/trunk/components/camel-mina/src/main/java/org/apache/camel/component/mina/MinaComponent.java (original)
+++ camel/trunk/components/camel-mina/src/main/java/org/apache/camel/component/mina/MinaComponent.java Sat Oct 3 08:45:31 2009
@@ -28,6 +28,7 @@
import org.apache.camel.ExchangePattern;
import org.apache.camel.impl.DefaultComponent;
import org.apache.camel.util.ObjectHelper;
+import org.apache.camel.util.concurrent.ExecutorServiceHelper;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.mina.common.DefaultIoFilterChainBuilder;
@@ -35,11 +36,13 @@
import org.apache.mina.common.IoConnector;
import org.apache.mina.common.IoFilter;
import org.apache.mina.common.IoServiceConfig;
+import org.apache.mina.common.ThreadModel;
import org.apache.mina.filter.LoggingFilter;
import org.apache.mina.filter.codec.ProtocolCodecFactory;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.codec.serialization.ObjectSerializationCodecFactory;
import org.apache.mina.filter.codec.textline.LineDelimiter;
+import org.apache.mina.filter.executor.ExecutorFilter;
import org.apache.mina.transport.socket.nio.DatagramAcceptor;
import org.apache.mina.transport.socket.nio.DatagramAcceptorConfig;
import org.apache.mina.transport.socket.nio.DatagramConnector;
@@ -159,14 +162,18 @@
long timeout = configuration.getTimeout();
boolean sync = configuration.isSync();
List<IoFilter> filters = configuration.getFilters();
+ final int processorCount = Runtime.getRuntime().availableProcessors() + 1;
- IoAcceptor acceptor = new SocketAcceptor();
+ IoAcceptor acceptor = new SocketAcceptor(processorCount, ExecutorServiceHelper.newCachedThreadPool("MinaSocketAcceptor", true));
+ IoConnector connector = new SocketConnector(processorCount, ExecutorServiceHelper.newCachedThreadPool("MinaSocketConnector", true));
SocketAddress address = new InetSocketAddress(configuration.getHost(), configuration.getPort());
- IoConnector connector = new SocketConnector();
// connector config
SocketConnectorConfig connectorConfig = new SocketConnectorConfig();
+ // must use manual thread model according to Mina documentation
+ connectorConfig.setThreadModel(ThreadModel.MANUAL);
configureCodecFactory("MinaProducer", connectorConfig, configuration);
+ connectorConfig.getFilterChain().addLast("threadPool", new ExecutorFilter(ExecutorServiceHelper.newCachedThreadPool("MinaThreadPool", true)));
if (minaLogger) {
connectorConfig.getFilterChain().addLast("logger", new LoggingFilter());
}
@@ -180,6 +187,7 @@
configureCodecFactory("MinaConsumer", acceptorConfig, configuration);
acceptorConfig.setReuseAddress(true);
acceptorConfig.setDisconnectOnUnbind(true);
+ acceptorConfig.getFilterChain().addLast("threadPool", new ExecutorFilter(ExecutorServiceHelper.newCachedThreadPool("MinaThreadPool", true)));
if (minaLogger) {
acceptorConfig.getFilterChain().addLast("logger", new LoggingFilter());
}
@@ -242,10 +250,11 @@
boolean transferExchange = configuration.isTransferExchange();
boolean sync = configuration.isSync();
List<IoFilter> filters = configuration.getFilters();
+ final int processorCount = Runtime.getRuntime().availableProcessors() + 1;
- IoAcceptor acceptor = new DatagramAcceptor();
+ IoAcceptor acceptor = new DatagramAcceptor(ExecutorServiceHelper.newCachedThreadPool("MinaDatagramAcceptor", true));
+ IoConnector connector = new DatagramConnector(ExecutorServiceHelper.newCachedThreadPool("MinaDatagramConnector", true));
SocketAddress address = new InetSocketAddress(configuration.getHost(), configuration.getPort());
- IoConnector connector = new DatagramConnector();
if (transferExchange) {
throw new IllegalArgumentException("transferExchange=true is not supported for datagram protocol");
@@ -253,6 +262,7 @@
DatagramConnectorConfig connectorConfig = new DatagramConnectorConfig();
configureDataGramCodecFactory("MinaProducer", connectorConfig, configuration);
+ connectorConfig.getFilterChain().addLast("threadPool", new ExecutorFilter(ExecutorServiceHelper.newCachedThreadPool("MinaThreadPool", true)));
if (minaLogger) {
connectorConfig.getFilterChain().addLast("logger", new LoggingFilter());
}
Modified: camel/trunk/components/camel-mina/src/main/java/org/apache/camel/component/mina/MinaConsumer.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-mina/src/main/java/org/apache/camel/component/mina/MinaConsumer.java?rev=821272&r1=821271&r2=821272&view=diff
==============================================================================
--- camel/trunk/components/camel-mina/src/main/java/org/apache/camel/component/mina/MinaConsumer.java (original)
+++ camel/trunk/components/camel-mina/src/main/java/org/apache/camel/component/mina/MinaConsumer.java Sat Oct 3 08:45:31 2009
@@ -74,7 +74,7 @@
}
/**
- * Handles comsuming messages and replying if the exchange is out capable.
+ * Handles consuming messages and replying if the exchange is out capable.
*/
private final class ReceiveHandler extends IoHandlerAdapter {
Modified: camel/trunk/components/camel-mina/src/main/java/org/apache/camel/component/mina/MinaProducer.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-mina/src/main/java/org/apache/camel/component/mina/MinaProducer.java?rev=821272&r1=821271&r2=821272&view=diff
==============================================================================
--- camel/trunk/components/camel-mina/src/main/java/org/apache/camel/component/mina/MinaProducer.java (original)
+++ camel/trunk/components/camel-mina/src/main/java/org/apache/camel/component/mina/MinaProducer.java Sat Oct 3 08:45:31 2009
@@ -36,7 +36,7 @@
import org.apache.mina.transport.socket.nio.SocketConnector;
/**
- * A {@link Producer} implementation for MINA
+ * A {@link org.apache.camel.Producer} implementation for MINA
*
* @version $Revision$
*/
@@ -117,7 +117,7 @@
ResponseHandler handler = (ResponseHandler) session.getHandler();
if (handler.getCause() != null) {
throw new CamelExchangeException("Response Handler had an exception", exchange, handler.getCause());
- } else if (!handler.isMessageRecieved()) {
+ } else if (!handler.isMessageReceived()) {
// no message received
throw new CamelExchangeException("No response received from remote server: " + endpoint.getEndpointUri(), exchange);
} else {
@@ -182,7 +182,7 @@
private MinaEndpoint endpoint;
private Object message;
private Throwable cause;
- private boolean messageRecieved;
+ private boolean messageReceived;
private ResponseHandler(MinaEndpoint endpoint) {
this.endpoint = endpoint;
@@ -191,7 +191,7 @@
public void reset() {
this.message = null;
this.cause = null;
- this.messageRecieved = false;
+ this.messageReceived = false;
}
@Override
@@ -200,7 +200,7 @@
LOG.debug("Message received: " + message);
}
this.message = message;
- messageRecieved = true;
+ messageReceived = true;
cause = null;
countDown();
}
@@ -230,7 +230,7 @@
LOG.error("Exception on receiving message from address: " + this.endpoint.getAddress()
+ " using connector: " + this.endpoint.getConnector(), cause);
this.message = null;
- this.messageRecieved = false;
+ this.messageReceived = false;
this.cause = cause;
if (ioSession != null) {
ioSession.close();
@@ -245,8 +245,8 @@
return this.message;
}
- public boolean isMessageRecieved() {
- return messageRecieved;
+ public boolean isMessageReceived() {
+ return messageReceived;
}
}