From 8280417876a8093c7127dc6ea6b7b00f74c34ec7 Mon Sep 17 00:00:00 2001 From: "Timothy A. Bish" Date: Tue, 15 Feb 2011 23:46:43 +0000 Subject: [PATCH] fix for: https://issues.apache.org/jira/browse/AMQ-3000 Scheduler wasn't taking into account that there might be a job that was scheduled earlier than the last wait time for jobs when they are fired. git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1071109 13f79535-47bb-0310-9956-ffa450edef68 --- .../broker/scheduler/JobSchedulerImpl.java | 15 ++- .../scheduler/JmsCronSchedulerTest.java | 110 ++++++++++++++++++ 2 files changed, 123 insertions(+), 2 deletions(-) create mode 100644 activemq-core/src/test/java/org/apache/activemq/broker/scheduler/JmsCronSchedulerTest.java diff --git a/activemq-core/src/main/java/org/apache/activemq/broker/scheduler/JobSchedulerImpl.java b/activemq-core/src/main/java/org/apache/activemq/broker/scheduler/JobSchedulerImpl.java index 8f5755a20b..cce75f71b1 100644 --- a/activemq-core/src/main/java/org/apache/activemq/broker/scheduler/JobSchedulerImpl.java +++ b/activemq-core/src/main/java/org/apache/activemq/broker/scheduler/JobSchedulerImpl.java @@ -455,7 +455,7 @@ class JobSchedulerImpl extends ServiceSupport implements Runnable, JobScheduler long currentTime = System.currentTimeMillis(); // Reads the list of the next entries and removes them from the store in one atomic step. - // Prevents race conditions on short delays, when storeJob() tries to append new items to the + // Prevents race conditions on short delays, when storeJob() tries to append new items to the // existing list during this read operation (see AMQ-3141). synchronized (this) { Map.Entry> first = getNextToSchedule(); @@ -464,7 +464,6 @@ class JobSchedulerImpl extends ServiceSupport implements Runnable, JobScheduler final long executionTime = first.getKey(); long nextExecutionTime = 0; if (executionTime <= currentTime) { - for (final JobLocation job : list) { int repeat = job.getRepeat(); nextExecutionTime = calculateNextExecutionTime(job, currentTime, repeat); @@ -488,6 +487,7 @@ class JobSchedulerImpl extends ServiceSupport implements Runnable, JobScheduler // execute at // this time - just a cron job - so fire it fireJob(job); + //this.scheduleTime.setWaitTime(this.scheduleTime.DEFAULT_WAIT); } if (nextExecutionTime > currentTime) { // we will run again ... @@ -517,6 +517,17 @@ class JobSchedulerImpl extends ServiceSupport implements Runnable, JobScheduler // now remove all jobs that have not been // rescheduled from this execution time remove(executionTime); + + // If there is a job that should fire before the currently set wait time + // we need to reset wait time otherwise we'll miss it. + Map.Entry> nextUp = getNextToSchedule(); + if (nextUp != null) { + final long timeUntilNextScheduled = nextUp.getKey() - currentTime; + if (timeUntilNextScheduled < this.scheduleTime.getWaitTime()) { + this.scheduleTime.setWaitTime(timeUntilNextScheduled); + } + } + } else { if (LOG.isDebugEnabled()) { LOG.debug("Not yet time to execute the job, waiting " + (executionTime - currentTime) + " ms"); diff --git a/activemq-core/src/test/java/org/apache/activemq/broker/scheduler/JmsCronSchedulerTest.java b/activemq-core/src/test/java/org/apache/activemq/broker/scheduler/JmsCronSchedulerTest.java new file mode 100644 index 0000000000..c25d09b5ad --- /dev/null +++ b/activemq-core/src/test/java/org/apache/activemq/broker/scheduler/JmsCronSchedulerTest.java @@ -0,0 +1,110 @@ +/** + * 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.activemq.broker.scheduler; + +import java.io.File; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import javax.jms.Connection; +import javax.jms.Message; +import javax.jms.MessageConsumer; +import javax.jms.MessageListener; +import javax.jms.MessageProducer; +import javax.jms.Session; +import javax.jms.TextMessage; + +import org.apache.activemq.EmbeddedBrokerTestSupport; +import org.apache.activemq.ScheduledMessage; +import org.apache.activemq.broker.BrokerService; +import org.apache.activemq.util.IOHelper; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class JmsCronSchedulerTest extends EmbeddedBrokerTestSupport { + + private static final Logger LOG = LoggerFactory.getLogger(JmsCronSchedulerTest.class); + + public void testSimulatenousCron() throws Exception { + + final int COUNT = 10; + final AtomicInteger count = new AtomicInteger(); + Connection connection = createConnection(); + + Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + + MessageConsumer consumer = session.createConsumer(destination); + + final CountDownLatch latch = new CountDownLatch(COUNT); + consumer.setMessageListener(new MessageListener() { + public void onMessage(Message message) { + latch.countDown(); + count.incrementAndGet(); + LOG.debug("Received one Message, count is at: " + count.get()); + } + }); + + connection.start(); + for (int i = 0; i < COUNT; i++) { + MessageProducer producer = session.createProducer(destination); + TextMessage message = session.createTextMessage("test msg "+i); + message.setStringProperty(ScheduledMessage.AMQ_SCHEDULED_CRON, "* * * * *"); + producer.send(message); + producer.close(); + //wait a couple sec so cron start time is different for next message + Thread.sleep(2000); + } + SchedulerBroker sb = (SchedulerBroker) this.broker.getBroker().getAdaptor(SchedulerBroker.class); + JobScheduler js = sb.getJobScheduler(); + List list = js.getAllJobs(); + assertEquals(COUNT, list.size()); + latch.await(2, TimeUnit.MINUTES); + //All should messages should have been received by now + assertEquals(COUNT, count.get()); + } + + @Override + protected void setUp() throws Exception { + bindAddress = "vm://localhost"; + super.setUp(); + } + + @Override + protected BrokerService createBroker() throws Exception { + return createBroker(true); + } + + protected BrokerService createBroker(boolean delete) throws Exception { + File schedulerDirectory = new File("target/scheduler"); + if (delete) { + IOHelper.mkdirs(schedulerDirectory); + IOHelper.deleteChildren(schedulerDirectory); + } + BrokerService answer = new BrokerService(); + answer.setPersistent(isPersistent()); + answer.getManagementContext().setCreateConnector(false); + answer.setDeleteAllMessagesOnStartup(true); + answer.setDataDirectory("target"); + answer.setSchedulerDirectoryFile(schedulerDirectory); + answer.setSchedulerSupport(true); + answer.setUseJmx(false); + answer.addConnector(bindAddress); + return answer; + } +}