SLING-7754: implementation of Resource Based Queues
diff --git a/pom.xml b/pom.xml
index 6bd547f..8f62adf 100644
--- a/pom.xml
+++ b/pom.xml
@@ -48,6 +48,10 @@
<tag>HEAD</tag>
</scm>
+ <properties>
+ <exam.version>4.11.0</exam.version>
+ </properties>
+
<!-- ======================================================================= -->
<!-- B U I L D -->
<!-- ======================================================================= -->
@@ -94,6 +98,9 @@
<artifactId>maven-surefire-plugin</artifactId>
<configuration>
<redirectTestOutputToFile>true</redirectTestOutputToFile>
+ <excludes>
+ <exclude>**/*IT.java</exclude>
+ </excludes>
</configuration>
</plugin>
<plugin>
@@ -112,6 +119,35 @@
</excludes>
</configuration>
</plugin>
+
+ <plugin>
+ <groupId>org.apache.servicemix.tooling</groupId>
+ <artifactId>depends-maven-plugin</artifactId>
+ <version>1.4.0</version>
+ <executions>
+ <execution>
+ <goals>
+ <goal>generate-depends-file</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-failsafe-plugin</artifactId>
+ <configuration>
+ <redirectTestOutputToFile>true</redirectTestOutputToFile>
+ <systemProperties>
+ <property>
+ <name>bundle.filename</name>
+ <value>${basedir}/target/${project.build.finalName}.jar</value>
+ </property>
+ </systemProperties>
+ <includes>
+ <include>**/*IT.java</include>
+ </includes>
+ </configuration>
+ </plugin>
</plugins>
</build>
@@ -310,6 +346,45 @@
<scope>provided</scope>
</dependency>
+
+ <!-- Pax Exam -->
+ <dependency>
+ <groupId>org.apache.sling</groupId>
+ <artifactId>org.apache.sling.testing.paxexam</artifactId>
+ <version>2.0.0</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.ops4j.pax.exam</groupId>
+ <artifactId>pax-exam</artifactId>
+ <version>${exam.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.ops4j.pax.exam</groupId>
+ <artifactId>pax-exam-cm</artifactId>
+ <version>${exam.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.ops4j.pax.exam</groupId>
+ <artifactId>pax-exam-container-forked</artifactId>
+ <version>${exam.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.ops4j.pax.exam</groupId>
+ <artifactId>pax-exam-junit4</artifactId>
+ <version>${exam.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.ops4j.pax.exam</groupId>
+ <artifactId>pax-exam-link-mvn</artifactId>
+ <version>${exam.version}</version>
+ <scope>test</scope>
+ </dependency>
+
</dependencies>
<reporting>
<plugins>
diff --git a/src/main/java/org/apache/sling/distribution/agent/impl/QueueDistributionAgentFactory.java b/src/main/java/org/apache/sling/distribution/agent/impl/QueueDistributionAgentFactory.java
index 7f3c507..3d0ad64 100644
--- a/src/main/java/org/apache/sling/distribution/agent/impl/QueueDistributionAgentFactory.java
+++ b/src/main/java/org/apache/sling/distribution/agent/impl/QueueDistributionAgentFactory.java
@@ -46,9 +46,10 @@
import org.apache.sling.distribution.packaging.impl.exporter.LocalDistributionPackageExporter;
import org.apache.sling.distribution.queue.impl.DistributionQueueProvider;
import org.apache.sling.distribution.queue.impl.DistributionQueueDispatchingStrategy;
+import org.apache.sling.distribution.queue.impl.DistributionQueueProviderFactory;
import org.apache.sling.distribution.queue.impl.PriorityQueueDispatchingStrategy;
import org.apache.sling.distribution.queue.impl.SingleQueueDispatchingStrategy;
-import org.apache.sling.distribution.queue.impl.jobhandling.JobHandlingDistributionQueueProvider;
+import org.apache.sling.distribution.queue.impl.resource.ResourceQueueProvider;
import org.apache.sling.distribution.trigger.DistributionTrigger;
import org.apache.sling.event.jobs.JobManager;
import org.apache.sling.jcr.api.SlingRepository;
@@ -111,6 +112,12 @@
private DistributionRequestAuthorizationStrategy requestAuthorizationStrategy;
+ @Property(name = "queueProviderFactory.target", label = "Queue Provider Factory", description = "The target reference for the DistributionQueueProviderFactory used to build queues," +
+ "e.g. use target=(name=...) to bind to services by name.", value = "(name=jobQueue)")
+ @Reference(name = "queueProviderFactory")
+ private DistributionQueueProviderFactory queueProviderFactory;
+
+
@Property(name = "packageBuilder.target", label = "Package Builder", description = "The target reference for the DistributionPackageBuilder used to create distribution packages, " +
"e.g. use target=(name=...) to bind to services by name.", value = SettingsUtils.COMPONENT_NAME_DEFAULT)
@Reference(name = "packageBuilder")
@@ -135,14 +142,13 @@
private SlingSettingsService settingsService;
@Reference
- private JobManager jobManager;
-
- @Reference
private ResourceResolverFactory resourceResolverFactory;
@Reference
private SlingRepository slingRepository;
+ DistributionQueueProvider queueProvider;
+
public QueueDistributionAgentFactory() {
super(QueueDistributionAgentMBean.class);
}
@@ -164,6 +170,8 @@
@Deactivate
protected void deactivate(BundleContext context) {
super.deactivate(context);
+
+ queueProviderFactory.releaseProvider(queueProvider);
}
@Override
@@ -177,8 +185,9 @@
Map<String, String> priorityQueues = PropertiesUtil.toMap(config.get(PRIORITY_QUEUES), new String[0]);
priorityQueues = SettingsUtils.removeEmptyEntries(priorityQueues);
+ queueProvider = queueProviderFactory.getProvider(agentName, serviceName);
- DistributionQueueProvider queueProvider = new MonitoringDistributionQueueProvider(new JobHandlingDistributionQueueProvider(agentName, jobManager, context), context);
+ MonitoringDistributionQueueProvider monitoringQueueProvider = new MonitoringDistributionQueueProvider(queueProvider, context);
DistributionQueueDispatchingStrategy exportQueueStrategy = null;
@@ -194,7 +203,7 @@
return new SimpleDistributionAgent(agentName, false, null,
serviceName, null, packageExporter, requestAuthorizationStrategy,
- queueProvider, exportQueueStrategy, null, distributionEventFactory, resourceResolverFactory, slingRepository,
+ monitoringQueueProvider, exportQueueStrategy, null, distributionEventFactory, resourceResolverFactory, slingRepository,
distributionLog, allowedRequests, allowedRoots, 0);
}
diff --git a/src/main/java/org/apache/sling/distribution/queue/impl/DistributionQueueProviderFactory.java b/src/main/java/org/apache/sling/distribution/queue/impl/DistributionQueueProviderFactory.java
new file mode 100644
index 0000000..ee84d5c
--- /dev/null
+++ b/src/main/java/org/apache/sling/distribution/queue/impl/DistributionQueueProviderFactory.java
@@ -0,0 +1,28 @@
+/*
+ * 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;
+
+public interface DistributionQueueProviderFactory {
+
+ DistributionQueueProvider getProvider(String agentName, String serviceName);
+
+ void releaseProvider(DistributionQueueProvider queueProvider);
+
+}
diff --git a/src/main/java/org/apache/sling/distribution/queue/impl/jobhandling/JobHandlingQueueDistributionProviderFactory.java b/src/main/java/org/apache/sling/distribution/queue/impl/jobhandling/JobHandlingQueueDistributionProviderFactory.java
new file mode 100644
index 0000000..d6a450d
--- /dev/null
+++ b/src/main/java/org/apache/sling/distribution/queue/impl/jobhandling/JobHandlingQueueDistributionProviderFactory.java
@@ -0,0 +1,61 @@
+/*
+ * 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.jobhandling;
+
+import org.apache.felix.scr.annotations.Activate;
+import org.apache.felix.scr.annotations.Component;
+import org.apache.felix.scr.annotations.Property;
+import org.apache.felix.scr.annotations.Reference;
+import org.apache.felix.scr.annotations.Service;
+import org.apache.sling.distribution.component.impl.DistributionComponentConstants;
+import org.apache.sling.distribution.queue.impl.DistributionQueueProvider;
+import org.apache.sling.distribution.queue.impl.DistributionQueueProviderFactory;
+import org.apache.sling.event.jobs.JobManager;
+import org.osgi.framework.BundleContext;
+
+import java.util.Map;
+
+@Component
+@Service(DistributionQueueProviderFactory.class)
+@Property(name = DistributionComponentConstants.PN_NAME, value = "jobQueue")
+public class JobHandlingQueueDistributionProviderFactory implements DistributionQueueProviderFactory {
+
+ @Reference
+ JobManager jobManager;
+
+ BundleContext context;
+
+
+ @Activate
+ protected void activate(BundleContext context, Map<String, Object> config)
+ {
+ this.context = context;
+ }
+
+ @Override
+ public DistributionQueueProvider getProvider(String agentName, String serviceName) {
+ return new JobHandlingDistributionQueueProvider(agentName, jobManager, context);
+ }
+
+ @Override
+ public void releaseProvider(DistributionQueueProvider queueProvider) {
+
+ }
+}
diff --git a/src/main/java/org/apache/sling/distribution/queue/impl/resource/ResourceIterator.java b/src/main/java/org/apache/sling/distribution/queue/impl/resource/ResourceIterator.java
new file mode 100644
index 0000000..4236d56
--- /dev/null
+++ b/src/main/java/org/apache/sling/distribution/queue/impl/resource/ResourceIterator.java
@@ -0,0 +1,123 @@
+/*
+ * 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.resource;
+
+
+import org.apache.sling.api.resource.Resource;
+
+import java.util.Iterator;
+import java.util.Stack;
+
+public class ResourceIterator implements Iterator<Resource> {
+
+ private final String folderResourceType;
+ private final boolean includeFolders;
+ private final boolean includeLeafs;
+ private final boolean includeRoot = false;
+ private final String rootPath;
+
+ private Stack<Iterator<Resource>> folderIterators = new Stack<Iterator<Resource>>();
+
+ private Resource currentFolder;
+ private Iterator<Resource> currentIterator;
+
+ private Resource next;
+
+
+ public ResourceIterator(Resource root, String folderResourceType, boolean includeFolders, boolean includeLeafs) {
+ this.folderResourceType = folderResourceType;
+ this.includeFolders = includeFolders;
+ this.includeLeafs = includeLeafs;
+ this.rootPath = root.getPath();
+
+ currentFolder = root;
+ currentIterator = currentFolder.listChildren();
+ next = seek();
+ }
+
+ @Override
+ public boolean hasNext() {
+ return next != null;
+ }
+
+ @Override
+ public Resource next() {
+ Resource result = next;
+ next = seek();
+ return result;
+ }
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+
+ Resource seek() {
+ Resource res;
+ while ((res = seekAll()) != null) {
+
+ if (rootPath.equals(res.getPath())) {
+ if (includeRoot) {
+ return res;
+ } else {
+ continue;
+ }
+ }
+
+ if (includeFolders && res.isResourceType(folderResourceType)) {
+ return res;
+ }
+
+ if (includeLeafs && !res.isResourceType(folderResourceType)) {
+ return res;
+ }
+ }
+
+ return null;
+ }
+
+ // depth first, post order (children before parents)
+ Resource seekAll() {
+
+ while (currentIterator != null) {
+ if (currentIterator.hasNext()) {
+ Resource res = currentIterator.next();
+
+ if (res.isResourceType(folderResourceType)) {
+ folderIterators.push(currentIterator);
+
+ currentFolder = res;
+ currentIterator = currentFolder.listChildren();
+ } else {
+ return res;
+ }
+ } else {
+ Resource folder = currentFolder;
+
+ currentFolder = currentFolder.getParent();
+ currentIterator = folderIterators.empty() ? null : folderIterators.pop();
+
+ return folder;
+ }
+ }
+
+ return null;
+ }
+}
diff --git a/src/main/java/org/apache/sling/distribution/queue/impl/resource/ResourceQueue.java b/src/main/java/org/apache/sling/distribution/queue/impl/resource/ResourceQueue.java
new file mode 100644
index 0000000..28162c3
--- /dev/null
+++ b/src/main/java/org/apache/sling/distribution/queue/impl/resource/ResourceQueue.java
@@ -0,0 +1,231 @@
+/*
+ * 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.resource;
+
+import org.apache.sling.api.resource.LoginException;
+import org.apache.sling.api.resource.PersistenceException;
+import org.apache.sling.api.resource.Resource;
+import org.apache.sling.api.resource.ResourceResolver;
+import org.apache.sling.api.resource.ResourceResolverFactory;
+import org.apache.sling.distribution.queue.DistributionQueueEntry;
+import org.apache.sling.distribution.queue.DistributionQueueItem;
+import org.apache.sling.distribution.queue.DistributionQueueState;
+import org.apache.sling.distribution.queue.DistributionQueueStatus;
+import org.apache.sling.distribution.queue.DistributionQueueType;
+import org.apache.sling.distribution.queue.spi.DistributionQueue;
+import org.apache.sling.distribution.util.impl.DistributionUtils;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+
+
+public class ResourceQueue implements DistributionQueue {
+ private final Logger log = LoggerFactory.getLogger(getClass());
+
+
+ private final ResourceResolverFactory resolverFactory;
+ private String serviceName;
+ private String queueName;
+ private final String queueRootPath;
+
+ public ResourceQueue(ResourceResolverFactory resolverFactory, String serviceName, String queueName, String rootPath) {
+ this.resolverFactory = resolverFactory;
+ this.serviceName = serviceName;
+ this.queueName = queueName;
+ this.queueRootPath = rootPath + "/" + queueName;
+ }
+
+ @NotNull
+ @Override
+ public String getName() {
+ return queueName;
+ }
+
+ @Nullable
+ @Override
+ public DistributionQueueEntry add(@NotNull DistributionQueueItem item) {
+
+ ResourceResolver resourceResolver = null;
+ try {
+ resourceResolver = DistributionUtils.loginService(resolverFactory, serviceName);
+
+ Resource queueRoot = ResourceQueueUtils.getRootResource(resourceResolver, queueRootPath);
+
+ Resource resource = ResourceQueueUtils.createResource(queueRoot, item);
+
+ DistributionQueueEntry entry = ResourceQueueUtils.readEntry(queueRoot, resource);
+
+ logEntry(entry, "add");
+
+ return entry;
+
+ } catch (LoginException e) {
+ throw new RuntimeException(e);
+ } catch (PersistenceException e) {
+ throw new RuntimeException(e);
+ } finally {
+ DistributionUtils.safelyLogout(resourceResolver);
+ }
+ }
+
+
+ @Override
+ public DistributionQueueEntry getHead() {
+ ResourceResolver resourceResolver = null;
+ try {
+ resourceResolver = DistributionUtils.loginService(resolverFactory, serviceName);
+ Resource queueRoot = ResourceQueueUtils.getRootResource(resourceResolver, queueRootPath);
+
+
+ DistributionQueueEntry head = ResourceQueueUtils.getHead(queueRoot);
+
+ logEntry(head, "getHead");
+
+ return head;
+
+ } catch (LoginException e) {
+ throw new RuntimeException(e);
+ } catch (PersistenceException e) {
+ throw new RuntimeException(e);
+ } finally {
+ DistributionUtils.safelyLogout(resourceResolver);
+ }
+ }
+
+ @NotNull
+ @Override
+ public Iterable<DistributionQueueEntry> getItems(int skip, int limit) {
+ ResourceResolver resourceResolver = null;
+ try {
+ resourceResolver = DistributionUtils.loginService(resolverFactory, serviceName);
+ Resource queueRoot = ResourceQueueUtils.getRootResource(resourceResolver, queueRootPath);
+
+ List<DistributionQueueEntry> entries = ResourceQueueUtils.getEntries(queueRoot, skip, limit);
+
+ log.debug("queue[{}] getItems entries={}", new Object[] { queueName, entries.size() });
+
+ return entries;
+ } catch (LoginException e) {
+ throw new RuntimeException(e);
+ } catch (PersistenceException e) {
+ throw new RuntimeException(e);
+ } finally {
+ DistributionUtils.safelyLogout(resourceResolver);
+ }
+
+ }
+
+ @Nullable
+ @Override
+ public DistributionQueueEntry getItem(@NotNull String itemId) {
+ ResourceResolver resourceResolver = null;
+ try {
+ resourceResolver = DistributionUtils.loginService(resolverFactory, serviceName);
+ Resource queueRoot = ResourceQueueUtils.getRootResource(resourceResolver, queueRootPath);
+
+ Resource itemResource = ResourceQueueUtils.getResourceById(queueRoot, itemId);
+
+ DistributionQueueEntry entry = ResourceQueueUtils.readEntry(queueRoot, itemResource);
+
+ logEntry(entry, "getItem");
+
+ return entry;
+
+ } catch (LoginException e) {
+ throw new RuntimeException(e);
+ } catch (PersistenceException e) {
+ throw new RuntimeException(e);
+ } finally {
+ DistributionUtils.safelyLogout(resourceResolver);
+ }
+ }
+
+ @Nullable
+ @Override
+ public DistributionQueueEntry remove(@NotNull String itemId) {
+ ResourceResolver resourceResolver = null;
+ try {
+ resourceResolver = DistributionUtils.loginService(resolverFactory, serviceName);
+ Resource queueRoot = ResourceQueueUtils.getRootResource(resourceResolver, queueRootPath);
+
+ Resource itemResource = ResourceQueueUtils.getResourceById(queueRoot, itemId);
+
+ DistributionQueueEntry entry = ResourceQueueUtils.readEntry(queueRoot, itemResource);
+
+ ResourceQueueUtils.deleteResource(itemResource);
+
+ logEntry(entry, "remove");
+
+ return entry;
+
+ } catch (LoginException e) {
+ throw new RuntimeException(e);
+ } catch (PersistenceException e) {
+ throw new RuntimeException(e);
+ } finally {
+ DistributionUtils.safelyLogout(resourceResolver);
+ }
+ }
+
+ @Nullable
+ @Override
+ public DistributionQueueStatus getStatus() {
+ ResourceResolver resourceResolver = null;
+ try {
+ resourceResolver = DistributionUtils.loginService(resolverFactory, serviceName);
+ Resource queueRoot = ResourceQueueUtils.getRootResource(resourceResolver, queueRootPath);
+
+ int count = ResourceQueueUtils.getResourceCount(queueRoot);
+
+ return new DistributionQueueStatus(count, DistributionQueueState.PASSIVE);
+ } catch (LoginException e) {
+ throw new RuntimeException(e);
+ } catch (PersistenceException e) {
+ throw new RuntimeException(e);
+ } finally {
+ DistributionUtils.safelyLogout(resourceResolver);
+ }
+ }
+
+ @NotNull
+ @Override
+ public DistributionQueueType getType() {
+ return DistributionQueueType.ORDERED;
+ }
+
+ void logEntry(DistributionQueueEntry entry, String scope) {
+ if (entry == null) {
+ log.debug("queue[{}] {} null entry", new Object[] { queueName, scope });
+ return;
+ }
+
+ if (entry.getItem() == null) {
+ log.debug("queue[{}] {} null item (should not happen)", new Object[] { queueName, scope });
+ return;
+ }
+
+ String entryId = entry.getId();
+ DistributionQueueItem item = entry.getItem();
+ log.debug("queue[{}] {} entryId={} packageId={}", new Object[] { queueName, scope, entryId, item.getPackageId() });
+ }
+}
diff --git a/src/main/java/org/apache/sling/distribution/queue/impl/resource/ResourceQueueCleanupTask.java b/src/main/java/org/apache/sling/distribution/queue/impl/resource/ResourceQueueCleanupTask.java
new file mode 100644
index 0000000..f41b16e
--- /dev/null
+++ b/src/main/java/org/apache/sling/distribution/queue/impl/resource/ResourceQueueCleanupTask.java
@@ -0,0 +1,112 @@
+/*
+ * 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.resource;
+
+import org.apache.sling.api.resource.PersistenceException;
+import org.apache.sling.api.resource.Resource;
+import org.apache.sling.api.resource.ResourceResolver;
+import org.apache.sling.api.resource.ResourceResolverFactory;
+import org.apache.sling.distribution.util.impl.DistributionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Calendar;
+import java.util.Iterator;
+
+
+public class ResourceQueueCleanupTask implements Runnable {
+ private final Logger log = LoggerFactory.getLogger(getClass());
+
+
+ private final ResourceResolverFactory resolverFactory;
+ private final String serviceName;
+ private final String rootPath;
+
+ public ResourceQueueCleanupTask(ResourceResolverFactory resolverFactory, String serviceName, String rootPath) {
+
+ this.resolverFactory = resolverFactory;
+ this.serviceName = serviceName;
+ this.rootPath = rootPath;
+ }
+
+ @Override
+ public void run() {
+ log.debug("Cleaning up resource queues at {}", rootPath);
+
+ ResourceResolver resourceResolver = null;
+ try {
+ resourceResolver = DistributionUtils.loginService(resolverFactory, serviceName);
+ Resource root = ResourceQueueUtils.getRootResource(resourceResolver, rootPath);
+
+ Iterator<Resource> it = root.listChildren();
+
+ while (it.hasNext()) {
+ Resource queueRoot = it.next();
+
+ log.debug("Starting cleaning up queue at {}", queueRoot.getPath());
+
+ removeEmptyFolders(queueRoot);
+
+ log.debug("Finished cleaning up queue at {}", queueRoot.getPath());
+ }
+ } catch (Throwable e) {
+ log.error("Error cleaning up resource queues", e);
+ } finally {
+ DistributionUtils.safelyLogout(resourceResolver);
+ }
+ }
+
+
+ public void removeEmptyFolders(Resource root) throws PersistenceException {
+ Calendar now = Calendar.getInstance();
+ now.add(Calendar.MINUTE, -5);
+
+ ResourceResolver resolver = root.getResourceResolver();
+ ResourceIterator it = new ResourceIterator(root, ResourceQueueUtils.RESOURCE_FOLDER, true, false);
+
+ String nowPath = ResourceQueueUtils.getTimePath(now);
+
+ while (it.hasNext()) {
+ Resource res = it.next();
+
+ String resPath = res.getPath().substring(root.getPath().length()+1);
+
+ if (!res.isResourceType(ResourceQueueUtils.RESOURCE_FOLDER)) {
+ continue;
+ }
+
+ // now 2018/03/15/03/48
+ // res 2018/02/15
+ if (!ResourceQueueUtils.isSafeToDelete(nowPath, resPath)) {
+ continue;
+ }
+
+ if (res.hasChildren()) {
+ continue;
+ }
+
+ log.debug("removing empty folder {}", res.getPath());
+
+ resolver.delete(res);
+ resolver.commit();
+ }
+
+ }
+}
diff --git a/src/main/java/org/apache/sling/distribution/queue/impl/resource/ResourceQueueProvider.java b/src/main/java/org/apache/sling/distribution/queue/impl/resource/ResourceQueueProvider.java
new file mode 100644
index 0000000..9c3c393
--- /dev/null
+++ b/src/main/java/org/apache/sling/distribution/queue/impl/resource/ResourceQueueProvider.java
@@ -0,0 +1,92 @@
+/*
+ * 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.resource;
+
+import org.apache.sling.api.resource.ResourceResolverFactory;
+import org.apache.sling.commons.scheduler.Scheduler;
+import org.apache.sling.distribution.common.DistributionException;
+import org.apache.sling.distribution.queue.DistributionQueueType;
+import org.apache.sling.distribution.queue.impl.DistributionQueueProcessor;
+import org.apache.sling.distribution.queue.impl.DistributionQueueProvider;
+import org.apache.sling.distribution.queue.spi.DistributionQueue;
+
+import org.jetbrains.annotations.NotNull;
+import org.osgi.framework.BundleContext;
+import org.osgi.framework.ServiceRegistration;
+
+import java.util.Dictionary;
+import java.util.Hashtable;
+
+public class ResourceQueueProvider implements DistributionQueueProvider {
+
+ private final static String QUEUES_ROOT = "/var/sling/distribution/queues/";
+ private ResourceResolverFactory resolverFactory;
+ private String serviceName;
+ private String agentRootPath;
+
+ private ServiceRegistration cleanupTask;
+
+
+ public ResourceQueueProvider(BundleContext context, ResourceResolverFactory resolverFactory, String serviceName, String agentName) {
+ this.resolverFactory = resolverFactory;
+ this.serviceName = serviceName;
+ this.agentRootPath = QUEUES_ROOT + agentName;
+
+ register(context);
+ }
+
+ @NotNull
+ @Override
+ public DistributionQueue getQueue(@NotNull String queueName) throws DistributionException {
+ return new ResourceQueue(resolverFactory, serviceName, queueName, agentRootPath);
+ }
+
+ @NotNull
+ @Override
+ public DistributionQueue getQueue(@NotNull String queueName, @NotNull DistributionQueueType type) {
+ return new ResourceQueue(resolverFactory, serviceName, queueName, agentRootPath);
+ }
+
+ @Override
+ public void enableQueueProcessing(@NotNull DistributionQueueProcessor queueProcessor, String... queueNames) throws DistributionException {
+ // processing not supported
+ }
+
+ @Override
+ public void disableQueueProcessing() throws DistributionException {
+ // processing not supported
+ }
+
+
+ private void register(BundleContext context) {
+ Runnable cleanup = new ResourceQueueCleanupTask(resolverFactory, serviceName, agentRootPath);
+ Dictionary<String, Object> props = new Hashtable<String, Object>();
+ props.put(Scheduler.PROPERTY_SCHEDULER_CONCURRENT, false);
+ props.put(Scheduler.PROPERTY_SCHEDULER_PERIOD, 300L);
+ cleanupTask = context.registerService(Runnable.class.getName(), cleanup, props);
+ }
+
+ public void close() {
+ if (cleanupTask != null) {
+ cleanupTask.unregister();
+ cleanupTask = null;
+ }
+ }
+}
diff --git a/src/main/java/org/apache/sling/distribution/queue/impl/resource/ResourceQueueProviderFactory.java b/src/main/java/org/apache/sling/distribution/queue/impl/resource/ResourceQueueProviderFactory.java
new file mode 100644
index 0000000..ac0bafd
--- /dev/null
+++ b/src/main/java/org/apache/sling/distribution/queue/impl/resource/ResourceQueueProviderFactory.java
@@ -0,0 +1,62 @@
+/*
+ * 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.resource;
+
+import org.apache.felix.scr.annotations.Activate;
+import org.apache.felix.scr.annotations.Component;
+import org.apache.felix.scr.annotations.Property;
+import org.apache.felix.scr.annotations.Reference;
+import org.apache.felix.scr.annotations.Service;
+import org.apache.sling.api.resource.ResourceResolverFactory;
+import org.apache.sling.distribution.component.impl.DistributionComponentConstants;
+import org.apache.sling.distribution.queue.impl.DistributionQueueProvider;
+import org.apache.sling.distribution.queue.impl.DistributionQueueProviderFactory;
+import org.osgi.framework.BundleContext;
+
+import java.util.Map;
+
+@Component
+@Service(DistributionQueueProviderFactory.class)
+@Property(name = DistributionComponentConstants.PN_NAME, value = "resourceQueue")
+public class ResourceQueueProviderFactory implements DistributionQueueProviderFactory {
+
+ @Reference
+ ResourceResolverFactory resourceResolverFactory;
+
+ BundleContext context;
+
+ @Activate
+ protected void activate(BundleContext context, Map<String, Object> config)
+ {
+ this.context = context;
+ }
+
+ @Override
+ public DistributionQueueProvider getProvider(String agentName, String serviceName) {
+ return new ResourceQueueProvider(context, resourceResolverFactory, serviceName, agentName);
+ }
+
+ @Override
+ public void releaseProvider(DistributionQueueProvider queueProvider) {
+ if (queueProvider instanceof ResourceQueueProvider) {
+ ((ResourceQueueProvider) queueProvider).close();
+ }
+ }
+}
\ No newline at end of file
diff --git a/src/main/java/org/apache/sling/distribution/queue/impl/resource/ResourceQueueUtils.java b/src/main/java/org/apache/sling/distribution/queue/impl/resource/ResourceQueueUtils.java
new file mode 100644
index 0000000..1408cb6
--- /dev/null
+++ b/src/main/java/org/apache/sling/distribution/queue/impl/resource/ResourceQueueUtils.java
@@ -0,0 +1,416 @@
+/*
+ * 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.resource;
+
+
+import org.apache.sling.api.resource.PersistenceException;
+import org.apache.sling.api.resource.Resource;
+import org.apache.sling.api.resource.ResourceResolver;
+import org.apache.sling.api.resource.ResourceUtil;
+import org.apache.sling.api.resource.ValueMap;
+import org.apache.sling.distribution.DistributionRequestType;
+import org.apache.sling.distribution.packaging.DistributionPackageInfo;
+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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+import javax.jcr.NodeIterator;
+import javax.jcr.RepositoryException;
+import javax.jcr.Session;
+import javax.jcr.query.Query;
+import javax.jcr.query.QueryManager;
+import javax.jcr.query.QueryResult;
+import java.text.SimpleDateFormat;
+import java.util.ArrayList;
+import java.util.Calendar;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicLong;
+
+public class ResourceQueueUtils {
+
+ // prefix for queue entry ids
+ private static final String ID_START = "distrq-";
+
+ // resource folder for queue roots
+ private static final String RESOURCE_ROOT = "sling:Folder";
+
+ // resource type for internal ordered folders
+ public static final String RESOURCE_FOLDER = "sling:OrderedFolder";
+
+ // resource type for internal entries
+ private static final String RESOURCE_ITEM = "nt:unstructured";
+
+ private static final String DISTRIBUTION_PACKAGE_PREFIX = "distribution.";
+ private static final String DISTRIBUTION_PACKAGE_ID = DISTRIBUTION_PACKAGE_PREFIX + "item.id";
+ private static final String DISTRIBUTION_PACKAGE_SIZE = DISTRIBUTION_PACKAGE_PREFIX + "package.size";
+
+
+ private static final AtomicLong itemCounter = new AtomicLong(0);
+ private static final Logger log = LoggerFactory.getLogger(ResourceQueueUtils.class);
+
+
+ private static Map<String, Object> serializeItem(DistributionQueueItem queueItem) {
+
+ Map<String, Object> properties = new HashMap<String, Object>();
+
+ for (String key : queueItem.keySet()) {
+ Object value = queueItem.get(key);
+
+ if (DistributionPackageInfo.PROPERTY_REQUEST_TYPE.equals(key)) {
+ if (value instanceof DistributionRequestType) {
+ value = ((DistributionRequestType) value).name();
+ }
+ }
+
+ if (value != null) {
+ properties.put(DISTRIBUTION_PACKAGE_PREFIX + key, value);
+ }
+ }
+
+ properties.put(DISTRIBUTION_PACKAGE_ID, queueItem.getPackageId());
+ properties.put(DISTRIBUTION_PACKAGE_SIZE, queueItem.getSize());
+
+ return properties;
+ }
+
+ private static DistributionQueueItem deserializeItem(ValueMap valueMap) {
+
+ String packageId = valueMap.get(DISTRIBUTION_PACKAGE_ID, String.class);
+ Long sizeProperty = valueMap.get(DISTRIBUTION_PACKAGE_SIZE, Long.class);
+ long size = sizeProperty == null ? -1 : sizeProperty;
+
+ Map<String, Object> properties = new HashMap<String, Object>();
+
+ for (String key : valueMap.keySet()) {
+ if (key.startsWith(DISTRIBUTION_PACKAGE_PREFIX)) {
+ String infoKey = key.substring(DISTRIBUTION_PACKAGE_PREFIX.length());
+ Object value = valueMap.get(key);
+
+ if (DistributionPackageInfo.PROPERTY_REQUEST_TYPE.equals(infoKey)) {
+ if (value instanceof String) {
+ value = DistributionRequestType.valueOf((String) value);
+ }
+ }
+
+ properties.put(infoKey, value);
+ }
+ }
+
+ DistributionQueueItem queueItem = new DistributionQueueItem(packageId, size, properties);
+ return queueItem;
+ }
+
+ static DistributionQueueEntry readEntry(Resource queueRoot, Resource resource) {
+
+ if (resource == null) {
+ return null;
+ }
+
+ if (!resource.getPath().startsWith(queueRoot.getPath() + "/")) {
+ return null;
+ }
+
+ if (!resource.isResourceType(RESOURCE_ITEM)) {
+ return null;
+ }
+
+ String queueName = queueRoot.getName();
+ DistributionQueueItem queueItem = deserializeItem(resource.getValueMap());
+ DistributionQueueItemStatus queueItemStatus = new DistributionQueueItemStatus(DistributionQueueItemState.QUEUED, queueName);
+
+ String entryId = getIdFromPath(queueRoot.getPath(), resource.getPath());
+
+ return new DistributionQueueEntry(entryId, queueItem, queueItemStatus);
+ }
+
+ static List<DistributionQueueEntry> getEntries(Resource queueRoot, int skip, int limit) {
+ Iterator<Resource> it = new ResourceIterator(queueRoot, RESOURCE_FOLDER, false, true);
+
+ List<DistributionQueueEntry> entries = new ArrayList<DistributionQueueEntry>();
+
+ int i = 0;
+ while (it.hasNext()) {
+ Resource resource = it.next();
+
+ if (i++ < skip) {
+ continue;
+ }
+
+ DistributionQueueEntry entry = readEntry(queueRoot, resource);
+ entries.add(entry);
+
+ if (entries.size() >= limit) {
+ break;
+ }
+ }
+
+ return entries;
+
+ }
+
+
+ static DistributionQueueEntry getHead(Resource root) {
+ Iterator<DistributionQueueEntry> it = getEntries(root, 0, 1).iterator();
+
+ if (it.hasNext()) {
+ return it.next();
+ }
+
+ return null;
+ }
+
+ public static Resource getRootResource(ResourceResolver resourceResolver, String rootPath) throws PersistenceException {
+ Resource resource = ResourceUtil.getOrCreateResource(resourceResolver, rootPath, RESOURCE_FOLDER, RESOURCE_ROOT, true);
+
+ return resource;
+ }
+
+ public static Resource getResourceById(Resource root, String entryId) {
+ String entryPath = getPathFromId(root.getPath(), entryId);
+ return root.getResourceResolver().getResource(entryPath);
+ }
+
+
+
+ public static Resource createResource(Resource root, DistributionQueueItem queueItem) throws PersistenceException {
+
+ Resource minuteResource = getOrCreateMinuteResource(root);
+
+ String entryPath = getUniqueEntryPath(minuteResource);
+
+ ResourceResolver resourceResolver = root.getResourceResolver();
+
+ Map<String, Object> properties = serializeItem(queueItem);
+
+ properties.put("sling:resourceType", RESOURCE_ITEM);
+ Resource resourceItem = ResourceUtil.getOrCreateResource(resourceResolver, entryPath, properties,
+ RESOURCE_FOLDER, true);
+
+ resourceResolver.commit();
+
+ return resourceItem;
+ }
+
+
+ /**
+ * Creates a minute resource by retrying several times. If it fails even the last time it will throw an exception.
+ */
+ private static Resource getOrCreateMinuteResource(Resource root) throws PersistenceException {
+
+ for (int i=0; i<2; i++) {
+ try {
+ return tryGetOrCreateMinutes(root);
+ } catch (PersistenceException e) {
+ log.warn("creating minute resource failed. retrying {} more times.", 5-i);
+ }
+
+ root.getResourceResolver().revert();
+ root.getResourceResolver().refresh();
+ }
+
+ return tryGetOrCreateMinutes(root);
+ }
+
+ /**
+ * Creates a set of resources for consecutive minutes.
+ * This ensures that consecutive minutes are created by a single thread, and that are created in order.
+ * This might fail due to concurrency issues and needs to be retried a couple of times.
+ */
+ private static Resource tryGetOrCreateMinutes(Resource root) throws PersistenceException {
+
+ ResourceResolver resourceResolver = root.getResourceResolver();
+ Calendar now = Calendar.getInstance();
+
+ String firstMinutePath = getTimePath(now);
+ Resource firstMinuteResource = resourceResolver.getResource(root, firstMinutePath);
+
+ if (firstMinuteResource != null) {
+ return firstMinuteResource;
+ }
+
+ for (int i=0; i < 3; i++) {
+ now.add(Calendar.MINUTE, i);
+ String newMinutePath = getTimePath(now);
+ Resource resource = createResource(root, newMinutePath);
+ }
+
+ resourceResolver.commit();
+
+ firstMinuteResource = resourceResolver.getResource(root, firstMinutePath);
+
+ return firstMinuteResource;
+ }
+
+ /*
+ * Creates a new resource at the specified path
+ * This is different than ResourceUtil.getOrCreateResource as it only creates the resource, it does not retrieve it.
+ * This ensures that consecutive minutes are always created atomically.
+ */
+ private static Resource createResource(Resource root, String relPath) throws PersistenceException {
+ ResourceResolver resourceResolver = root.getResourceResolver();
+
+ String path = root.getPath() + "/" + relPath;
+ final String parentPath = ResourceUtil.getParent(path);
+ final String name = ResourceUtil.getName(path);
+
+ Resource parent = ResourceUtil.getOrCreateResource(resourceResolver, parentPath, RESOURCE_FOLDER,
+ RESOURCE_FOLDER, false);
+
+ Map<String, Object> props = Collections.singletonMap(ResourceResolver.PROPERTY_RESOURCE_TYPE, (Object) RESOURCE_FOLDER);
+
+ return resourceResolver.create(parent, name, props);
+ }
+
+ public static void deleteResource(Resource resource) throws PersistenceException {
+ ResourceResolver resolver = resource.getResourceResolver();
+
+ String path = resource.getPath();
+
+ try {
+ resolver.delete(resource);
+ resolver.commit();
+ } catch (PersistenceException var10) {
+ resolver.revert();
+ resolver.refresh();
+ resource = resolver.getResource(path);
+ if (resource != null) {
+ resolver.delete(resource);
+ resolver.commit();
+ }
+ }
+ }
+
+
+ public static int getResourceCount(Resource root) {
+ ResourceResolver resolver = root.getResourceResolver();
+ Session session = resolver.adaptTo(Session.class);
+
+ StringBuilder buf = new StringBuilder();
+ buf.append("/jcr:root");
+ buf.append(root.getPath());
+ buf.append("//element(*,");
+ buf.append(RESOURCE_ITEM);
+ buf.append(")");
+
+ try {
+ QueryManager qManager = session.getWorkspace().getQueryManager();
+ Query q = qManager.createQuery(buf.toString(), "xpath");
+ final QueryResult res = q.execute();
+
+ NodeIterator it = res.getNodes();
+ return (int) it.getSize();
+ } catch (RepositoryException e) {
+ return -1;
+ }
+ }
+
+
+ private static String getUniqueEntryPath(Resource parent) {
+ final StringBuilder sb = new StringBuilder();
+ sb.append(parent.getPath());
+ sb.append('/');
+ sb.append(UUID.randomUUID().toString().replace("-", ""));
+ sb.append('_');
+ sb.append(itemCounter.getAndIncrement());
+
+ return sb.toString();
+ }
+
+ /**
+ * Transforms current time to path 2018/01/03/23/54
+ * @param now the current time
+ * @return the serialized time
+ */
+ public static String getTimePath(Calendar now) {
+ SimpleDateFormat sdf = new SimpleDateFormat("yyyy/MM/dd/HH/mm");
+
+ return sdf.format(now.getTime());
+ }
+
+
+ /**
+ * Checks if path is safe to delete at this time.
+ * A path is safe to delete if the nowPath does not overlap with it.
+ *
+ * @param nowPath represents a full path of current time (e.g. 2018/01/03/23/54)
+ * @param path the path to be checked (it can be a partial path e.g. 2018/01)
+ * @return true if checked path is in the past
+ */
+ public static boolean isSafeToDelete(String nowPath, String path) {
+
+ // should not happen
+ if (nowPath.length() < path.length()) {
+ return false;
+ }
+
+ nowPath = nowPath.substring(0, path.length());
+
+ return nowPath.compareTo(path) > 0;
+ }
+
+
+
+ private static String getPathFromId(String roothPath, String entryId) {
+ String entryPath = unescapeId(entryId);
+ return roothPath + "/" + entryPath;
+ }
+
+ private static String getIdFromPath(String rootPath, String path) {
+
+ if (path.startsWith(rootPath)) {
+ String entryPath = path.substring(rootPath.length()+1);
+
+ String entryId = escapeId(entryPath);
+
+ return entryId;
+ }
+ throw new IllegalArgumentException("entry path does not start with " + rootPath);
+ }
+
+
+ private static String escapeId(String jobId) {
+ //return id;
+ if (jobId == null) {
+ return null;
+ }
+ return ID_START + jobId.replace("/", "--");
+ }
+
+ public static String unescapeId(String itemId) {
+ if (itemId == null) {
+ return null;
+ }
+ if (!itemId.startsWith(ID_START)) {
+ return null;
+ }
+
+ return itemId.replace(ID_START, "").replace("--", "/");
+ }
+
+}
diff --git a/src/test/java/org/apache/sling/distribution/DistributionBaseIT.java b/src/test/java/org/apache/sling/distribution/DistributionBaseIT.java
new file mode 100644
index 0000000..707d3f8
--- /dev/null
+++ b/src/test/java/org/apache/sling/distribution/DistributionBaseIT.java
@@ -0,0 +1,152 @@
+/*
+ * 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;
+
+import org.apache.sling.api.resource.ResourceResolverFactory;
+import org.apache.sling.distribution.agent.impl.PrivilegeDistributionRequestAuthorizationStrategyFactory;
+import org.apache.sling.distribution.agent.impl.QueueDistributionAgentFactory;
+import org.apache.sling.distribution.agent.spi.DistributionAgent;
+import org.apache.sling.distribution.serialization.impl.vlt.VaultDistributionPackageBuilderFactory;
+import org.apache.sling.testing.paxexam.SlingOptions;
+import org.apache.sling.testing.paxexam.TestSupport;
+import org.ops4j.pax.exam.Configuration;
+import org.ops4j.pax.exam.CoreOptions;
+import org.ops4j.pax.exam.Option;
+import org.osgi.framework.BundleContext;
+import org.osgi.framework.Filter;
+import org.osgi.framework.InvalidSyntaxException;
+import org.osgi.framework.ServiceReference;
+
+import javax.inject.Inject;
+
+import java.io.File;
+
+import static org.apache.sling.testing.paxexam.SlingOptions.logback;
+import static org.apache.sling.testing.paxexam.SlingOptions.slingDistribution;
+import static org.apache.sling.testing.paxexam.SlingOptions.slingQuickstartOakTar;
+import static org.ops4j.pax.exam.CoreOptions.composite;
+import static org.ops4j.pax.exam.CoreOptions.junitBundles;
+import static org.ops4j.pax.exam.cm.ConfigurationAdminOptions.factoryConfiguration;
+import static org.ops4j.pax.exam.cm.ConfigurationAdminOptions.newConfiguration;
+
+public class DistributionBaseIT extends TestSupport {
+
+ protected static final String AGENT_RESOURCE_QUEUE = "agentResourceQueue";
+ protected static final String AGENT_JOB_QUEUE = "agentJobQueue";
+
+
+ @Inject
+ protected ResourceResolverFactory resolverFactory;
+
+ @Inject
+ protected BundleContext context;
+
+
+ @Configuration
+ public Option[] configuration() {
+ return new Option[]{
+ baseConfiguration(),
+ slingQuickstart(),
+ logback(),
+ // build artifact
+ slingDistribution(),
+ // testing
+ testBundle( "bundle.filename"),
+ defaultOsgiConfigs(),
+ junitBundles()
+ };
+ }
+
+ protected Option slingQuickstart() {
+ final String workingDirectory = workingDirectory(); // from TestSupport
+ final int httpPort = findFreePort(); // from TestSupport
+ return composite(
+ slingQuickstartOakTar(workingDirectory, httpPort) // from SlingOptions
+ );
+ }
+
+ public static Option defaultOsgiConfigs() {
+ return composite(
+ newConfiguration("org.apache.sling.jcr.resource.internal.JcrSystemUserValidator")
+ .put("allow.only.system.user", false).asOption(),
+
+ newConfiguration("org.apache.sling.jcr.base.internal.LoginAdminWhitelist")
+ .put("whitelist.bypass", "true").asOption(),
+
+ // For production the users would be: replication-service,content-writer-service
+ factoryConfiguration("org.apache.sling.serviceusermapping.impl.ServiceUserMapperImpl.amended")
+ .put("user.mapping", new String[]{"org.apache.sling.distribution.core:testService=admin"})
+ .asOption(),
+
+ factoryConfiguration(PrivilegeDistributionRequestAuthorizationStrategyFactory.class.getName())
+ .put("name", "default")
+ .put("jcrPrivilege", "jcr:read")
+ .asOption(),
+
+ factoryConfiguration(VaultDistributionPackageBuilderFactory.class.getName())
+ .put("name", "default")
+ .put("type", "jcrvlt")
+ .asOption(),
+
+ factoryConfiguration(QueueDistributionAgentFactory.class.getName())
+ .put("name", AGENT_RESOURCE_QUEUE)
+ .put("serviceName", "testService")
+ .put("enabled", true)
+ .put("queueProviderFactory.target", "(name=resourceQueue)")
+ .asOption(),
+
+ factoryConfiguration(QueueDistributionAgentFactory.class.getName())
+ .put("name", AGENT_JOB_QUEUE)
+ .put("serviceName", "testService")
+ .put("enabled", true)
+ .put("queueProviderFactory.target", "(name=jobQueue)")
+ .asOption()
+ );
+ }
+
+
+ protected DistributionAgent getAgent(String agentName) {
+
+ for (int i=0; i< 10; i++) {
+ try {
+ Filter filter = context.createFilter("(name="+agentName+")");
+
+ ServiceReference[] srs = this.context.getServiceReferences(DistributionAgent.class.getName(), filter.toString());
+ if (srs == null || srs.length == 0) {
+ Thread.sleep(1000);
+ continue;
+ }
+ ServiceReference sr = srs[0];
+
+ Object service = context.getService(sr);
+
+ return (DistributionAgent) service;
+ } catch (InvalidSyntaxException e) {
+
+ } catch (InterruptedException e) {
+ }
+
+
+ }
+
+ return null;
+ }
+
+}
diff --git a/src/test/java/org/apache/sling/distribution/queue/impl/resource/ResourceIteratorTest.java b/src/test/java/org/apache/sling/distribution/queue/impl/resource/ResourceIteratorTest.java
new file mode 100644
index 0000000..285ca3a
--- /dev/null
+++ b/src/test/java/org/apache/sling/distribution/queue/impl/resource/ResourceIteratorTest.java
@@ -0,0 +1,166 @@
+/*
+ * 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.resource;
+
+import org.apache.sling.api.resource.PersistenceException;
+import org.apache.sling.api.resource.Resource;
+import org.apache.sling.testing.mock.sling.junit.SlingContext;
+import org.apache.sling.testing.resourceresolver.MockHelper;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+
+public class ResourceIteratorTest {
+
+ static final String FOLDER_TYPE = ResourceQueueUtils.RESOURCE_FOLDER;
+ static final String ITEM_TYPE = "nt:unstructured";
+
+ @Rule
+ public final SlingContext context = new SlingContext();
+
+ @Before
+ public void testSetup() {
+ MockHelper helper = MockHelper.create(context.resourceResolver());
+
+ helper.resource("/root").p("prop", "value")
+ .resource("2018").p("sling:resourceType", FOLDER_TYPE)
+ .resource("10").p("sling:resourceType", FOLDER_TYPE)
+ .resource("15").p("sling:resourceType", FOLDER_TYPE)
+ .resource("11").p("sling:resourceType", FOLDER_TYPE)
+ .resource("35").p("sling:resourceType", FOLDER_TYPE)
+ .resource("item1").p("sling:resourceType", ITEM_TYPE)
+ .resource("../item2").p("sling:resourceType", ITEM_TYPE)
+ .resource("../../36").p("sling:resourceType", FOLDER_TYPE)
+ .resource("item3").p("sling:resourceType", ITEM_TYPE);
+
+ helper.resource("/root2").p("prop", "value")
+ .resource("2018").p("sling:resourceType", FOLDER_TYPE)
+ .resource("10").p("sling:resourceType", FOLDER_TYPE)
+ .resource("15").p("sling:resourceType", FOLDER_TYPE)
+ .resource("11").p("sling:resourceType", FOLDER_TYPE)
+ .resource("35").p("sling:resourceType", FOLDER_TYPE)
+ .resource("../36").p("sling:resourceType", FOLDER_TYPE)
+ .resource("item3").p("sling:resourceType", ITEM_TYPE);
+
+ try {
+ helper.commit();
+ } catch (PersistenceException e) {
+
+ }
+
+ }
+
+ @Test
+ public void testEmptyRoot() {
+ Resource root = context.resourceResolver().getResource("/root");
+
+ ResourceIterator it = new ResourceIterator(root, ResourceQueueUtils.RESOURCE_FOLDER, false, false);
+
+
+ test(it, new String[0]);
+ }
+
+ @Test
+ public void testFullPath() {
+ Resource root = context.resourceResolver().getResource("/root");
+
+ ResourceIterator it = new ResourceIterator(root, ResourceQueueUtils.RESOURCE_FOLDER, true, true);
+
+ test(it, new String[] {
+ "/root/2018/10/15/11/35/item1",
+ "/root/2018/10/15/11/35/item2",
+ "/root/2018/10/15/11/35",
+ "/root/2018/10/15/11/36/item3",
+ "/root/2018/10/15/11/36",
+ "/root/2018/10/15/11",
+ "/root/2018/10/15",
+ "/root/2018/10",
+ "/root/2018"
+ });
+ }
+
+ @Test
+ public void testFullPath2() {
+ Resource root = context.resourceResolver().getResource("/root2");
+
+ ResourceIterator it = new ResourceIterator(root, ResourceQueueUtils.RESOURCE_FOLDER, true, true);
+
+ test(it, new String[] {
+ "/root2/2018/10/15/11/35",
+ "/root2/2018/10/15/11/36/item3",
+ "/root2/2018/10/15/11/36",
+ "/root2/2018/10/15/11",
+ "/root2/2018/10/15",
+ "/root2/2018/10",
+ "/root2/2018"
+ });
+ }
+
+ @Test
+ public void testFullPathNoFolder() {
+ Resource root = context.resourceResolver().getResource("/root");
+
+ ResourceIterator it = new ResourceIterator(root, ResourceQueueUtils.RESOURCE_FOLDER, false, true);
+
+
+ test(it, new String[] {
+ "/root/2018/10/15/11/35/item1",
+ "/root/2018/10/15/11/35/item2",
+ "/root/2018/10/15/11/36/item3",
+ });
+ }
+
+ @Test
+ public void testFullPathNoLeafs() {
+ Resource root = context.resourceResolver().getResource("/root");
+
+ ResourceIterator it = new ResourceIterator(root, ResourceQueueUtils.RESOURCE_FOLDER, true, false);
+
+ test(it, new String[] {
+ "/root/2018/10/15/11/35",
+ "/root/2018/10/15/11/36",
+ "/root/2018/10/15/11",
+ "/root/2018/10/15",
+ "/root/2018/10",
+ "/root/2018"
+ });
+ }
+
+ public void test(ResourceIterator it, String[] paths) {
+ List<String> expected = Arrays.asList(paths);
+
+ assertEquals(expected, toPaths(it));
+ }
+
+
+ List<String> toPaths(ResourceIterator it) {
+ List<String> paths = new ArrayList<String>();
+ while (it.hasNext()) {
+ paths.add(it.next().getPath());
+ }
+ return paths;
+ }
+}
diff --git a/src/test/java/org/apache/sling/distribution/queue/impl/resource/ResourceQueueIT.java b/src/test/java/org/apache/sling/distribution/queue/impl/resource/ResourceQueueIT.java
new file mode 100644
index 0000000..b599aa5
--- /dev/null
+++ b/src/test/java/org/apache/sling/distribution/queue/impl/resource/ResourceQueueIT.java
@@ -0,0 +1,237 @@
+/*
+ * 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.resource;
+
+import org.apache.sling.api.resource.LoginException;
+import org.apache.sling.api.resource.ResourceResolver;
+import org.apache.sling.distribution.DistributionRequestType;
+import org.apache.sling.distribution.DistributionResponse;
+import org.apache.sling.distribution.Distributor;
+import org.apache.sling.distribution.SimpleDistributionRequest;
+import org.apache.sling.distribution.agent.spi.DistributionAgent;
+import org.apache.sling.distribution.common.DistributionException;
+import org.apache.sling.distribution.queue.DistributionQueueEntry;
+import org.apache.sling.distribution.queue.DistributionQueueItem;
+import org.apache.sling.distribution.DistributionBaseIT;
+import org.apache.sling.distribution.queue.spi.DistributionQueue;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.ops4j.pax.exam.junit.PaxExam;
+
+import javax.inject.Inject;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+@RunWith(PaxExam.class)
+public class ResourceQueueIT extends DistributionBaseIT {
+
+ static final int ITERATIONS = 1000;
+ static final String QUEUE_NAME = "default";
+
+ @Inject
+ protected Distributor distributor;
+
+
+ @Test
+ public void testExecute() throws LoginException, DistributionException {
+
+ DistributionAgent agent = getAgent(AGENT_RESOURCE_QUEUE);
+
+ assertNotNull(agent);
+
+ ResourceResolver resourceResolver = resolverFactory.getAdministrativeResourceResolver(null);
+
+ DistributionResponse response = agent.execute(resourceResolver,
+ new SimpleDistributionRequest(DistributionRequestType.ADD, "/content" ));
+ assertTrue(response.isSuccessful());
+
+ DistributionQueue queue = agent.getQueue(QUEUE_NAME);
+
+ assertEquals(1, queue.getStatus().getItemsCount());
+
+ clear(queue);
+
+ }
+
+
+ @Test
+ public void testFifo() {
+ DistributionAgent agent = getAgent(AGENT_RESOURCE_QUEUE);
+
+ DistributionQueue queue = agent.getQueue(QUEUE_NAME);
+
+ queue.add(new DistributionQueueItem("packageId", 10, new HashMap<String, Object>()));
+
+ assertEquals(1, queue.getStatus().getItemsCount());
+
+ clear(queue);
+ }
+
+ @Test
+ public void testConcurrentFifo () throws InterruptedException {
+ final DistributionAgent agent = getAgent(AGENT_RESOURCE_QUEUE);
+ final DistributionQueue queue = agent.getQueue(QUEUE_NAME);
+
+ Producer p1 = new Producer(queue, "p1");
+ Producer p2 = new Producer(queue, "p2");
+ Producer p3 = new Producer(queue, "p3");
+ Producer p4 = new Producer(queue, "p4");
+
+ Map<String, Producer> producerMap = new HashMap<String, Producer>();
+ producerMap.put("p1", p1);
+ producerMap.put("p2", p2);
+ producerMap.put("p3", p3);
+ producerMap.put("p4", p4);
+
+ Consumer c1 = new Consumer(queue, "c1", new String[] { "p1", "p2" });
+ Consumer c2 = new Consumer(queue, "c2", new String[] { "p3", "p4" });
+
+ Map<String, Consumer> consumerMap = new HashMap<String, Consumer>();
+ consumerMap.put("c1", c1);
+ consumerMap.put("c2", c2);
+
+ List<Thread> threads = new ArrayList<Thread>();
+ for (Producer p : producerMap.values()) {
+ threads.add(new Thread(p));
+ }
+ for (Consumer c : consumerMap.values()) {
+ threads.add(new Thread(c));
+ }
+
+ for (Thread t : threads) {
+ t.start();
+ }
+
+ for (Thread t : threads) {
+ t.join(600 * 1000);
+ }
+
+ for (Producer p : producerMap.values()) {
+ assertEquals(ITERATIONS, p.created.size());
+ }
+
+ for (Consumer c : consumerMap.values()) {
+ for (String source : c.sources) {
+ Producer p = producerMap.get(source);
+ assertEquals(p.created, c.removedBySource.get(source));
+ }
+ }
+
+ assertEquals(0, queue.getStatus().getItemsCount());
+
+ clear(queue);
+ }
+
+ void clear(DistributionQueue queue) {
+ for (DistributionQueueEntry entry : queue.getItems(0, -1)) {
+ queue.remove(entry.getId());
+ }
+ }
+
+ class Producer implements Runnable {
+ private final DistributionQueue queue;
+ private final String name;
+ final List<String> created = Collections.synchronizedList(new ArrayList<String>());
+
+ Producer(DistributionQueue queue, String name) {
+ this.queue = queue;
+ this.name = name;
+ }
+
+
+ public void run() {
+ for(int i=0; i<ITERATIONS; i++) {
+
+ if (i % 1000 == 0) {
+ System.out.println("Producer " + name + " " + i);
+ }
+
+ final Map<String, Object> props = new HashMap<String, Object>();
+ final String packageId = UUID.randomUUID().toString();
+ final long size = 10;
+
+ props.put("auuid", UUID.randomUUID().toString());
+ props.put("no", i);
+ props.put("source", name);
+
+ final DistributionQueueItem item = new DistributionQueueItem(packageId, size, props);
+
+ final DistributionQueueEntry entry = queue.add(item);
+
+ created.add(entry.getId());
+ }
+ }
+ }
+
+ class Consumer implements Runnable {
+ private final DistributionQueue queue;
+ private final String name;
+ private final String[] sources;
+
+ final List<String> removed = new ArrayList<String>();
+
+ final Map<String, List<String>> removedBySource = new HashMap<String, List<String>>();
+
+
+ Consumer(DistributionQueue queue, String name, String[] sources) {
+
+ this.queue = queue;
+ this.name = name;
+ this.sources = sources;
+ for (String source: sources) {
+ removedBySource.put(source, new ArrayList<String>());
+ }
+ }
+
+ public void run() {
+ while (removed.size() < sources.length * ITERATIONS) {
+
+ final DistributionQueueEntry entry = queue.getHead();
+
+ if (entry != null) {
+
+ String source = (String) entry.getItem().get("source");
+
+ if (removedBySource.containsKey(source)) {
+ if (removed.size() % 1000 == 0) {
+ System.out.println("Consumer " + name + " " + removed.size());
+ }
+
+ final DistributionQueueEntry removedEntry = queue.remove(entry.getId());
+
+ assertEquals(removedEntry.getId(), entry.getId());
+
+ removed.add(entry.getId());
+ removedBySource.get(source).add(entry.getId());
+ }
+ }
+ }
+ }
+ }
+}
diff --git a/src/test/java/org/apache/sling/distribution/queue/impl/resource/ResourceQueueUtilsTest.java b/src/test/java/org/apache/sling/distribution/queue/impl/resource/ResourceQueueUtilsTest.java
new file mode 100644
index 0000000..af04a81
--- /dev/null
+++ b/src/test/java/org/apache/sling/distribution/queue/impl/resource/ResourceQueueUtilsTest.java
@@ -0,0 +1,100 @@
+/*
+ * 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.resource;
+
+import org.junit.Test;
+
+import java.util.GregorianCalendar;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class ResourceQueueUtilsTest {
+
+ @Test
+ public void testTimePath() throws Exception {
+
+
+ assertEquals("2018/01/01/00/00",
+ ResourceQueueUtils.getTimePath(new GregorianCalendar(2018, 0, 1, 0, 0)));
+
+
+ assertEquals("2018/12/01/00/00",
+ ResourceQueueUtils.getTimePath(new GregorianCalendar(2018, 11, 1, 0, 0)));
+
+ assertEquals("2018/12/31/23/59",
+ ResourceQueueUtils.getTimePath(new GregorianCalendar(2018, 11, 31, 23, 59)));
+ }
+
+
+ @Test
+ public void testIsSafeToDelete() throws Exception {
+
+ assertTrue(ResourceQueueUtils.isSafeToDelete("2018/02/15/03/09", "2017"));
+ assertFalse(ResourceQueueUtils.isSafeToDelete("2018/02/15/03/09", "2018"));
+ assertFalse(ResourceQueueUtils.isSafeToDelete("2018/02/15/03/09", "2019"));
+
+ assertTrue(ResourceQueueUtils.isSafeToDelete("2018/02/15/03/09", "2018/01"));
+ assertFalse(ResourceQueueUtils.isSafeToDelete("2018/02/15/03/09", "2018/02"));
+ assertFalse(ResourceQueueUtils.isSafeToDelete("2018/02/15/03/09", "2018/03"));
+
+ assertTrue(ResourceQueueUtils.isSafeToDelete("2018/02/15/03/09", "2018/02/14"));
+ assertFalse(ResourceQueueUtils.isSafeToDelete("2018/02/15/03/09", "2018/02/15"));
+ assertFalse(ResourceQueueUtils.isSafeToDelete("2018/02/15/03/09", "2018/02/16"));
+
+ assertTrue(ResourceQueueUtils.isSafeToDelete("2018/02/15/03/09", "2018/02/15/02"));
+ assertFalse(ResourceQueueUtils.isSafeToDelete("2018/02/15/03/09", "2018/02/15/03"));
+ assertFalse(ResourceQueueUtils.isSafeToDelete("2018/02/15/03/09", "2018/02/15/04"));
+
+ assertTrue(ResourceQueueUtils.isSafeToDelete("2018/02/15/03/09", "2018/02/15/03/08"));
+ assertFalse(ResourceQueueUtils.isSafeToDelete("2018/02/15/03/09", "2018/02/15/03/09"));
+ assertFalse(ResourceQueueUtils.isSafeToDelete("2018/02/15/03/09", "2018/02/15/03/10"));
+
+ assertFalse(ResourceQueueUtils.isSafeToDelete("2018/02/15/03/09", "2018/02/15/03/10"));
+ assertFalse(ResourceQueueUtils.isSafeToDelete("2018/02/15/03/09", "2018/03/16/04/09"));
+
+
+
+
+ // new year
+ assertTrue(ResourceQueueUtils.isSafeToDelete("2018/12/31/23/59", "2017"));
+ assertFalse(ResourceQueueUtils.isSafeToDelete("2018/12/31/23/59", "2018"));
+ assertFalse(ResourceQueueUtils.isSafeToDelete("2018/12/31/23/59", "2019"));
+
+
+ assertTrue(ResourceQueueUtils.isSafeToDelete("2018/12/31/23/59", "2018/11"));
+ assertFalse(ResourceQueueUtils.isSafeToDelete("2018/12/31/23/59", "2018/12"));
+ assertFalse(ResourceQueueUtils.isSafeToDelete("2018/12/31/23/59", "2019/01"));
+
+
+ assertTrue(ResourceQueueUtils.isSafeToDelete("2018/12/31/23/59", "2018/12/30"));
+ assertFalse(ResourceQueueUtils.isSafeToDelete("2018/12/31/23/59", "2018/12/31"));
+ assertFalse(ResourceQueueUtils.isSafeToDelete("2018/12/31/23/59", "2019/01/00"));
+
+ assertTrue(ResourceQueueUtils.isSafeToDelete("2018/12/31/23/59", "2018/12/31/22"));
+ assertFalse(ResourceQueueUtils.isSafeToDelete("2018/12/31/23/59", "2018/12/31/23"));
+ assertFalse(ResourceQueueUtils.isSafeToDelete("2018/12/31/23/59", "2019/01/01/00"));
+
+ assertTrue(ResourceQueueUtils.isSafeToDelete("2018/12/31/23/59", "2018/12/31/23/58"));
+ assertFalse(ResourceQueueUtils.isSafeToDelete("2018/12/31/23/59", "2018/12/31/23/59"));
+ assertFalse(ResourceQueueUtils.isSafeToDelete("2018/12/31/23/59", "2019/01/01/00/00"));
+
+ }
+}
