From 08ca383fd59b9d5fb105dc7aab293b3843a22633 Mon Sep 17 00:00:00 2001 From: kimchy Date: Sat, 17 Jul 2010 15:20:13 +0300 Subject: [PATCH] more work on cloud gateway --- .idea/dictionaries/kimchy.xml | 1 + .../ImmutableAppendableBlobContainer.java | 2 + .../blobstore/AbstractCloudBlobContainer.java | 56 +++++++------------ .../CloudImmutableBlobContainer.java | 2 +- 4 files changed, 23 insertions(+), 38 deletions(-) diff --git a/.idea/dictionaries/kimchy.xml b/.idea/dictionaries/kimchy.xml index 174cd34a592..d475fd801a8 100644 --- a/.idea/dictionaries/kimchy.xml +++ b/.idea/dictionaries/kimchy.xml @@ -107,6 +107,7 @@ startup stopwords streamable + substring successul tagline threadpool diff --git a/modules/elasticsearch/src/main/java/org/elasticsearch/common/blobstore/support/ImmutableAppendableBlobContainer.java b/modules/elasticsearch/src/main/java/org/elasticsearch/common/blobstore/support/ImmutableAppendableBlobContainer.java index 45ba0a44f5e..0f9698e9ba9 100644 --- a/modules/elasticsearch/src/main/java/org/elasticsearch/common/blobstore/support/ImmutableAppendableBlobContainer.java +++ b/modules/elasticsearch/src/main/java/org/elasticsearch/common/blobstore/support/ImmutableAppendableBlobContainer.java @@ -86,6 +86,8 @@ public class ImmutableAppendableBlobContainer extends AbstractBlobContainer impl part++; if (container.blobExists(blobName + ".a" + part)) { container.readBlob(blobName + ".a" + part, this); + } else { + listener.onCompleted(); } } diff --git a/plugins/cloud/src/main/java/org/elasticsearch/cloud/blobstore/AbstractCloudBlobContainer.java b/plugins/cloud/src/main/java/org/elasticsearch/cloud/blobstore/AbstractCloudBlobContainer.java index 5b29c3b745f..13a0b38c74a 100644 --- a/plugins/cloud/src/main/java/org/elasticsearch/cloud/blobstore/AbstractCloudBlobContainer.java +++ b/plugins/cloud/src/main/java/org/elasticsearch/cloud/blobstore/AbstractCloudBlobContainer.java @@ -22,10 +22,10 @@ package org.elasticsearch.cloud.blobstore; import com.google.common.util.concurrent.ListenableFuture; import org.elasticsearch.common.blobstore.BlobMetaData; import org.elasticsearch.common.blobstore.BlobPath; +import org.elasticsearch.common.blobstore.BlobStoreException; import org.elasticsearch.common.blobstore.support.AbstractBlobContainer; import org.elasticsearch.common.blobstore.support.PlainBlobMetaData; import org.elasticsearch.common.collect.ImmutableMap; -import org.elasticsearch.common.collect.Maps; import org.jclouds.blobstore.domain.Blob; import org.jclouds.blobstore.domain.PageSet; import org.jclouds.blobstore.domain.StorageMetadata; @@ -33,7 +33,6 @@ import org.jclouds.blobstore.options.ListContainerOptions; import java.io.IOException; import java.io.InputStream; -import java.util.Map; import java.util.concurrent.ExecutionException; /** @@ -60,13 +59,17 @@ public class AbstractCloudBlobContainer extends AbstractBlobContainer { return cloudBlobStore.sync().blobExists(cloudBlobStore.container(), buildBlobPath(blobName)); } - @Override public void readBlob(String blobName, final ReadBlobListener listener) { + @Override public void readBlob(final String blobName, final ReadBlobListener listener) { final ListenableFuture future = cloudBlobStore.async().getBlob(cloudBlobStore.container(), buildBlobPath(blobName)); future.addListener(new Runnable() { @Override public void run() { Blob blob; try { blob = future.get(); + if (blob == null) { + listener.onFailure(new BlobStoreException("No blob found for [" + buildBlobPath(blobName) + "]")); + return; + } } catch (InterruptedException e) { listener.onFailure(e); return; @@ -94,20 +97,23 @@ public class AbstractCloudBlobContainer extends AbstractBlobContainer { }, cloudBlobStore.executorService()); } - @Override public ImmutableMap listBlobsByPrefix(String blobNamePrefix) throws IOException { - PageSet list = cloudBlobStore.sync().list(cloudBlobStore.container(), ListContainerOptions.Builder.recursive().inDirectory(blobNamePrefix)); - ImmutableMap.Builder blobs = ImmutableMap.builder(); - for (StorageMetadata storageMetadata : list) { - blobs.put(storageMetadata.getName(), new PlainBlobMetaData(storageMetadata.getName(), storageMetadata.getSize(), null)); - } - return blobs.build(); - } + // inDirectory expects a directory, not a blob prefix +// @Override public ImmutableMap listBlobsByPrefix(String blobNamePrefix) throws IOException { +// PageSet list = cloudBlobStore.sync().list(cloudBlobStore.container(), ListContainerOptions.Builder.recursive().inDirectory(buildBlobPath(blobNamePrefix))); +// ImmutableMap.Builder blobs = ImmutableMap.builder(); +// for (StorageMetadata storageMetadata : list) { +// String name = storageMetadata.getName().substring(cloudPath.length() + 1); +// blobs.put(name, new PlainBlobMetaData(name, storageMetadata.getSize(), null)); +// } +// return blobs.build(); +// } @Override public ImmutableMap listBlobs() throws IOException { - PageSet list = cloudBlobStore.sync().list(cloudBlobStore.container(), ListContainerOptions.Builder.recursive()); + PageSet list = cloudBlobStore.sync().list(cloudBlobStore.container(), ListContainerOptions.Builder.recursive().inDirectory(cloudPath)); ImmutableMap.Builder blobs = ImmutableMap.builder(); for (StorageMetadata storageMetadata : list) { - blobs.put(storageMetadata.getName(), new PlainBlobMetaData(storageMetadata.getName(), storageMetadata.getSize(), null)); + String name = storageMetadata.getName().substring(cloudPath.length() + 1); + blobs.put(name, new PlainBlobMetaData(name, storageMetadata.getSize(), null)); } return blobs.build(); } @@ -115,28 +121,4 @@ public class AbstractCloudBlobContainer extends AbstractBlobContainer { protected String buildBlobPath(String blobName) { return cloudPath + "/" + blobName; } - - private Map list(String container, String prefix) { - final Map allMetaDatas = Maps.newHashMap(); - - String nextMarker = null; - while (true) { - ListContainerOptions options = ListContainerOptions.Builder.recursive(); - if (prefix != null) { - options = options.inDirectory(prefix); - } - if (nextMarker != null) { - options.afterMarker(nextMarker); - } - PageSet pageSet = cloudBlobStore.sync().list(container, options); - for (StorageMetadata metadata : pageSet) { - allMetaDatas.put(metadata.getName(), metadata); - } - nextMarker = pageSet.getNextMarker(); - if (nextMarker == null) { - break; - } - } - return allMetaDatas; - } } diff --git a/plugins/cloud/src/main/java/org/elasticsearch/cloud/blobstore/CloudImmutableBlobContainer.java b/plugins/cloud/src/main/java/org/elasticsearch/cloud/blobstore/CloudImmutableBlobContainer.java index 98163db1cc9..9497a5b985f 100644 --- a/plugins/cloud/src/main/java/org/elasticsearch/cloud/blobstore/CloudImmutableBlobContainer.java +++ b/plugins/cloud/src/main/java/org/elasticsearch/cloud/blobstore/CloudImmutableBlobContainer.java @@ -39,7 +39,7 @@ public class CloudImmutableBlobContainer extends AbstractCloudBlobContainer impl } @Override public void writeBlob(String blobName, InputStream is, long sizeInBytes, final WriterListener listener) { - Blob blob = cloudBlobStore.sync().newBlob(blobName); + Blob blob = cloudBlobStore.sync().newBlob(buildBlobPath(blobName)); blob.setPayload(is); blob.setContentLength(sizeInBytes); final ListenableFuture future = cloudBlobStore.async().putBlob(cloudBlobStore.container(), blob);