Fixed multiple calls to listener.

Added simplest bounded delete.
Added test for bounded delete.
Added test to check if TileDeleted events are received when a layer is deleted. There is a race in this test, so it can pass even though tileDeleted is sent.

Author: alanmcdade

geowebcache/s3storage/src/main/java/org/geowebcache/s3/S3BlobStore.java

@@ -486,6 +486,14 @@ public Bounds(long[] bound) {
             maxY = Math.max(bound[1], bound[3]);
         }
 
+        public long getMinX() {
+            return minX;
+        }
+
+        public long getMaxX() {
+            return maxX;
+        }
+
         static Optional<Bounds> createBounds(String prefix) {
             Matcher matcher = boundsRegex.matcher(prefix);
             if (!matcher.matches()) {

geowebcache/s3storage/src/main/java/org/geowebcache/s3/S3Ops.java

@@ -35,6 +35,7 @@
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.Properties;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ConcurrentHashMap;
@@ -54,9 +55,10 @@
 import org.geowebcache.locks.NoOpLockProvider;
 import org.geowebcache.s3.S3BlobStore.Bounds;
 import org.geowebcache.s3.streams.BatchingIterator;
+import org.geowebcache.s3.streams.BoundedS3KeySupplier;
 import org.geowebcache.s3.streams.DeleteBatchesOfS3Objects;
-import org.geowebcache.s3.streams.S3ObjectForPrefixSupplier;
 import org.geowebcache.s3.streams.TileDeletionListenerNotifier;
+import org.geowebcache.s3.streams.UnboundedS3KeySupplier;
 import org.geowebcache.storage.BlobStoreListenerList;
 import org.geowebcache.storage.StorageException;
 import org.geowebcache.util.TMSKeyBuilder;
@@ -426,7 +428,7 @@ private long deleteBatchesOfTilesAndInformListeners() {
                     possibleBounds.isPresent() ? tileDeletionListenerNotifier : NO_OPERATION_POST_PROCESSOR;
 
             return BatchingIterator.batchedStreamOf(
-                            createS3ObjectStream()
+                            createS3ObjectStream(possibleBounds)
                                     .takeWhile(Objects::nonNull)
                                     .takeWhile(o -> !Thread.currentThread().isInterrupted())
                                     .filter(timeStampFilter),
@@ -437,8 +439,7 @@ private long deleteBatchesOfTilesAndInformListeners() {
                     .sum();
         }
 
-        private Stream<S3ObjectSummary> createS3ObjectStream() {
-            var possibleBounds = Bounds.createBounds(prefix);
+        private Stream<S3ObjectSummary> createS3ObjectStream(Optional<Bounds> possibleBounds) {
             if (possibleBounds.isPresent()) {
                 String prefixWithoutBounds = prefixWithoutBounds(prefix);
                 return boundedStreamOfS3Objects(prefixWithoutBounds, possibleBounds.get());
@@ -449,15 +450,16 @@ private Stream<S3ObjectSummary> createS3ObjectStream() {
 
         private Stream<S3ObjectSummary> unboundedStreamOfS3Objects(String prefix) {
             S3Objects s3Objects = S3Objects.withPrefix(conn, bucketName, prefix).withBatchSize(BATCH_SIZE);
-            S3ObjectForPrefixSupplier supplier = new S3ObjectForPrefixSupplier(prefix, bucketName, s3Objects, logger);
+            UnboundedS3KeySupplier supplier = new UnboundedS3KeySupplier(prefix, bucketName, s3Objects, logger);
             return Stream.generate(supplier).takeWhile(Objects::nonNull);
         }
 
         private Stream<S3ObjectSummary> boundedStreamOfS3Objects(String prefixWithoutBounds, Bounds bounds) {
-            S3Objects s3Objects =
-                    S3Objects.withPrefix(conn, bucketName, prefixWithoutBounds).withBatchSize(BATCH_SIZE);
-            S3ObjectForPrefixSupplier supplier = new S3ObjectForPrefixSupplier(prefix, bucketName, s3Objects, logger);
-            return Stream.generate(supplier).takeWhile(Objects::nonNull).filter(bounds::predicate);
+            BoundedS3KeySupplier supplier =
+                    new BoundedS3KeySupplier(prefixWithoutBounds, logger, conn, bounds, bucketName, BATCH_SIZE);
+            return Stream.generate(supplier)
+                    .takeWhile(Objects::nonNull)
+                    .filter(bounds::predicate); // Filter Y bounds as X is taken care of by the supplier
         }
 
         private void checkInterrupted() throws InterruptedException {

geowebcache/s3storage/src/main/java/org/geowebcache/s3/streams/BoundedS3KeySupplier.java

@@ -0,0 +1,78 @@
+package org.geowebcache.s3.streams;
+
+import static java.lang.String.format;
+
+import com.amazonaws.services.s3.AmazonS3;
+import com.amazonaws.services.s3.iterable.S3Objects;
+import com.amazonaws.services.s3.model.S3ObjectSummary;
+
+import java.util.Iterator;
+import java.util.function.Supplier;
+import java.util.logging.Logger;
+
+import org.geowebcache.s3.S3BlobStore.Bounds;
+
+/**
+ * Similar to the UnboundedS3KeySupplier it retrieves keys from S3. It is slightly more optimised as it respects the x
+ * bounds only fetching objects that are from the range of x bounds S3ObjectPathsForPrefixSupplier This class will
+ * interact with the AmazonS3 connection to retrieve all the objects with prefix and bucket provided <br>
+ * It will return these lazily one by one as the get methods is called
+ */
+public class BoundedS3KeySupplier implements Supplier<S3ObjectSummary> {
+    private final String prefixWithoutBounds;
+    private final Logger logger;
+    private final AmazonS3 conn;
+    private final Bounds bounds;
+    private final String bucket;
+    private final int batch;
+
+    public BoundedS3KeySupplier(
+            String prefixWithoutBounds, Logger logger, AmazonS3 conn, Bounds bounds, String bucket, int batch) {
+        this.prefixWithoutBounds = prefixWithoutBounds;
+        this.logger = logger;
+        this.conn = conn;
+        this.bounds = bounds;
+        this.nextX = bounds.getMinX();
+        this.bucket = bucket;
+        this.batch = batch;
+    }
+
+    private Iterator<S3ObjectSummary> iterator;
+    private long nextX;
+    private long count = 0;
+
+    @Override
+    public S3ObjectSummary get() {
+        return next();
+    }
+
+    private synchronized S3ObjectSummary next() {
+        boolean hasNext = false;
+        do {
+            hasNext = iterator != null && iterator.hasNext();
+            if (!hasNext) {
+                iterator = null;
+            }
+
+            if (iterator == null && nextX <= bounds.getMaxX()) {
+                String prefixWithX = format("%s%d/", prefixWithoutBounds, nextX);
+                S3Objects s3Objects =
+                        S3Objects.withPrefix(conn, bucket, prefixWithX).withBatchSize(batch);
+                iterator = s3Objects.iterator();
+                hasNext = iterator.hasNext();
+                nextX++;
+            }
+        } while (!hasNext && nextX <= bounds.getMaxX()); // It is exhausted if
+
+        if (hasNext) {
+            count++;
+            S3ObjectSummary summary = iterator.next();
+            logger.fine(format("%s: %s", summary.getKey(), bounds));
+            return summary;
+        } else {
+            logger.info(String.format(
+                    "Exhausted objects with prefix: %s supplied %d", prefixWithoutBounds + bounds, count));
+            return null;
+        }
+    }
+}

geowebcache/s3storage/src/main/java/org/geowebcache/s3/streams/DeleteBatchesOfS3Objects.java

@@ -25,6 +25,7 @@
 import java.util.logging.Logger;
 import java.util.stream.Collectors;
 
+/** @param <T> The type of the data object used to track abstract the s3 class */
 public class DeleteBatchesOfS3Objects<T> implements Function<List<T>, List<T>> {
     private final String bucket;
     private final AmazonS3 conn;

…3/streams/S3ObjectForPrefixSupplier.java …e/s3/streams/UnboundedS3KeySupplier.javageowebcache/s3storage/src/main/java/org/geowebcache/s3/streams/S3ObjectForPrefixSupplier.java renamed to geowebcache/s3storage/src/main/java/org/geowebcache/s3/streams/UnboundedS3KeySupplier.java geowebcache/s3storage/src/main/java/org/geowebcache/s3/streams/S3ObjectForPrefixSupplier.java renamed to geowebcache/s3storage/src/main/java/org/geowebcache/s3/streams/UnboundedS3KeySupplier.java

@@ -20,19 +20,19 @@
 import java.util.logging.Logger;
 
 /**
- * S3ObjectPathsForPrefixSupplier This class will interact with the AmazonS3 connection to retrieve all the objects with
- * prefix and bucket provided <br>
+ * UnboundedS3KeySupplier This class will interact with the AmazonS3 connection to retrieve all the objects with prefix
+ * and bucket provided <br>
  * It will return these lazily one by one as the get methods is called
  */
-public class S3ObjectForPrefixSupplier implements Supplier<S3ObjectSummary> {
+public class UnboundedS3KeySupplier implements Supplier<S3ObjectSummary> {
     private final String prefix;
     private long count = 0;
     private final Logger logger;
     private final S3Objects s3Objects;
 
     private Iterator<S3ObjectSummary> iterator;
 
-    public S3ObjectForPrefixSupplier(String prefix, String bucket, S3Objects s3Objects, Logger logger) {
+    public UnboundedS3KeySupplier(String prefix, String bucket, S3Objects s3Objects, Logger logger) {
         checkNotNull(prefix, "prefix must not be null");
         checkNotNull(bucket, "bucket must not be null");
         checkNotNull(s3Objects, "s3Objects must not be null");

geowebcache/s3storage/src/test/java/org/geowebcache/s3/AbstractS3BlobStoreIntegrationTest.java

@@ -17,13 +17,11 @@
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
-import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.ArgumentMatchers.isNull;
-import static org.mockito.Mockito.doNothing;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
@@ -432,12 +430,8 @@ public void testBoundedLayerDeletion() throws StorageException, MimeException {
 
         int level = 3;
         seed(level, level);
-        BlobStoreListener listener = mock(BlobStoreListener.class);
-        blobStore.addListener(listener);
-        doNothing()
-                .when(listener)
-                .tileDeleted(
-                        anyString(), anyString(), anyString(), isNull(), anyLong(), anyLong(), anyInt(), anyLong());
+        FakeListener fakeListener = new FakeListener();
+        blobStore.addListener(fakeListener);
 
         long[][] rangeBounds = {{2, 2, 3, 3, level}};
 
@@ -455,9 +449,9 @@ public void testBoundedLayerDeletion() throws StorageException, MimeException {
         }
 
         int wantedNumberOfInvocations =
-                (int) ((rangeBounds[0][2] - rangeBounds[0][0] + 1) * (rangeBounds[0][level] - rangeBounds[0][1] + 1));
-        Awaitility.await().untilAsserted(() -> verify(listener, times(wantedNumberOfInvocations))
-                .tileDeleted(anyString(), anyString(), anyString(), any(), anyLong(), anyLong(), anyInt(), anyLong()));
+                (int) ((rangeBounds[0][2] - rangeBounds[0][0] + 1) * (rangeBounds[0][3] - rangeBounds[0][1] + 1));
+        Awaitility.await().untilAsserted(() -> assertEquals(wantedNumberOfInvocations, fakeListener.tileDeleted));
+        assertEquals(wantedNumberOfInvocations, fakeListener.total());
     }
 
     private TileRange tileRange(