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 2011/06/16 07:22:02 UTC
svn commit: r1136290 - in /camel/trunk/camel-core/src:
main/java/org/apache/camel/impl/ test/java/org/apache/camel/component/file/
test/java/org/apache/camel/impl/ test/java/org/apache/camel/management/
Author: davsclaus
Date: Thu Jun 16 05:22:02 2011
New Revision: 1136290
URL: http://svn.apache.org/viewvc?rev=1136290&view=rev
Log:
CAMEL-4105: ScheduledPollConsumer improved run thread to catch Error to avoid thread from dying. Added runLoggingLevel to control log level for start/complete logs for when the poll task runs. Changed default to useFixedDelay.
Added:
camel/trunk/camel-core/src/test/java/org/apache/camel/component/file/FileConsumeRunLoggingLevelTest.java
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/ScheduledPollConsumer.java
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/ScheduledPollEndpoint.java
camel/trunk/camel-core/src/test/java/org/apache/camel/impl/DefaultComponentValidateURITest.java
camel/trunk/camel-core/src/test/java/org/apache/camel/management/ManagedScheduledPollConsumerTest.java
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/impl/ScheduledPollConsumer.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/impl/ScheduledPollConsumer.java?rev=1136290&r1=1136289&r2=1136290&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/impl/ScheduledPollConsumer.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/impl/ScheduledPollConsumer.java Thu Jun 16 05:22:02 2011
@@ -21,6 +21,7 @@ import java.util.concurrent.ScheduledFut
import java.util.concurrent.TimeUnit;
import org.apache.camel.Endpoint;
+import org.apache.camel.LoggingLevel;
import org.apache.camel.Processor;
import org.apache.camel.SuspendableService;
import org.apache.camel.spi.PollingConsumerPollStrategy;
@@ -43,8 +44,9 @@ public abstract class ScheduledPollConsu
private long initialDelay = 1000;
private long delay = 500;
private TimeUnit timeUnit = TimeUnit.MILLISECONDS;
- private boolean useFixedDelay;
+ private boolean useFixedDelay = true;
private PollingConsumerPollStrategy pollStrategy = new DefaultPollingConsumerPollStrategy();
+ private LoggingLevel runLoggingLevel = LoggingLevel.TRACE;
public ScheduledPollConsumer(Endpoint endpoint, Processor processor) {
super(endpoint, processor);
@@ -65,6 +67,44 @@ public abstract class ScheduledPollConsu
* Invoked whenever we should be polled
*/
public void run() {
+ // avoid this thread to throw exceptions because the thread pool wont re-schedule a new thread
+ try {
+ // log starting
+ if (LoggingLevel.ERROR == runLoggingLevel) {
+ LOG.error("Scheduled task started on: {}", this.getEndpoint());
+ } else if (LoggingLevel.WARN == runLoggingLevel) {
+ LOG.warn("Scheduled task started on: {}", this.getEndpoint());
+ } else if (LoggingLevel.INFO == runLoggingLevel) {
+ LOG.info("Scheduled task started on: {}", this.getEndpoint());
+ } else if (LoggingLevel.DEBUG == runLoggingLevel) {
+ LOG.debug("Scheduled task started on: {}", this.getEndpoint());
+ } else {
+ LOG.trace("Scheduled task started on: {}", this.getEndpoint());
+ }
+
+ // execute scheduled task
+ doRun();
+
+ // log completed
+ if (LoggingLevel.ERROR == runLoggingLevel) {
+ LOG.error("Scheduled task completed on: {}", this.getEndpoint());
+ } else if (LoggingLevel.WARN == runLoggingLevel) {
+ LOG.warn("Scheduled task completed on: {}", this.getEndpoint());
+ } else if (LoggingLevel.INFO == runLoggingLevel) {
+ LOG.info("Scheduled task completed on: {}", this.getEndpoint());
+ } else if (LoggingLevel.DEBUG == runLoggingLevel) {
+ LOG.debug("Scheduled task completed on: {}", this.getEndpoint());
+ } else {
+ LOG.trace("Scheduled task completed on: {}", this.getEndpoint());
+ }
+
+ } catch (Error e) {
+ // must catch Error, to ensure the task is re-scheduled
+ LOG.error("Error occurred during running scheduled task on: " + this.getEndpoint() + ", due: " + e.getMessage(), e);
+ }
+ }
+
+ private void doRun() {
if (isSuspended()) {
LOG.trace("Cannot start to poll: {} as its suspended", this.getEndpoint());
return;
@@ -169,6 +209,14 @@ public abstract class ScheduledPollConsu
this.useFixedDelay = useFixedDelay;
}
+ public LoggingLevel getRunLoggingLevel() {
+ return runLoggingLevel;
+ }
+
+ public void setRunLoggingLevel(LoggingLevel runLoggingLevel) {
+ this.runLoggingLevel = runLoggingLevel;
+ }
+
public PollingConsumerPollStrategy getPollStrategy() {
return pollStrategy;
}
@@ -191,6 +239,9 @@ public abstract class ScheduledPollConsu
@Override
protected void doStart() throws Exception {
super.doStart();
+ ObjectHelper.notNull(executor, "executor", this);
+ ObjectHelper.notNull(pollStrategy, "pollStrategy", this);
+
if (isUseFixedDelay()) {
if (LOG.isDebugEnabled()) {
LOG.debug("Scheduling poll (fixed delay) with initialDelay: {}, delay: {} ({}) for: {}",
@@ -209,6 +260,7 @@ public abstract class ScheduledPollConsu
@Override
protected void doStop() throws Exception {
if (future != null) {
+ LOG.debug("This consumer is stopping, so cancelling scheduled task: " + future);
future.cancel(false);
}
super.doStop();
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/impl/ScheduledPollEndpoint.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/impl/ScheduledPollEndpoint.java?rev=1136290&r1=1136289&r2=1136290&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/impl/ScheduledPollEndpoint.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/impl/ScheduledPollEndpoint.java Thu Jun 16 05:22:02 2011
@@ -91,7 +91,8 @@ public abstract class ScheduledPollEndpo
Object timeUnit = options.remove("timeUnit");
Object useFixedDelay = options.remove("useFixedDelay");
Object pollStrategy = options.remove("pollStrategy");
- if (initialDelay != null || delay != null || timeUnit != null || useFixedDelay != null || pollStrategy != null) {
+ Object runLoggingLevel = options.remove("runLoggingLevel");
+ if (initialDelay != null || delay != null || timeUnit != null || useFixedDelay != null || pollStrategy != null || runLoggingLevel != null) {
if (consumerProperties == null) {
consumerProperties = new HashMap<String, Object>();
}
@@ -110,6 +111,9 @@ public abstract class ScheduledPollEndpo
if (pollStrategy != null) {
consumerProperties.put("pollStrategy", pollStrategy);
}
+ if (runLoggingLevel != null) {
+ consumerProperties.put("runLoggingLevel", runLoggingLevel);
+ }
}
}
}
Added: camel/trunk/camel-core/src/test/java/org/apache/camel/component/file/FileConsumeRunLoggingLevelTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/component/file/FileConsumeRunLoggingLevelTest.java?rev=1136290&view=auto
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/component/file/FileConsumeRunLoggingLevelTest.java (added)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/component/file/FileConsumeRunLoggingLevelTest.java Thu Jun 16 05:22:02 2011
@@ -0,0 +1,46 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * 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 obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.file;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+
+/**
+ *
+ */
+public class FileConsumeRunLoggingLevelTest extends ContextTestSupport {
+
+ public void testRunLoggingLevel() throws Exception {
+ getMockEndpoint("mock:result").expectedMessageCount(1);
+
+ template.sendBodyAndHeader("file:target/files", "Hello World", Exchange.FILE_NAME, "hello.txt");
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("file:target/files?runLoggingLevel=INFO")
+ .to("mock:result");
+ }
+ };
+ }
+}
Modified: camel/trunk/camel-core/src/test/java/org/apache/camel/impl/DefaultComponentValidateURITest.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/impl/DefaultComponentValidateURITest.java?rev=1136290&r1=1136289&r2=1136290&view=diff
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/impl/DefaultComponentValidateURITest.java (original)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/impl/DefaultComponentValidateURITest.java Thu Jun 16 05:22:02 2011
@@ -76,7 +76,7 @@ public class DefaultComponentValidateURI
endpint = context.getEndpoint("file://target/foo?consumer.delay=1000&consumer.initialDelay=5000");
assertNotNull(endpint);
- endpint = context.getEndpoint("file://target/foo?consumer.delay=1000&consumer.initialDelay=5000&consumer.useFixedDelay=true");
+ endpint = context.getEndpoint("file://target/foo?consumer.delay=1000&consumer.initialDelay=5000&consumer.useFixedDelay=false");
assertNotNull(endpint);
// without consumer. prefix
@@ -86,11 +86,11 @@ public class DefaultComponentValidateURI
endpint = context.getEndpoint("file://foo2?delay=1000&initialDelay=5000");
assertNotNull(endpint);
- endpint = context.getEndpoint("file://foo2?delay=1000&initialDelay=5000&useFixedDelay=true");
+ endpint = context.getEndpoint("file://foo2?delay=1000&initialDelay=5000&useFixedDelay=false");
assertNotNull(endpint);
// combined with and without consumer. prefix
- endpint = context.getEndpoint("file://foo3?delay=1000&consumer.initialDelay=5000&useFixedDelay=true");
+ endpint = context.getEndpoint("file://foo3?delay=1000&consumer.initialDelay=5000&useFixedDelay=false");
assertNotNull(endpint);
}
Modified: camel/trunk/camel-core/src/test/java/org/apache/camel/management/ManagedScheduledPollConsumerTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/management/ManagedScheduledPollConsumerTest.java?rev=1136290&r1=1136289&r2=1136290&view=diff
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/management/ManagedScheduledPollConsumerTest.java (original)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/management/ManagedScheduledPollConsumerTest.java Thu Jun 16 05:22:02 2011
@@ -48,7 +48,7 @@ public class ManagedScheduledPollConsume
assertEquals(1000, initialDelay.longValue());
Boolean fixedDelay = (Boolean) mbeanServer.getAttribute(on, "UseFixedDelay");
- assertEquals(Boolean.FALSE, fixedDelay);
+ assertEquals(Boolean.TRUE, fixedDelay);
String timeUnit = (String) mbeanServer.getAttribute(on, "TimeUnit");
assertEquals(TimeUnit.MILLISECONDS.toString(), timeUnit);
@@ -69,9 +69,9 @@ public class ManagedScheduledPollConsume
assertEquals(2000, delay.longValue());
// change some options
- mbeanServer.setAttribute(on, new Attribute("UseFixedDelay", Boolean.TRUE));
+ mbeanServer.setAttribute(on, new Attribute("UseFixedDelay", Boolean.FALSE));
fixedDelay = (Boolean) mbeanServer.getAttribute(on, "UseFixedDelay");
- assertEquals(Boolean.TRUE, fixedDelay);
+ assertEquals(Boolean.FALSE, fixedDelay);
mbeanServer.setAttribute(on, new Attribute("TimeUnit", TimeUnit.SECONDS.name()));
timeUnit = (String) mbeanServer.getAttribute(on, "TimeUnit");