You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@sling.apache.org by cs...@apache.org on 2020/06/23 11:18:32 UTC
[sling-org-apache-sling-distribution-journal] branch master updated
(26204b4 -> 91a8442)
This is an automated email from the ASF dual-hosted git repository.
cschneider pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/sling-org-apache-sling-distribution-journal.git.
from 26204b4 SLING-9259 - Create separate class for imported event
new 52ee4fd SLING-9259 - Queue package messages instead of queue items
new 91a8442 SLING-9259 - Move PackageRegtries and LocalStore to shared
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../journal/impl/queue/impl/PubQueueCache.java | 2 +-
.../impl/queue/impl/PubQueueCacheService.java | 3 +-
.../journal/impl/queue/impl/SubQueue.java | 1 +
.../impl/{subscriber => shared}/LocalStore.java | 2 +-
.../{queue/impl => shared}/PackageRetries.java | 2 +-
.../journal/impl/subscriber/BookKeeper.java | 3 +-
.../impl/subscriber/DistributionSubscriber.java | 46 +++++++++-------------
.../journal/impl/queue/impl/PubQueueCacheTest.java | 2 +-
.../impl/queue/impl/PubQueueProviderTest.java | 2 +-
.../journal/impl/queue/impl/SubQueueTest.java | 2 +
.../{subscriber => shared}/LocalStoreTest.java | 3 +-
.../{queue/impl => shared}/PackageRetriesTest.java | 4 +-
12 files changed, 35 insertions(+), 37 deletions(-)
rename src/main/java/org/apache/sling/distribution/journal/impl/{subscriber => shared}/LocalStore.java (98%)
rename src/main/java/org/apache/sling/distribution/journal/impl/{queue/impl => shared}/PackageRetries.java (96%)
rename src/test/java/org/apache/sling/distribution/journal/impl/{subscriber => shared}/LocalStoreTest.java (96%)
rename src/test/java/org/apache/sling/distribution/journal/impl/{queue/impl => shared}/PackageRetriesTest.java (91%)
[sling-org-apache-sling-distribution-journal] 02/02: SLING-9259 -
Move PackageRegtries and LocalStore to shared
Posted by cs...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
cschneider pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/sling-org-apache-sling-distribution-journal.git
commit 91a8442985db0334636bbcfe9671de40cc098ef9
Author: Christian Schneider <cs...@adobe.com>
AuthorDate: Tue Jun 23 13:13:21 2020 +0200
SLING-9259 - Move PackageRegtries and LocalStore to shared
---
.../sling/distribution/journal/impl/queue/impl/PubQueueCache.java | 2 +-
.../distribution/journal/impl/queue/impl/PubQueueCacheService.java | 3 +--
.../apache/sling/distribution/journal/impl/queue/impl/SubQueue.java | 1 +
.../distribution/journal/impl/{subscriber => shared}/LocalStore.java | 2 +-
.../journal/impl/{queue/impl => shared}/PackageRetries.java | 2 +-
.../apache/sling/distribution/journal/impl/subscriber/BookKeeper.java | 3 ++-
.../sling/distribution/journal/impl/queue/impl/PubQueueCacheTest.java | 2 +-
.../distribution/journal/impl/queue/impl/PubQueueProviderTest.java | 2 +-
.../sling/distribution/journal/impl/queue/impl/SubQueueTest.java | 2 ++
.../journal/impl/{subscriber => shared}/LocalStoreTest.java | 3 ++-
.../journal/impl/{queue/impl => shared}/PackageRetriesTest.java | 4 +++-
11 files changed, 16 insertions(+), 10 deletions(-)
diff --git a/src/main/java/org/apache/sling/distribution/journal/impl/queue/impl/PubQueueCache.java b/src/main/java/org/apache/sling/distribution/journal/impl/queue/impl/PubQueueCache.java
index 6cf0be7..fb83127 100644
--- a/src/main/java/org/apache/sling/distribution/journal/impl/queue/impl/PubQueueCache.java
+++ b/src/main/java/org/apache/sling/distribution/journal/impl/queue/impl/PubQueueCache.java
@@ -41,8 +41,8 @@ import org.apache.sling.api.resource.PersistenceException;
import org.apache.sling.distribution.journal.impl.event.DistributionEvent;
import org.apache.sling.distribution.journal.impl.shared.DistributionMetricsService;
import org.apache.sling.distribution.journal.impl.shared.JMXRegistration;
+import org.apache.sling.distribution.journal.impl.shared.LocalStore;
import org.apache.commons.io.IOUtils;
-import org.apache.sling.distribution.journal.impl.subscriber.LocalStore;
import org.apache.sling.distribution.journal.messages.PackageMessage;
import org.apache.sling.distribution.queue.DistributionQueueItem;
import org.osgi.service.event.Event;
diff --git a/src/main/java/org/apache/sling/distribution/journal/impl/queue/impl/PubQueueCacheService.java b/src/main/java/org/apache/sling/distribution/journal/impl/queue/impl/PubQueueCacheService.java
index 2eddc6f..5e98ab4 100644
--- a/src/main/java/org/apache/sling/distribution/journal/impl/queue/impl/PubQueueCacheService.java
+++ b/src/main/java/org/apache/sling/distribution/journal/impl/queue/impl/PubQueueCacheService.java
@@ -23,13 +23,12 @@ import javax.annotation.ParametersAreNonnullByDefault;
import org.apache.sling.api.resource.ResourceResolverFactory;
import org.apache.sling.distribution.journal.impl.shared.DistributionMetricsService;
+import org.apache.sling.distribution.journal.impl.shared.LocalStore;
import org.apache.sling.distribution.journal.impl.shared.PublisherConfigurationAvailable;
import org.apache.sling.distribution.journal.impl.shared.Topics;
import org.apache.sling.distribution.journal.impl.queue.OffsetQueue;
import org.apache.sling.distribution.journal.MessagingProvider;
import org.apache.sling.distribution.journal.JournalAvailable;
-
-import org.apache.sling.distribution.journal.impl.subscriber.LocalStore;
import org.apache.sling.distribution.queue.DistributionQueueItem;
import org.apache.sling.settings.SlingSettingsService;
import org.osgi.service.component.annotations.Activate;
diff --git a/src/main/java/org/apache/sling/distribution/journal/impl/queue/impl/SubQueue.java b/src/main/java/org/apache/sling/distribution/journal/impl/queue/impl/SubQueue.java
index 5d34e6d..4bfa2bf 100644
--- a/src/main/java/org/apache/sling/distribution/journal/impl/queue/impl/SubQueue.java
+++ b/src/main/java/org/apache/sling/distribution/journal/impl/queue/impl/SubQueue.java
@@ -29,6 +29,7 @@ import javax.annotation.Nullable;
import javax.annotation.ParametersAreNonnullByDefault;
import org.apache.sling.distribution.queue.spi.DistributionQueue;
+import org.apache.sling.distribution.journal.impl.shared.PackageRetries;
import org.apache.sling.distribution.queue.DistributionQueueEntry;
import org.apache.sling.distribution.queue.DistributionQueueItem;
import org.apache.sling.distribution.queue.DistributionQueueItemState;
diff --git a/src/main/java/org/apache/sling/distribution/journal/impl/subscriber/LocalStore.java b/src/main/java/org/apache/sling/distribution/journal/impl/shared/LocalStore.java
similarity index 98%
rename from src/main/java/org/apache/sling/distribution/journal/impl/subscriber/LocalStore.java
rename to src/main/java/org/apache/sling/distribution/journal/impl/shared/LocalStore.java
index 878c831..b8a3a73 100644
--- a/src/main/java/org/apache/sling/distribution/journal/impl/subscriber/LocalStore.java
+++ b/src/main/java/org/apache/sling/distribution/journal/impl/shared/LocalStore.java
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.sling.distribution.journal.impl.subscriber;
+package org.apache.sling.distribution.journal.impl.shared;
import org.apache.sling.api.resource.LoginException;
import org.apache.sling.api.resource.ModifiableValueMap;
diff --git a/src/main/java/org/apache/sling/distribution/journal/impl/queue/impl/PackageRetries.java b/src/main/java/org/apache/sling/distribution/journal/impl/shared/PackageRetries.java
similarity index 96%
rename from src/main/java/org/apache/sling/distribution/journal/impl/queue/impl/PackageRetries.java
rename to src/main/java/org/apache/sling/distribution/journal/impl/shared/PackageRetries.java
index 23f9463..a124eca 100644
--- a/src/main/java/org/apache/sling/distribution/journal/impl/queue/impl/PackageRetries.java
+++ b/src/main/java/org/apache/sling/distribution/journal/impl/shared/PackageRetries.java
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.sling.distribution.journal.impl.queue.impl;
+package org.apache.sling.distribution.journal.impl.shared;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
diff --git a/src/main/java/org/apache/sling/distribution/journal/impl/subscriber/BookKeeper.java b/src/main/java/org/apache/sling/distribution/journal/impl/subscriber/BookKeeper.java
index 4ec2fc0..0541b41 100644
--- a/src/main/java/org/apache/sling/distribution/journal/impl/subscriber/BookKeeper.java
+++ b/src/main/java/org/apache/sling/distribution/journal/impl/subscriber/BookKeeper.java
@@ -38,8 +38,9 @@ import org.apache.sling.api.resource.ResourceResolverFactory;
import org.apache.sling.api.resource.ValueMap;
import org.apache.sling.commons.metrics.Timer;
import org.apache.sling.distribution.common.DistributionException;
-import org.apache.sling.distribution.journal.impl.queue.impl.PackageRetries;
import org.apache.sling.distribution.journal.impl.shared.DistributionMetricsService;
+import org.apache.sling.distribution.journal.impl.shared.LocalStore;
+import org.apache.sling.distribution.journal.impl.shared.PackageRetries;
import org.apache.sling.distribution.journal.impl.shared.DistributionMetricsService.GaugeService;
import org.apache.sling.distribution.journal.messages.PackageMessage;
import org.apache.sling.distribution.journal.messages.PackageStatusMessage;
diff --git a/src/test/java/org/apache/sling/distribution/journal/impl/queue/impl/PubQueueCacheTest.java b/src/test/java/org/apache/sling/distribution/journal/impl/queue/impl/PubQueueCacheTest.java
index e7f2444..957062e 100644
--- a/src/test/java/org/apache/sling/distribution/journal/impl/queue/impl/PubQueueCacheTest.java
+++ b/src/test/java/org/apache/sling/distribution/journal/impl/queue/impl/PubQueueCacheTest.java
@@ -47,8 +47,8 @@ import org.apache.sling.distribution.journal.MessagingProvider;
import org.apache.sling.distribution.journal.Reset;
import org.apache.sling.distribution.journal.impl.queue.OffsetQueue;
import org.apache.sling.distribution.journal.impl.shared.DistributionMetricsService;
+import org.apache.sling.distribution.journal.impl.shared.LocalStore;
import org.apache.sling.distribution.journal.impl.shared.TestMessageInfo;
-import org.apache.sling.distribution.journal.impl.subscriber.LocalStore;
import org.apache.sling.distribution.journal.messages.PackageMessage;
import org.apache.sling.distribution.journal.messages.PackageMessage.ReqType;
import org.apache.sling.distribution.queue.DistributionQueueItem;
diff --git a/src/test/java/org/apache/sling/distribution/journal/impl/queue/impl/PubQueueProviderTest.java b/src/test/java/org/apache/sling/distribution/journal/impl/queue/impl/PubQueueProviderTest.java
index 2435c3a..70885e5 100644
--- a/src/test/java/org/apache/sling/distribution/journal/impl/queue/impl/PubQueueProviderTest.java
+++ b/src/test/java/org/apache/sling/distribution/journal/impl/queue/impl/PubQueueProviderTest.java
@@ -41,13 +41,13 @@ import javax.management.ReflectionException;
import org.apache.sling.api.resource.PersistenceException;
import org.apache.sling.api.resource.ResourceResolverFactory;
+import org.apache.sling.distribution.journal.impl.shared.LocalStore;
import org.apache.sling.distribution.journal.impl.shared.Topics;
import org.apache.sling.distribution.journal.messages.PackageMessage;
import org.apache.sling.distribution.journal.messages.PackageMessage.ReqType;
import org.apache.sling.distribution.journal.messages.PackageStatusMessage;
import org.apache.sling.distribution.journal.messages.PackageStatusMessage.Status;
import org.apache.sling.distribution.journal.MessageSender;
-import org.apache.sling.distribution.journal.impl.subscriber.LocalStore;
import org.apache.sling.distribution.queue.DistributionQueueEntry;
import org.apache.sling.distribution.queue.DistributionQueueItem;
import org.apache.sling.distribution.queue.spi.DistributionQueue;
diff --git a/src/test/java/org/apache/sling/distribution/journal/impl/queue/impl/SubQueueTest.java b/src/test/java/org/apache/sling/distribution/journal/impl/queue/impl/SubQueueTest.java
index d2784e9..3b73074 100644
--- a/src/test/java/org/apache/sling/distribution/journal/impl/queue/impl/SubQueueTest.java
+++ b/src/test/java/org/apache/sling/distribution/journal/impl/queue/impl/SubQueueTest.java
@@ -26,6 +26,8 @@ import org.junit.Assert;
import org.junit.Test;
import org.apache.sling.distribution.journal.impl.queue.QueueItemFactory;
+import org.apache.sling.distribution.journal.impl.shared.PackageRetries;
+
import com.google.common.collect.Lists;
public class SubQueueTest {
diff --git a/src/test/java/org/apache/sling/distribution/journal/impl/subscriber/LocalStoreTest.java b/src/test/java/org/apache/sling/distribution/journal/impl/shared/LocalStoreTest.java
similarity index 96%
rename from src/test/java/org/apache/sling/distribution/journal/impl/subscriber/LocalStoreTest.java
rename to src/test/java/org/apache/sling/distribution/journal/impl/shared/LocalStoreTest.java
index c989b4a..f95c470 100644
--- a/src/test/java/org/apache/sling/distribution/journal/impl/subscriber/LocalStoreTest.java
+++ b/src/test/java/org/apache/sling/distribution/journal/impl/shared/LocalStoreTest.java
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.sling.distribution.journal.impl.subscriber;
+package org.apache.sling.distribution.journal.impl.shared;
import java.util.HashMap;
import java.util.Map;
@@ -29,6 +29,7 @@ import static org.junit.Assert.assertEquals;
import org.apache.sling.api.resource.LoginException;
import org.apache.sling.api.resource.PersistenceException;
import org.apache.sling.api.resource.ResourceResolver;
+import org.apache.sling.distribution.journal.impl.shared.LocalStore;
import org.apache.sling.testing.resourceresolver.MockResourceResolverFactory;
import org.junit.Test;
diff --git a/src/test/java/org/apache/sling/distribution/journal/impl/queue/impl/PackageRetriesTest.java b/src/test/java/org/apache/sling/distribution/journal/impl/shared/PackageRetriesTest.java
similarity index 91%
rename from src/test/java/org/apache/sling/distribution/journal/impl/queue/impl/PackageRetriesTest.java
rename to src/test/java/org/apache/sling/distribution/journal/impl/shared/PackageRetriesTest.java
index 6c33412..8e7adb9 100644
--- a/src/test/java/org/apache/sling/distribution/journal/impl/queue/impl/PackageRetriesTest.java
+++ b/src/test/java/org/apache/sling/distribution/journal/impl/shared/PackageRetriesTest.java
@@ -16,12 +16,14 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.sling.distribution.journal.impl.queue.impl;
+package org.apache.sling.distribution.journal.impl.shared;
import org.junit.Test;
import static org.junit.Assert.assertEquals;
+import org.apache.sling.distribution.journal.impl.shared.PackageRetries;
+
public class PackageRetriesTest {
@Test
[sling-org-apache-sling-distribution-journal] 01/02: SLING-9259 -
Queue package messages instead of queue items
Posted by cs...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
cschneider pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/sling-org-apache-sling-distribution-journal.git
commit 52ee4fd5eade48b4fd22d841588f3ad23b3e4937
Author: Christian Schneider <cs...@adobe.com>
AuthorDate: Tue Jun 23 13:10:41 2020 +0200
SLING-9259 - Queue package messages instead of queue items
---
.../impl/subscriber/DistributionSubscriber.java | 46 +++++++++-------------
1 file changed, 19 insertions(+), 27 deletions(-)
diff --git a/src/main/java/org/apache/sling/distribution/journal/impl/subscriber/DistributionSubscriber.java b/src/main/java/org/apache/sling/distribution/journal/impl/subscriber/DistributionSubscriber.java
index 94ed9ff..9a5625b 100644
--- a/src/main/java/org/apache/sling/distribution/journal/impl/subscriber/DistributionSubscriber.java
+++ b/src/main/java/org/apache/sling/distribution/journal/impl/subscriber/DistributionSubscriber.java
@@ -19,13 +19,9 @@
package org.apache.sling.distribution.journal.impl.subscriber;
import static java.lang.String.format;
-import static java.util.Collections.emptyMap;
import static java.util.Objects.requireNonNull;
import static java.util.stream.Collectors.toSet;
import static org.apache.sling.distribution.journal.RunnableUtil.startBackgroundThread;
-import static org.apache.sling.distribution.journal.impl.queue.QueueItemFactory.PACKAGE_MSG;
-import static org.apache.sling.distribution.journal.impl.queue.QueueItemFactory.RECORD_OFFSET;
-import static org.apache.sling.distribution.journal.impl.queue.QueueItemFactory.RECORD_TIMESTAMP;
import java.io.Closeable;
import java.io.IOException;
@@ -54,19 +50,18 @@ import org.apache.sling.commons.osgi.PropertiesUtil;
import org.apache.sling.distribution.agent.DistributionAgentState;
import org.apache.sling.distribution.agent.spi.DistributionAgent;
import org.apache.sling.distribution.common.DistributionException;
+import org.apache.sling.distribution.journal.FullMessage;
import org.apache.sling.distribution.journal.HandlerAdapter;
import org.apache.sling.distribution.journal.JournalAvailable;
import org.apache.sling.distribution.journal.MessageInfo;
import org.apache.sling.distribution.journal.MessagingProvider;
import org.apache.sling.distribution.journal.Reset;
import org.apache.sling.distribution.journal.impl.precondition.Precondition;
-import org.apache.sling.distribution.journal.impl.queue.QueueItemFactory;
import org.apache.sling.distribution.journal.impl.shared.DistributionMetricsService;
import org.apache.sling.distribution.journal.impl.shared.Topics;
import org.apache.sling.distribution.journal.messages.PackageMessage;
import org.apache.sling.distribution.journal.messages.PackageStatusMessage;
import org.apache.sling.distribution.packaging.DistributionPackageBuilder;
-import org.apache.sling.distribution.queue.DistributionQueueItem;
import org.apache.sling.settings.SlingSettingsService;
import org.osgi.framework.BundleContext;
import org.osgi.framework.ServiceRegistration;
@@ -94,8 +89,6 @@ public class DistributionSubscriber {
private static final Logger LOG = LoggerFactory.getLogger(DistributionSubscriber.class);
- private static final DistributionQueueItem STOPPED_ITEM = new DistributionQueueItem("stop-item", emptyMap());
-
@Reference(name = "packageBuilder")
private DistributionPackageBuilder packageBuilder;
@@ -141,7 +134,7 @@ public class DistributionSubscriber {
// Use a bounded internal buffer to allow reading further packages while working
// on one at a time
- private final BlockingQueue<DistributionQueueItem> queueItemsBuffer = new LinkedBlockingQueue<>(8);
+ private final BlockingQueue<FullMessage<PackageMessage>> messageBuffer = new LinkedBlockingQueue<>(8);
private Set<String> queueNames = Collections.emptySet();
@@ -239,13 +232,12 @@ public class DistributionSubscriber {
if (isBlocked) {
return DistributionAgentState.BLOCKED;
}
- return queueItemsBuffer.size() > 0 ? DistributionAgentState.RUNNING : DistributionAgentState.IDLE;
+ return messageBuffer.size() > 0 ? DistributionAgentState.RUNNING : DistributionAgentState.IDLE;
}
private void handlePackageMessage(MessageInfo info, PackageMessage message) {
if (shouldEnqueue(info, message)) {
- DistributionQueueItem queueItem = QueueItemFactory.fromPackage(info, message, true);
- enqueue(queueItem);
+ enqueue(new FullMessage<PackageMessage>(info, message));
} else {
try {
bookKeeper.skipPackage(info.getOffset());
@@ -272,10 +264,10 @@ public class DistributionSubscriber {
* packages fetched in memory. Note that each queued item contains the binary
* package to be imported.
*/
- private void enqueue(DistributionQueueItem queueItem) {
+ private void enqueue(FullMessage<PackageMessage> message) {
try {
while (running) {
- if (queueItemsBuffer.offer(queueItem, 1000, TimeUnit.MILLISECONDS)) {
+ if (messageBuffer.offer(message, 1000, TimeUnit.MILLISECONDS)) {
distributionMetricsService.getItemsBufferSize().increment();
return;
}
@@ -301,13 +293,13 @@ public class DistributionSubscriber {
return;
}
- DistributionQueueItem item = blockingPeekQueueItem();
- if (STOPPED_ITEM == item) {
+ Optional<FullMessage<PackageMessage>> item = blockingPeekQueueItem();
+ if (!item.isPresent()) {
return;
}
try (Timer.Context context = distributionMetricsService.getProcessQueueItemDuration().time()) {
- processQueueItem(item);
+ processQueueItem(item.get().getInfo(), item.get().getMessage());
} finally {
subscriberIdle.ifPresent(SubscriberIdle::idle);
}
@@ -340,30 +332,30 @@ public class DistributionSubscriber {
return running;
}
- private DistributionQueueItem blockingPeekQueueItem() throws InterruptedException {
+ private Optional<FullMessage<PackageMessage>> blockingPeekQueueItem() throws InterruptedException {
while (running) {
- DistributionQueueItem queueItem = queueItemsBuffer.peek();
- if (queueItem != null) {
- return queueItem;
+ FullMessage<PackageMessage> message = messageBuffer.peek();
+ if (message != null) {
+ return Optional.of(message);
} else {
Thread.sleep(QUEUE_FETCH_DELAY);
}
}
- return STOPPED_ITEM;
+ return Optional.empty();
}
- private void processQueueItem(DistributionQueueItem queueItem) throws PersistenceException, LoginException, DistributionException, TimeoutException {
- long offset = queueItem.get(RECORD_OFFSET, Long.class);
- PackageMessage pkgMsg = queueItem.get(PACKAGE_MSG, PackageMessage.class);
+ private void processQueueItem(MessageInfo info, PackageMessage queueItem) throws PersistenceException, LoginException, DistributionException, TimeoutException {
+ long offset = info.getOffset();
+ PackageMessage pkgMsg = queueItem;
boolean skip = shouldSkip(offset);
subscriberIdle.ifPresent(SubscriberIdle::busy);
if (skip) {
bookKeeper.removePackage(pkgMsg, offset);
} else {
- long createdTime = queueItem.get(RECORD_TIMESTAMP, Long.class);
+ long createdTime = info.getCreateTime();
bookKeeper.importPackage(pkgMsg, offset, createdTime);
}
- queueItemsBuffer.remove();
+ messageBuffer.remove();
distributionMetricsService.getItemsBufferSize().decrement();
}