Refactored tile range deletion to be streamed.

Author: axl8713

geowebcache/azureblob/src/main/java/org/geowebcache/azure/AzureBlobStore.java

@@ -28,9 +28,11 @@
 import java.io.UncheckedIOException;
 import java.time.OffsetDateTime;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.NoSuchElementException;
 import java.util.Optional;
 import java.util.Properties;
 import java.util.concurrent.Callable;
@@ -39,6 +41,7 @@
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
+import java.util.stream.StreamSupport;
 import javax.annotation.Nullable;
 import org.geotools.util.logging.Logging;
 import org.geowebcache.filter.parameters.ParametersUtils;
@@ -187,66 +190,101 @@ public boolean delete(TileObject obj) throws StorageException {
 
     @Override
     public boolean delete(TileRange tileRange) throws StorageException {
-        // see if there is anything to delete in that range by computing a prefix
-        final String coordsPrefix = keyBuilder.coordinatesPrefix(tileRange, false);
-        if (!client.prefixExists(coordsPrefix)) {
-            return false;
-        }
+        try {
+            // see if there is anything to delete in that range by computing a prefix
+            final String coordsPrefix = keyBuilder.coordinatesPrefix(tileRange, false);
+            if (!client.prefixExists(coordsPrefix)) {
+                return false;
+            }
 
-        List<BlobItem> blobsToDelete = findTileBlobsToDelete(tileRange, coordsPrefix);
+            findTileBlobsToDelete(tileRange, coordsPrefix).forEach(blobs -> deleteTileBlobs(blobs, tileRange));
 
-        if (listeners.isEmpty()) {
-            // if there are no listeners, don't bother requesting every tile
-            // metadata to notify the listeners
-            // split the iteration in parts to avoid memory accumulation
-            List<String> keysToDelete =
-                    blobsToDelete.stream().map(BlobItem::getName).collect(Collectors.toList());
+            return true;
 
-            Iterator<List<String>> partition = Iterators.partition(keysToDelete.iterator(), DeleteManager.PAGE_SIZE);
+        } catch (RuntimeException e) {
+            throw new StorageException("Failed to delete tile range", e);
+        }
+    }
 
-            while (partition.hasNext() && !shutDown) {
-                deleteManager.deleteParallel(partition.next());
-            }
+    /** Splits the deletion in parts to avoid memory accumulation. */
+    private Stream<List<BlobItem>> findTileBlobsToDelete(TileRange tileRange, String coordsPrefix) {
 
-        } else {
-            // if we need to gather info, we'll end up just calling "delete" on each tile
-            // this is run here instead of inside the delete manager as we need high level info
-            // about tiles, e.g., TileObject, to inform the listeners
-            List<Callable<?>> tilesDeletions = blobsToDelete.stream()
-                    .map(blobItem -> {
-                        TileObject tile = createTileObject(blobItem, tileRange);
-                        tile.setParametersId(tileRange.getParametersId());
-                        return (Callable<Object>) () -> delete(tile);
-                    })
-                    .collect(Collectors.toList());
-
-            executeParallelDeletions(tilesDeletions);
-        }
+        Iterable<List<BlobItem>> iterable = () -> new Iterator<>() {
+            private final int zoomStart = tileRange.getZoomStart();
+            private final int zoomStop = tileRange.getZoomStop();
+            private int currentZoom = zoomStart;
+            private Iterator<BlobItem> currentBlobIterator = Collections.emptyIterator();
 
-        return true;
-    }
+            @Override
+            public boolean hasNext() {
+                while (!currentBlobIterator.hasNext() && currentZoom <= zoomStop) {
+                    String zoomPrefix = coordsPrefix + "/" + currentZoom;
+
+                    if (!client.prefixExists(zoomPrefix)) {
+                        // empty level, skipping
+                        currentZoom++;
+                        continue;
+                    }
 
-    private List<BlobItem> findTileBlobsToDelete(TileRange tileRange, String coordsPrefix) {
+                    long[] rangeBoundsAtZoom = tileRange.rangeBounds(currentZoom);
+                    currentBlobIterator = client.listBlobs(zoomPrefix)
+                            .filter(tb -> isTileBlobInBounds(tb, rangeBoundsAtZoom))
+                            .iterator();
 
-        List<BlobItem> blobsToDelete = new ArrayList<>();
+                    currentZoom++;
+                }
 
-        for (int zoom = tileRange.getZoomStart(); zoom <= tileRange.getZoomStop(); zoom++) {
+                return currentBlobIterator.hasNext();
+            }
 
-            String zoomPrefix = coordsPrefix + "/" + zoom;
+            @Override
+            public List<BlobItem> next() {
+                if (!hasNext()) {
+                    throw new NoSuchElementException();
+                }
 
-            if (!client.prefixExists(zoomPrefix)) {
-                // empty level, skipping
-                continue;
+                List<BlobItem> batch = new ArrayList<>(DeleteManager.PAGE_SIZE);
+                while (currentBlobIterator.hasNext() && batch.size() < DeleteManager.PAGE_SIZE) {
+                    batch.add(currentBlobIterator.next());
+                }
+                return batch;
             }
+        };
 
-            long[] rangeBoundsAtZoom = tileRange.rangeBounds(zoom);
+        return StreamSupport.stream(iterable.spliterator(), false);
+    }
 
-            client.listBlobs(zoomPrefix)
-                    .filter(tb -> isTileBlobInBounds(tb, rangeBoundsAtZoom))
-                    .forEach(blobsToDelete::add);
+    private void deleteTileBlobs(List<BlobItem> blobs, TileRange tileRange) {
+        try {
+            if (listeners.isEmpty()) {
+                // if there are no listeners, don't bother requesting every tile
+                // metadata to notify the listeners
+                List<String> keysToDelete =
+                        blobs.stream().map(BlobItem::getName).collect(Collectors.toList());
+
+                if (!shutDown) {
+                    deleteManager.deleteParallel(keysToDelete);
+                }
+            } else {
+                // if we need to gather info, we'll end up just calling "delete" on each tile
+                // this is run here instead of inside the delete manager as we need high level info
+                // about tiles, e.g., TileObject, to inform the listeners
+                List<Callable<?>> tilesDeletions = blobs.stream()
+                        .map(blobItem -> {
+                            TileObject tile = createTileObject(blobItem, tileRange);
+                            tile.setParametersId(tileRange.getParametersId());
+
+                            // PMD is being picky with this lambda inlined
+                            Callable<Boolean> deleteCallable = () -> delete(tile);
+                            return deleteCallable;
+                        })
+                        .collect(Collectors.toList());
+
+                executeParallelDeletions(tilesDeletions);
+            }
+        } catch (StorageException e) {
+            throw new RuntimeException(e);
         }
-
-        return blobsToDelete;
     }
 
     private boolean isTileBlobInBounds(BlobItem tileBlob, long[] bounds) {