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/04/24 11:05:22 UTC
svn commit: r1471304 - in /camel/branches/camel-2.10.x: ./
camel-core/src/main/java/org/apache/camel/api/management/mbean/
camel-core/src/main/java/org/apache/camel/impl/
camel-core/src/main/java/org/apache/camel/management/mbean/
camel-core/src/test/j...
Author: davsclaus
Date: Wed Apr 24 09:05:22 2013
New Revision: 1471304
URL: http://svn.apache.org/r1471304
Log:
CAMEL-6308: Make calling startScheduler easier on ScheduledPollConsumers. Also from JMX.
Modified:
camel/branches/camel-2.10.x/ (props changed)
camel/branches/camel-2.10.x/camel-core/src/main/java/org/apache/camel/api/management/mbean/ManagedSchedulePollConsumerMBean.java
camel/branches/camel-2.10.x/camel-core/src/main/java/org/apache/camel/impl/ScheduledPollConsumer.java
camel/branches/camel-2.10.x/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedScheduledPollConsumer.java
camel/branches/camel-2.10.x/camel-core/src/test/java/org/apache/camel/management/ManagedScheduledPollConsumerTest.java
Propchange: camel/branches/camel-2.10.x/
------------------------------------------------------------------------------
Merged /camel/trunk:r1471293
Merged /camel/branches/camel-2.11.x:r1471297
Propchange: camel/branches/camel-2.10.x/
------------------------------------------------------------------------------
Binary property 'svnmerge-integrated' - no diff available.
Modified: camel/branches/camel-2.10.x/camel-core/src/main/java/org/apache/camel/api/management/mbean/ManagedSchedulePollConsumerMBean.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.10.x/camel-core/src/main/java/org/apache/camel/api/management/mbean/ManagedSchedulePollConsumerMBean.java?rev=1471304&r1=1471303&r2=1471304&view=diff
==============================================================================
--- camel/branches/camel-2.10.x/camel-core/src/main/java/org/apache/camel/api/management/mbean/ManagedSchedulePollConsumerMBean.java (original)
+++ camel/branches/camel-2.10.x/camel-core/src/main/java/org/apache/camel/api/management/mbean/ManagedSchedulePollConsumerMBean.java Wed Apr 24 09:05:22 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/branches/camel-2.10.x/camel-core/src/main/java/org/apache/camel/impl/ScheduledPollConsumer.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.10.x/camel-core/src/main/java/org/apache/camel/impl/ScheduledPollConsumer.java?rev=1471304&r1=1471303&r2=1471304&view=diff
==============================================================================
--- camel/branches/camel-2.10.x/camel-core/src/main/java/org/apache/camel/impl/ScheduledPollConsumer.java (original)
+++ camel/branches/camel-2.10.x/camel-core/src/main/java/org/apache/camel/impl/ScheduledPollConsumer.java Wed Apr 24 09:05:22 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
private boolean startScheduler = true;
@@ -301,6 +301,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
@@ -345,19 +356,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());
}
}
@@ -366,6 +385,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/branches/camel-2.10.x/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedScheduledPollConsumer.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.10.x/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedScheduledPollConsumer.java?rev=1471304&r1=1471303&r2=1471304&view=diff
==============================================================================
--- camel/branches/camel-2.10.x/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedScheduledPollConsumer.java (original)
+++ camel/branches/camel-2.10.x/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedScheduledPollConsumer.java Wed Apr 24 09:05:22 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/branches/camel-2.10.x/camel-core/src/test/java/org/apache/camel/management/ManagedScheduledPollConsumerTest.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.10.x/camel-core/src/test/java/org/apache/camel/management/ManagedScheduledPollConsumerTest.java?rev=1471304&r1=1471303&r2=1471304&view=diff
==============================================================================
--- camel/branches/camel-2.10.x/camel-core/src/test/java/org/apache/camel/management/ManagedScheduledPollConsumerTest.java (original)
+++ camel/branches/camel-2.10.x/camel-core/src/test/java/org/apache/camel/management/ManagedScheduledPollConsumerTest.java Wed Apr 24 09:05:22 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));