Author: davsclaus
Date: Wed Apr 24 08:43:13 2013
New Revision: 1471293
URL: http://svn.apache.org/r1471293
Log:
CAMEL-6308: Make calling startScheduler easier on ScheduledPollConsumers. Also
from JMX.
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/api/management/mbean/ManagedSchedulePollConsumerMBean.java
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/ScheduledPollConsumer.java
camel/trunk/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedScheduledPollConsumer.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/api/management/mbean/ManagedSchedulePollConsumerMBean.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/api/management/mbean/ManagedSchedulePollConsumerMBean.java?rev=1471293&r1=1471292&r2=1471293&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/api/management/mbean/ManagedSchedulePollConsumerMBean.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/api/management/mbean/ManagedSchedulePollConsumerMBean.java
Wed Apr 24 08:43:13 2013
@@ -17,6 +17,7 @@
package org.apache.camel.api.management.mbean;
import org.apache.camel.api.management.ManagedAttribute;
+import org.apache.camel.api.management.ManagedOperation;
public interface ManagedSchedulePollConsumerMBean extends ManagedConsumerMBean
{
@@ -44,4 +45,10 @@ public interface ManagedSchedulePollCons
@ManagedAttribute(description = "Scheduled TimeUnit")
void setTimeUnit(String timeUnit);
+ @ManagedAttribute(description = "Is the scheduler started")
+ boolean isSchedulerStarted();
+
+ @ManagedOperation(description = "Starts the scheduler")
+ void startScheduler();
+
}
\ No newline at end of file
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=1471293&r1=1471292&r2=1471293&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
Wed Apr 24 08:43:13 2013
@@ -44,7 +44,7 @@ public abstract class ScheduledPollConsu
private ScheduledExecutorService scheduledExecutorService;
private boolean shutdownExecutor;
- private ScheduledFuture<?> future;
+ private volatile ScheduledFuture<?> future;
// if adding more options then align with
ScheduledPollEndpoint#configureScheduledPollConsumerProperties
@UriParam
@@ -309,6 +309,17 @@ public abstract class ScheduledPollConsu
}
/**
+ * Whether the scheduler has been started.
+ * <p/>
+ * The scheduler can be started with the {@link #startScheduler()} method.
+ *
+ * @return <tt>true</tt> if started, <tt>false</tt> if not.
+ */
+ public boolean isSchedulerStarted() {
+ return future != null;
+ }
+
+ /**
* Sets a custom shared {@link ScheduledExecutorService} to use as thread
pool
* <p/>
* <b>Notice: </b> When using a custom thread pool, then the lifecycle of
this thread
@@ -353,19 +364,27 @@ public abstract class ScheduledPollConsu
}
}
- protected void startScheduler() {
- if (isUseFixedDelay()) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Scheduling poll (fixed delay) with initialDelay:
{}, delay: {} ({}) for: {}",
- new Object[]{getInitialDelay(), getDelay(),
getTimeUnit().name().toLowerCase(Locale.ENGLISH), getEndpoint()});
- }
- future = scheduledExecutorService.scheduleWithFixedDelay(this,
getInitialDelay(), getDelay(), getTimeUnit());
- } else {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Scheduling poll (fixed rate) with initialDelay: {},
delay: {} ({}) for: {}",
- new Object[]{getInitialDelay(), getDelay(),
getTimeUnit().name().toLowerCase(Locale.ENGLISH), getEndpoint()});
+ /**
+ * Starts the scheduler.
+ * <p/>
+ * If the scheduler is already started, then this is a noop method call.
+ */
+ public void startScheduler() {
+ // only schedule task if we have not already done that
+ if (future == null) {
+ if (isUseFixedDelay()) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Scheduling poll (fixed delay) with
initialDelay: {}, delay: {} ({}) for: {}",
+ new Object[]{getInitialDelay(), getDelay(),
getTimeUnit().name().toLowerCase(Locale.ENGLISH), getEndpoint()});
+ }
+ future = scheduledExecutorService.scheduleWithFixedDelay(this,
getInitialDelay(), getDelay(), getTimeUnit());
+ } else {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Scheduling poll (fixed rate) with initialDelay:
{}, delay: {} ({}) for: {}",
+ new Object[]{getInitialDelay(), getDelay(),
getTimeUnit().name().toLowerCase(Locale.ENGLISH), getEndpoint()});
+ }
+ future = scheduledExecutorService.scheduleAtFixedRate(this,
getInitialDelay(), getDelay(), getTimeUnit());
}
- future = scheduledExecutorService.scheduleAtFixedRate(this,
getInitialDelay(), getDelay(), getTimeUnit());
}
}
@@ -374,6 +393,7 @@ public abstract class ScheduledPollConsu
if (future != null) {
LOG.debug("This consumer is stopping, so cancelling scheduled
task: " + future);
future.cancel(false);
+ future = null;
}
super.doStop();
}
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedScheduledPollConsumer.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedScheduledPollConsumer.java?rev=1471293&r1=1471292&r2=1471293&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedScheduledPollConsumer.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedScheduledPollConsumer.java
Wed Apr 24 08:43:13 2013
@@ -70,4 +70,12 @@ public class ManagedScheduledPollConsume
public void setTimeUnit(String timeUnit) {
getConsumer().setTimeUnit(TimeUnit.valueOf(timeUnit));
}
+
+ public boolean isSchedulerStarted() {
+ return getConsumer().isSchedulerStarted();
+ }
+
+ public void startScheduler() {
+ getConsumer().startScheduler();
+ }
}
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=1471293&r1=1471292&r2=1471293&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
Wed Apr 24 08:43:13 2013
@@ -50,6 +50,9 @@ public class ManagedScheduledPollConsume
Boolean fixedDelay = (Boolean) mbeanServer.getAttribute(on,
"UseFixedDelay");
assertEquals(Boolean.TRUE, fixedDelay);
+ Boolean schedulerStarted = (Boolean) mbeanServer.getAttribute(on,
"SchedulerStarted");
+ assertEquals(Boolean.TRUE, schedulerStarted);
+
String timeUnit = (String) mbeanServer.getAttribute(on, "TimeUnit");
assertEquals(TimeUnit.MILLISECONDS.toString(), timeUnit);
@@ -59,6 +62,9 @@ public class ManagedScheduledPollConsume
// stop it
mbeanServer.invoke(on, "stop", null, null);
+ schedulerStarted = (Boolean) mbeanServer.getAttribute(on,
"SchedulerStarted");
+ assertEquals(Boolean.FALSE, schedulerStarted);
+
// change delay
mbeanServer.setAttribute(on, new Attribute("Delay", 2000));
