You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@sling.apache.org by to...@apache.org on 2016/05/24 12:36:52 UTC
svn commit: r1745352 - in
/sling/trunk/contrib/extensions/distribution/core/src:
main/java/org/apache/sling/distribution/queue/impl/simple/
test/java/org/apache/sling/distribution/queue/impl/simple/
Author: tommaso
Date: Tue May 24 12:36:51 2016
New Revision: 1745352
URL: http://svn.apache.org/viewvc?rev=1745352&view=rev
Log:
SLING-5737 - added light file based checkpointing
Added:
sling/trunk/contrib/extensions/distribution/core/src/main/java/org/apache/sling/distribution/queue/impl/simple/SimpleDistributionQueueCheckpoint.java (with props)
sling/trunk/contrib/extensions/distribution/core/src/main/java/org/apache/sling/distribution/queue/impl/simple/SimpleDistributionQueueProcessor.java
- copied, changed from r1745339, sling/trunk/contrib/extensions/distribution/core/src/main/java/org/apache/sling/distribution/queue/impl/simple/ScheduledDistributionQueueProcessorTask.java
sling/trunk/contrib/extensions/distribution/core/src/test/java/org/apache/sling/distribution/queue/impl/simple/SimpleDistributionQueueCheckpointTest.java (with props)
sling/trunk/contrib/extensions/distribution/core/src/test/java/org/apache/sling/distribution/queue/impl/simple/SimpleDistributionQueueProcessorTest.java
- copied, changed from r1745339, sling/trunk/contrib/extensions/distribution/core/src/test/java/org/apache/sling/distribution/queue/impl/simple/ScheduledDistributionQueueProcessorTaskTest.java
Removed:
sling/trunk/contrib/extensions/distribution/core/src/main/java/org/apache/sling/distribution/queue/impl/simple/ScheduledDistributionQueueProcessorTask.java
sling/trunk/contrib/extensions/distribution/core/src/test/java/org/apache/sling/distribution/queue/impl/simple/ScheduledDistributionQueueProcessorTaskTest.java
Modified:
sling/trunk/contrib/extensions/distribution/core/src/main/java/org/apache/sling/distribution/queue/impl/simple/SimpleDistributionQueue.java
sling/trunk/contrib/extensions/distribution/core/src/main/java/org/apache/sling/distribution/queue/impl/simple/SimpleDistributionQueueProvider.java
Modified: sling/trunk/contrib/extensions/distribution/core/src/main/java/org/apache/sling/distribution/queue/impl/simple/SimpleDistributionQueue.java
URL: http://svn.apache.org/viewvc/sling/trunk/contrib/extensions/distribution/core/src/main/java/org/apache/sling/distribution/queue/impl/simple/SimpleDistributionQueue.java?rev=1745352&r1=1745351&r2=1745352&view=diff
==============================================================================
--- sling/trunk/contrib/extensions/distribution/core/src/main/java/org/apache/sling/distribution/queue/impl/simple/SimpleDistributionQueue.java (original)
+++ sling/trunk/contrib/extensions/distribution/core/src/main/java/org/apache/sling/distribution/queue/impl/simple/SimpleDistributionQueue.java Tue May 24 12:36:51 2016
@@ -59,7 +59,7 @@ public class SimpleDistributionQueue imp
private final Map<DistributionQueueItem, DistributionQueueItemStatus> statusMap;
public SimpleDistributionQueue(String agentName, String name) {
- log.debug("starting a simple queue for agent {}", agentName);
+ log.debug("starting a simple queue {} for agent {}", name, agentName);
this.name = name;
this.queue = new LinkedBlockingQueue<DistributionQueueItem>();
this.statusMap = new WeakHashMap<DistributionQueueItem, DistributionQueueItemStatus>(10);
Added: sling/trunk/contrib/extensions/distribution/core/src/main/java/org/apache/sling/distribution/queue/impl/simple/SimpleDistributionQueueCheckpoint.java
URL: http://svn.apache.org/viewvc/sling/trunk/contrib/extensions/distribution/core/src/main/java/org/apache/sling/distribution/queue/impl/simple/SimpleDistributionQueueCheckpoint.java?rev=1745352&view=auto
==============================================================================
--- sling/trunk/contrib/extensions/distribution/core/src/main/java/org/apache/sling/distribution/queue/impl/simple/SimpleDistributionQueueCheckpoint.java (added)
+++ sling/trunk/contrib/extensions/distribution/core/src/main/java/org/apache/sling/distribution/queue/impl/simple/SimpleDistributionQueueCheckpoint.java Tue May 24 12:36:51 2016
@@ -0,0 +1,88 @@
+/*
+ * 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.sling.distribution.queue.impl.simple;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.StringWriter;
+import java.nio.charset.Charset;
+import java.util.Collection;
+import java.util.LinkedList;
+import java.util.Map;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.sling.commons.json.io.JSONWriter;
+import org.apache.sling.distribution.queue.DistributionQueue;
+import org.apache.sling.distribution.queue.DistributionQueueEntry;
+import org.apache.sling.distribution.queue.DistributionQueueItem;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A light checkpointing of a {@link SimpleDistributionQueue} to {@link File}
+ */
+class SimpleDistributionQueueCheckpoint implements Runnable {
+
+ private static final Logger log = LoggerFactory.getLogger(SimpleDistributionQueueCheckpoint.class);
+
+ private final DistributionQueue queue;
+ private final File checkpointDirectory;
+
+ public SimpleDistributionQueueCheckpoint(DistributionQueue queue, File checkpointDirectory) {
+ this.queue = queue;
+ this.checkpointDirectory = checkpointDirectory;
+ }
+
+ @Override
+ public void run() {
+ String fileName = queue.getName() + "-checkpoint";
+ File checkpointFile = new File(checkpointDirectory, fileName + "-new");
+ log.debug("started checkpointing");
+
+ try {
+ if (checkpointFile.exists()) {
+ assert checkpointFile.delete();
+ }
+ assert checkpointFile.createNewFile();
+ Collection<String> lines = new LinkedList<String>();
+ FileOutputStream fileOutputStream = new FileOutputStream(checkpointFile);
+ for (DistributionQueueEntry queueEntry : queue.getItems(0, -1)) {
+ DistributionQueueItem item = queueEntry.getItem();
+ String packageId = item.getPackageId();
+ StringWriter w = new StringWriter();
+ JSONWriter jsonWriter = new JSONWriter(w);
+ jsonWriter.object();
+ for (Map.Entry entry : item.entrySet()) {
+ jsonWriter.key(String.valueOf(entry.getKey()));
+ jsonWriter.value(String.valueOf(entry.getValue()));
+ }
+ jsonWriter.endObject();
+ lines.add(packageId + " " + w.toString());
+ }
+ log.debug("parsed {} items", lines.size());
+ IOUtils.writeLines(lines, Charset.defaultCharset().name(), fileOutputStream);
+ fileOutputStream.flush();
+ fileOutputStream.close();
+ boolean success = checkpointFile.renameTo(new File(checkpointDirectory, fileName));
+ log.debug("checkpoint succeeded: {}", success);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+}
Propchange: sling/trunk/contrib/extensions/distribution/core/src/main/java/org/apache/sling/distribution/queue/impl/simple/SimpleDistributionQueueCheckpoint.java
------------------------------------------------------------------------------
svn:eol-style = native
Copied: sling/trunk/contrib/extensions/distribution/core/src/main/java/org/apache/sling/distribution/queue/impl/simple/SimpleDistributionQueueProcessor.java (from r1745339, sling/trunk/contrib/extensions/distribution/core/src/main/java/org/apache/sling/distribution/queue/impl/simple/ScheduledDistributionQueueProcessorTask.java)
URL: http://svn.apache.org/viewvc/sling/trunk/contrib/extensions/distribution/core/src/main/java/org/apache/sling/distribution/queue/impl/simple/SimpleDistributionQueueProcessor.java?p2=sling/trunk/contrib/extensions/distribution/core/src/main/java/org/apache/sling/distribution/queue/impl/simple/SimpleDistributionQueueProcessor.java&p1=sling/trunk/contrib/extensions/distribution/core/src/main/java/org/apache/sling/distribution/queue/impl/simple/ScheduledDistributionQueueProcessorTask.java&r1=1745339&r2=1745352&rev=1745352&view=diff
==============================================================================
--- sling/trunk/contrib/extensions/distribution/core/src/main/java/org/apache/sling/distribution/queue/impl/simple/ScheduledDistributionQueueProcessorTask.java (original)
+++ sling/trunk/contrib/extensions/distribution/core/src/main/java/org/apache/sling/distribution/queue/impl/simple/SimpleDistributionQueueProcessor.java Tue May 24 12:36:51 2016
@@ -27,14 +27,14 @@ import org.slf4j.LoggerFactory;
/**
* a simple scheduled {@link SimpleDistributionQueue}s processor
*/
-class ScheduledDistributionQueueProcessorTask implements Runnable {
+class SimpleDistributionQueueProcessor implements Runnable {
private final Logger log = LoggerFactory.getLogger(getClass());
private final DistributionQueue queue;
private final DistributionQueueProcessor queueProcessor;
- public ScheduledDistributionQueueProcessorTask(DistributionQueue queue,
- DistributionQueueProcessor queueProcessor) {
+ public SimpleDistributionQueueProcessor(DistributionQueue queue,
+ DistributionQueueProcessor queueProcessor) {
this.queue = queue;
this.queueProcessor = queueProcessor;
}
Modified: sling/trunk/contrib/extensions/distribution/core/src/main/java/org/apache/sling/distribution/queue/impl/simple/SimpleDistributionQueueProvider.java
URL: http://svn.apache.org/viewvc/sling/trunk/contrib/extensions/distribution/core/src/main/java/org/apache/sling/distribution/queue/impl/simple/SimpleDistributionQueueProvider.java?rev=1745352&r1=1745351&r2=1745352&view=diff
==============================================================================
--- sling/trunk/contrib/extensions/distribution/core/src/main/java/org/apache/sling/distribution/queue/impl/simple/SimpleDistributionQueueProvider.java (original)
+++ sling/trunk/contrib/extensions/distribution/core/src/main/java/org/apache/sling/distribution/queue/impl/simple/SimpleDistributionQueueProvider.java Tue May 24 12:36:51 2016
@@ -19,13 +19,25 @@
package org.apache.sling.distribution.queue.impl.simple;
import javax.annotation.Nonnull;
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.FileReader;
+import java.io.FilenameFilter;
import java.util.Collection;
+import java.util.HashMap;
+import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.io.LineIterator;
+import org.apache.sling.commons.json.JSONException;
+import org.apache.sling.commons.json.JSONObject;
+import org.apache.sling.commons.json.JSONTokener;
import org.apache.sling.commons.scheduler.ScheduleOptions;
import org.apache.sling.commons.scheduler.Scheduler;
import org.apache.sling.distribution.queue.DistributionQueue;
+import org.apache.sling.distribution.queue.DistributionQueueItem;
import org.apache.sling.distribution.queue.DistributionQueueProcessor;
import org.apache.sling.distribution.queue.DistributionQueueProvider;
import org.slf4j.Logger;
@@ -45,15 +57,26 @@ public class SimpleDistributionQueueProv
private final Scheduler scheduler;
private final Map<String, SimpleDistributionQueue> queueMap = new ConcurrentHashMap<String, SimpleDistributionQueue>();
+ private File checkpointDirectory;
public SimpleDistributionQueueProvider(Scheduler scheduler, String name) {
if (name == null || scheduler == null) {
throw new IllegalArgumentException("all arguments are required");
}
+ this.checkpointDirectory = new File(name + "-simple-queues-checkpoints");
+ log.info("creating checkpoint directory {}", checkpointDirectory.getAbsoluteFile());
+ if (checkpointDirectory.exists() && !checkpointDirectory.isDirectory()) {
+ assert checkpointDirectory.delete();
+ }
+ boolean created = false;
+ if (!checkpointDirectory.exists()) {
+ created = checkpointDirectory.mkdir();
+ }
+ log.info("checkpoint directory created: {}, exists {}", created, checkpointDirectory.isDirectory() && checkpointDirectory.exists());
+
this.scheduler = scheduler;
this.name = name;
-
}
@Nonnull
@@ -70,28 +93,81 @@ public class SimpleDistributionQueueProv
return queue;
}
-
Collection<SimpleDistributionQueue> getQueues() {
return queueMap.values();
}
public void enableQueueProcessing(@Nonnull DistributionQueueProcessor queueProcessor, String... queueNames) {
+
+ // recover from checkpoints
+ log.debug("recovering from checkpoints if needed");
+ for (final String queueName : queueNames) {
+ log.debug("recovering for queue {}", queueName);
+ DistributionQueue queue = getQueue(queueName);
+ FilenameFilter filenameFilter = new FilenameFilter() {
+ @Override
+ public boolean accept(File file, String name) {
+ return name.equals(queueName + "-checkpoint");
+ }
+ };
+ for (File qf : checkpointDirectory.listFiles(filenameFilter)) {
+ log.info("recovering from checkpoint {}", qf);
+ try {
+ LineIterator lineIterator = IOUtils.lineIterator(new FileReader(qf));
+ while (lineIterator.hasNext()) {
+ String s = lineIterator.nextLine();
+ String[] split = s.split(" ");
+ String id = split[0];
+ String infoString = split[1];
+ Map<String, Object> info = new HashMap<String, Object>();
+ JSONTokener jsonTokener = new JSONTokener(infoString);
+ JSONObject jsonObject = new JSONObject(jsonTokener);
+ Iterator<String> keys = jsonObject.keys();
+ while (keys.hasNext()) {
+ String key = keys.next();
+ info.put(key, jsonObject.get(key));
+ }
+ queue.add(new DistributionQueueItem(id, info));
+ }
+ log.info("recovered {} items from queue {}", queue.getStatus().getItemsCount(), queueName);
+ } catch (FileNotFoundException e) {
+ log.warn("could not read checkpoint file {}", qf.getAbsolutePath());
+ } catch (JSONException e) {
+ log.warn("could not parse info from checkpoint file {}", qf.getAbsolutePath());
+ }
+ }
+ }
+
+ // enable processing
for (String queueName : queueNames) {
ScheduleOptions options = scheduler.NOW(-1, 1)
.canRunConcurrently(false)
.name(getJobName(queueName));
- scheduler.schedule(new ScheduledDistributionQueueProcessorTask(getQueue(queueName), queueProcessor), options);
+ scheduler.schedule(new SimpleDistributionQueueProcessor(getQueue(queueName), queueProcessor), options);
+ }
+
+ // enable checkpointing
+ for (String queueName : queueNames) {
+ ScheduleOptions options = scheduler.NOW(-1, 15)
+ .canRunConcurrently(false)
+ .name(getJobName(queueName + "-checkpoint"));
+ scheduler.schedule(new SimpleDistributionQueueCheckpoint(getQueue(queueName), checkpointDirectory), options);
}
}
public void disableQueueProcessing() {
for (DistributionQueue queue : getQueues()) {
- String queueName = queue.getName();
- if (scheduler.unschedule(getJobName(queueName))) {
- log.debug("queue processing on {} stopped", queue);
- } else {
- log.warn("could not disable queue processing on {}", queue);
- }
+ String queueName = queue.getName();
+ if (scheduler.unschedule(getJobName(queueName))) {
+ log.debug("queue processing on {} stopped", queue);
+ } else {
+ log.warn("could not disable queue processing on {}", queue);
+ }
+ if (scheduler.unschedule(getJobName(queueName) + "-checkpoint")) {
+ log.debug("checkpoint on {} stopped", queue);
+ } else {
+ log.warn("could not disable checkpoint on {}", queue);
+ }
}
}
Added: sling/trunk/contrib/extensions/distribution/core/src/test/java/org/apache/sling/distribution/queue/impl/simple/SimpleDistributionQueueCheckpointTest.java
URL: http://svn.apache.org/viewvc/sling/trunk/contrib/extensions/distribution/core/src/test/java/org/apache/sling/distribution/queue/impl/simple/SimpleDistributionQueueCheckpointTest.java?rev=1745352&view=auto
==============================================================================
--- sling/trunk/contrib/extensions/distribution/core/src/test/java/org/apache/sling/distribution/queue/impl/simple/SimpleDistributionQueueCheckpointTest.java (added)
+++ sling/trunk/contrib/extensions/distribution/core/src/test/java/org/apache/sling/distribution/queue/impl/simple/SimpleDistributionQueueCheckpointTest.java Tue May 24 12:36:51 2016
@@ -0,0 +1,71 @@
+/*
+ * 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.sling.distribution.queue.impl.simple;
+
+import java.io.File;
+import java.util.LinkedList;
+
+import org.apache.commons.collections.map.HashedMap;
+import org.apache.commons.io.FileUtils;
+import org.apache.sling.distribution.queue.DistributionQueue;
+import org.apache.sling.distribution.queue.DistributionQueueEntry;
+import org.apache.sling.distribution.queue.DistributionQueueItem;
+import org.apache.sling.distribution.queue.DistributionQueueItemState;
+import org.apache.sling.distribution.queue.DistributionQueueItemStatus;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests for {@link SimpleDistributionQueueCheckpoint}
+ */
+public class SimpleDistributionQueueCheckpointTest {
+
+ @Test
+ public void testRunWithNothingInQueue() throws Exception {
+ DistributionQueue queue = mock(DistributionQueue.class);
+ Iterable<DistributionQueueEntry> items = new LinkedList<DistributionQueueEntry>();
+ when(queue.getItems(0, -1)).thenReturn(items);
+ File checkpointDirectory = FileUtils.getTempDirectory();
+ SimpleDistributionQueueCheckpoint simpleDistributionQueueCheckpoint = new SimpleDistributionQueueCheckpoint(queue,
+ checkpointDirectory);
+ simpleDistributionQueueCheckpoint.run();
+ }
+
+ @Test
+ public void testRunWithOneItemInTheQueue() throws Exception {
+ DistributionQueue queue = mock(DistributionQueue.class);
+ String queueName = "sample-queue";
+ when(queue.getName()).thenReturn(queueName);
+ LinkedList<DistributionQueueEntry> entries = new LinkedList<DistributionQueueEntry>();
+ entries.add(new DistributionQueueEntry("123", new DistributionQueueItem("pid123", new HashedMap()),
+ new DistributionQueueItemStatus(DistributionQueueItemState.QUEUED, queueName)));
+ when(queue.getItems(0, -1)).thenReturn(entries);
+ File checkpointDirectory = FileUtils.getTempDirectory();
+ SimpleDistributionQueueCheckpoint simpleDistributionQueueCheckpoint = new SimpleDistributionQueueCheckpoint(queue,
+ checkpointDirectory);
+ simpleDistributionQueueCheckpoint.run();
+ File checkpointFile = new File(checkpointDirectory, "sample-queue-checkpoint");
+ assertTrue(checkpointFile.exists());
+ }
+}
\ No newline at end of file
Propchange: sling/trunk/contrib/extensions/distribution/core/src/test/java/org/apache/sling/distribution/queue/impl/simple/SimpleDistributionQueueCheckpointTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Copied: sling/trunk/contrib/extensions/distribution/core/src/test/java/org/apache/sling/distribution/queue/impl/simple/SimpleDistributionQueueProcessorTest.java (from r1745339, sling/trunk/contrib/extensions/distribution/core/src/test/java/org/apache/sling/distribution/queue/impl/simple/ScheduledDistributionQueueProcessorTaskTest.java)
URL: http://svn.apache.org/viewvc/sling/trunk/contrib/extensions/distribution/core/src/test/java/org/apache/sling/distribution/queue/impl/simple/SimpleDistributionQueueProcessorTest.java?p2=sling/trunk/contrib/extensions/distribution/core/src/test/java/org/apache/sling/distribution/queue/impl/simple/SimpleDistributionQueueProcessorTest.java&p1=sling/trunk/contrib/extensions/distribution/core/src/test/java/org/apache/sling/distribution/queue/impl/simple/ScheduledDistributionQueueProcessorTaskTest.java&r1=1745339&r2=1745352&rev=1745352&view=diff
==============================================================================
--- sling/trunk/contrib/extensions/distribution/core/src/test/java/org/apache/sling/distribution/queue/impl/simple/ScheduledDistributionQueueProcessorTaskTest.java (original)
+++ sling/trunk/contrib/extensions/distribution/core/src/test/java/org/apache/sling/distribution/queue/impl/simple/SimpleDistributionQueueProcessorTest.java Tue May 24 12:36:51 2016
@@ -31,17 +31,17 @@ import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
/**
- * Testcase for {@link ScheduledDistributionQueueProcessorTask}
+ * Testcase for {@link SimpleDistributionQueueProcessor}
*/
-public class ScheduledDistributionQueueProcessorTaskTest {
+public class SimpleDistributionQueueProcessorTest {
@Test
public void testRunWithNoQueue() throws Exception {
DistributionQueue queue = mock(DistributionQueue.class);
DistributionQueueProcessor queueProcessor = mock(DistributionQueueProcessor.class);
- ScheduledDistributionQueueProcessorTask scheduledDistributionQueueProcessorTask = new ScheduledDistributionQueueProcessorTask(
+ SimpleDistributionQueueProcessor simpleDistributionQueueProcessor = new SimpleDistributionQueueProcessor(
queue, queueProcessor);
- scheduledDistributionQueueProcessorTask.run();
+ simpleDistributionQueueProcessor.run();
}
@Test
@@ -53,9 +53,9 @@ public class ScheduledDistributionQueueP
queues.add(queue);
when(queueProvider.getQueues()).thenReturn(queues);
DistributionQueueProcessor queueProcessor = mock(DistributionQueueProcessor.class);
- ScheduledDistributionQueueProcessorTask scheduledDistributionQueueProcessorTask = new ScheduledDistributionQueueProcessorTask(
+ SimpleDistributionQueueProcessor simpleDistributionQueueProcessor = new SimpleDistributionQueueProcessor(
queue, queueProcessor);
- scheduledDistributionQueueProcessorTask.run();
+ simpleDistributionQueueProcessor.run();
}
@Test
@@ -69,8 +69,8 @@ public class ScheduledDistributionQueueP
queues.add(queue);
when(queueProvider.getQueues()).thenReturn(queues);
DistributionQueueProcessor queueProcessor = mock(DistributionQueueProcessor.class);
- ScheduledDistributionQueueProcessorTask scheduledDistributionQueueProcessorTask = new ScheduledDistributionQueueProcessorTask(
+ SimpleDistributionQueueProcessor simpleDistributionQueueProcessor = new SimpleDistributionQueueProcessor(
queue, queueProcessor);
- scheduledDistributionQueueProcessorTask.run();
+ simpleDistributionQueueProcessor.run();
}
}
\ No newline at end of file