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 2015/02/21 16:02:35 UTC
[8/8] camel git commit: CAMEL-8368: Adding new scheduler component
CAMEL-8368: Adding new scheduler component
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/cdadacb8
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/cdadacb8
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/cdadacb8
Branch: refs/heads/master
Commit: cdadacb80b61d9f44611e807dc063151bf80d328
Parents: 0478551
Author: Claus Ibsen <da...@apache.org>
Authored: Sat Feb 21 16:00:21 2015 +0100
Committer: Claus Ibsen <da...@apache.org>
Committed: Sat Feb 21 16:00:21 2015 +0100
----------------------------------------------------------------------
.../component/scheduler/SchedulerConsumer.java | 52 ++++++++++++++++----
1 file changed, 43 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/cdadacb8/camel-core/src/main/java/org/apache/camel/component/scheduler/SchedulerConsumer.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/component/scheduler/SchedulerConsumer.java b/camel-core/src/main/java/org/apache/camel/component/scheduler/SchedulerConsumer.java
index 49b3b2d..a18b031 100644
--- a/camel-core/src/main/java/org/apache/camel/component/scheduler/SchedulerConsumer.java
+++ b/camel-core/src/main/java/org/apache/camel/component/scheduler/SchedulerConsumer.java
@@ -16,28 +16,61 @@
*/
package org.apache.camel.component.scheduler;
-import org.apache.camel.Endpoint;
+import java.util.Date;
+
+import org.apache.camel.AsyncCallback;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.impl.ScheduledPollConsumer;
public class SchedulerConsumer extends ScheduledPollConsumer {
- public SchedulerConsumer(Endpoint endpoint, Processor processor) {
+ public SchedulerConsumer(SchedulerEndpoint endpoint, Processor processor) {
super(endpoint, processor);
}
@Override
+ public SchedulerEndpoint getEndpoint() {
+ return (SchedulerEndpoint) super.getEndpoint();
+ }
+
+ @Override
protected int poll() throws Exception {
- Exchange exchange = getEndpoint().createExchange();
- try {
- getProcessor().process(exchange);
- } catch (Exception e) {
- exchange.setException(e);
+ return sendTimerExchange();
+ }
+
+ protected int sendTimerExchange() {
+ final Exchange exchange = getEndpoint().createExchange();
+ exchange.setProperty(Exchange.TIMER_NAME, getEndpoint().getName());
+
+ Date now = new Date();
+ exchange.setProperty(Exchange.TIMER_FIRED_TIME, now);
+
+ if (log.isTraceEnabled()) {
+ log.trace("Timer {} is firing", getEndpoint().getName());
}
- if (exchange.getException() != null) {
- getExceptionHandler().handleException("Error processing exchange", exchange, exchange.getException());
+ if (!getEndpoint().isSynchronous()) {
+ getAsyncProcessor().process(exchange, new AsyncCallback() {
+ @Override
+ public void done(boolean doneSync) {
+ // handle any thrown exception
+ if (exchange.getException() != null) {
+ getExceptionHandler().handleException("Error processing exchange", exchange, exchange.getException());
+ }
+ }
+ });
+ } else {
+ try {
+ getProcessor().process(exchange);
+ } catch (Exception e) {
+ exchange.setException(e);
+ }
+
+ // handle any thrown exception
+ if (exchange.getException() != null) {
+ getExceptionHandler().handleException("Error processing exchange", exchange, exchange.getException());
+ }
}
// a property can be used to control if the scheduler polled a message or not
@@ -47,4 +80,5 @@ public class SchedulerConsumer extends ScheduledPollConsumer {
return polled ? 1 : 0;
}
+
}