gce - firewall api

apis/atmos/src/main/java/org/jclouds/atmos/blobstore/AtmosAsyncBlobStore.java

@@ -19,12 +19,12 @@
 package org.jclouds.atmos.blobstore;
 
 import static com.google.common.base.Preconditions.checkNotNull;
+import static com.google.common.util.concurrent.Futures.transform;
 import static org.jclouds.atmos.options.PutOptions.Builder.publicRead;
 
 import java.net.URI;
 import java.util.Set;
 import java.util.concurrent.Callable;
-import java.util.concurrent.ExecutorService;
 
 import javax.inject.Inject;
 import javax.inject.Named;
@@ -56,7 +56,6 @@
 import org.jclouds.blobstore.strategy.internal.FetchBlobMetadata;
 import org.jclouds.blobstore.util.BlobUtils;
 import org.jclouds.collect.Memoized;
-import org.jclouds.concurrent.Futures;
 import org.jclouds.crypto.Crypto;
 import org.jclouds.domain.Location;
 import org.jclouds.http.options.GetOptions;
@@ -66,6 +65,7 @@
 import com.google.common.cache.CacheLoader;
 import com.google.common.cache.LoadingCache;
 import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.ListeningExecutorService;
 
 /**
  * @author Adrian Cole
@@ -86,14 +86,14 @@ public class AtmosAsyncBlobStore extends BaseAsyncBlobStore {
 
    @Inject
    AtmosAsyncBlobStore(BlobStoreContext context, BlobUtils blobUtils,
-            @Named(Constants.PROPERTY_USER_THREADS) ExecutorService service, Supplier<Location> defaultLocation,
+            @Named(Constants.PROPERTY_USER_THREADS) ListeningExecutorService userExecutor, Supplier<Location> defaultLocation,
             @Memoized Supplier<Set<? extends Location>> locations, AtmosAsyncClient async, AtmosClient sync,
             ObjectToBlob object2Blob, ObjectToBlobMetadata object2BlobMd, BlobToObject blob2Object,
             BlobStoreListOptionsToListOptions container2ContainerListOptions,
             DirectoryEntryListToResourceMetadataList container2ResourceList, Crypto crypto,
             BlobToHttpGetOptions blob2ObjectGetOptions, Provider<FetchBlobMetadata> fetchBlobMetadataProvider,
             LoadingCache<String, Boolean> isPublic) {
-      super(context, blobUtils, service, defaultLocation, locations);
+      super(context, blobUtils, userExecutor, defaultLocation, locations);
       this.blob2ObjectGetOptions = checkNotNull(blob2ObjectGetOptions, "blob2ObjectGetOptions");
       this.sync = checkNotNull(sync, "sync");
       this.async = checkNotNull(async, "async");
@@ -113,12 +113,12 @@ public class AtmosAsyncBlobStore extends BaseAsyncBlobStore {
     */
    @Override
    public ListenableFuture<BlobMetadata> blobMetadata(String container, String key) {
-      return Futures.compose(async.headFile(container + "/" + key), new Function<AtmosObject, BlobMetadata>() {
+      return transform(async.headFile(container + "/" + key), new Function<AtmosObject, BlobMetadata>() {
          @Override
          public BlobMetadata apply(AtmosObject from) {
             return object2BlobMd.apply(from);
          }
-      }, service);
+      }, userExecutor);
    }
 
    /**
@@ -128,27 +128,23 @@ public BlobMetadata apply(AtmosObject from) {
     */
    @Override
    public ListenableFuture<Boolean> createContainerInLocation(Location location, String container) {
-      return Futures.compose(async.createDirectory(container), new Function<URI, Boolean>() {
-
+      return transform(async.createDirectory(container), new Function<URI, Boolean>() {
          public Boolean apply(URI from) {
             return true;
          }
-
-      }, service);
+      }, userExecutor);
    }
 
    /**
     * This implementation invokes {@link AtmosAsyncClient#createDirectory}
     */
    @Override
    public ListenableFuture<Void> createDirectory(String container, String directory) {
-      return Futures.compose(async.createDirectory(container + "/" + directory), new Function<URI, Void>() {
-
+      return transform(async.createDirectory(container + "/" + directory), new Function<URI, Void>() {
          public Void apply(URI from) {
             return null;// no etag
          }
-
-      }, service);
+      }, userExecutor);
    }
 
    /**
@@ -204,15 +200,15 @@ public ListenableFuture<Boolean> blobExists(String container, String key) {
    public ListenableFuture<Blob> getBlob(String container, String key, org.jclouds.blobstore.options.GetOptions options) {
       GetOptions httpOptions = blob2ObjectGetOptions.apply(options);
       ListenableFuture<AtmosObject> returnVal = async.readFile(container + "/" + key, httpOptions);
-      return Futures.compose(returnVal, object2Blob, service);
+      return transform(returnVal, object2Blob, userExecutor);
    }
 
    /**
     * This implementation invokes {@link AtmosAsyncClient#listDirectories}
     */
    @Override
    public ListenableFuture<PageSet<? extends StorageMetadata>> list() {
-      return Futures.compose(async.listDirectories(), container2ResourceList, service);
+      return transform(async.listDirectories(), container2ResourceList, userExecutor);
    }
 
    /**
@@ -224,10 +220,10 @@ public ListenableFuture<PageSet<? extends StorageMetadata>> list(String containe
       container = AtmosUtils.adjustContainerIfDirOptionPresent(container, options);
       ListOptions nativeOptions = container2ContainerListOptions.apply(options);
       ListenableFuture<BoundedSet<? extends DirectoryEntry>> returnVal = async.listDirectory(container, nativeOptions);
-      ListenableFuture<PageSet<? extends StorageMetadata>> list = Futures.compose(returnVal, container2ResourceList,
-               service);
-      return (ListenableFuture<PageSet<? extends StorageMetadata>>) (options.isDetailed() ? Futures.compose(list,
-               fetchBlobMetadataProvider.get().setContainerName(container), service) : list);
+      ListenableFuture<PageSet<? extends StorageMetadata>> list = transform(returnVal, container2ResourceList,
+            userExecutor);
+      return (ListenableFuture<PageSet<? extends StorageMetadata>>) (options.isDetailed() ? transform(list,
+               fetchBlobMetadataProvider.get().setContainerName(container)) : list);
    }
 
    /**
@@ -244,7 +240,7 @@ public ListenableFuture<String> putBlob(final String container, final Blob blob)
       } catch (CacheLoader.InvalidCacheLoadException e) {
          // nulls not permitted
       }
-      return Futures.makeListenable(service.submit(new Callable<String>() {
+      return userExecutor.submit(new Callable<String>() {
 
          @Override
          public String call() throws Exception {
@@ -255,7 +251,7 @@ public String call() throws Exception {
          public String toString() {
             return "putBlob(" + container + "," + blob.getMetadata().getName() + ")";
          }
-      }), service);
+      });
 
    }
 
@@ -277,13 +273,13 @@ public ListenableFuture<String> putBlob(String container, Blob blob, PutOptions
    public ListenableFuture<Boolean> createContainerInLocation(Location location, String container,
             CreateContainerOptions options) {
       if (options.isPublicRead())
-         return Futures.compose(async.createDirectory(container, publicRead()), new Function<URI, Boolean>() {
+         return transform(async.createDirectory(container, publicRead()), new Function<URI, Boolean>() {
 
             public Boolean apply(URI from) {
                return true;
             }
 
-         }, service);
+         }, userExecutor);
       return createContainerInLocation(location, container);
    }
 

apis/atmos/src/main/java/org/jclouds/atmos/blobstore/AtmosBlobRequestSigner.java

@@ -33,6 +33,7 @@
 import org.jclouds.blobstore.functions.BlobToHttpGetOptions;
 import org.jclouds.http.HttpRequest;
 import org.jclouds.http.options.GetOptions;
+import org.jclouds.reflect.Invocation;
 import org.jclouds.rest.internal.RestAnnotationProcessor;
 
 import com.google.common.collect.ImmutableList;
@@ -44,7 +45,7 @@
  */
 @Singleton
 public class AtmosBlobRequestSigner implements BlobRequestSigner {
-   private final RestAnnotationProcessor processor;
+   private final RestAnnotationProcessor<AtmosAsyncClient> processor;
    private final BlobToObject blobToObject;
    private final BlobToHttpGetOptions blob2ObjectGetOptions;
 
@@ -53,9 +54,9 @@ public class AtmosBlobRequestSigner implements BlobRequestSigner {
    private final Invokable<?, ?> createMethod;
 
    @Inject
-   public AtmosBlobRequestSigner(RestAnnotationProcessor.Factory processor, BlobToObject blobToObject,
+   public AtmosBlobRequestSigner(RestAnnotationProcessor<AtmosAsyncClient> processor, BlobToObject blobToObject,
          BlobToHttpGetOptions blob2ObjectGetOptions) throws SecurityException, NoSuchMethodException {
-      this.processor = checkNotNull(processor, "processor").declaring(AtmosAsyncClient.class);
+      this.processor = checkNotNull(processor, "processor");
       this.blobToObject = checkNotNull(blobToObject, "blobToObject");
       this.blob2ObjectGetOptions = checkNotNull(blob2ObjectGetOptions, "blob2ObjectGetOptions");
       this.getMethod = Invokable.from(AtmosAsyncClient.class.getMethod("readFile", String.class, GetOptions[].class));
@@ -68,7 +69,8 @@ public AtmosBlobRequestSigner(RestAnnotationProcessor.Factory processor, BlobToO
    public HttpRequest signGetBlob(String container, String name) {
       checkNotNull(container, "container");
       checkNotNull(name, "name");
-      return cleanRequest(processor.createRequest(getMethod, ImmutableList.<Object> of(getPath(container, name))));
+      return cleanRequest(processor.apply(Invocation.create(getMethod,
+            ImmutableList.<Object> of(getPath(container, name)))));
    }
 
    @Override
@@ -80,8 +82,8 @@ public HttpRequest signGetBlob(String container, String name, long timeInSeconds
    public HttpRequest signPutBlob(String container, Blob blob) {
       checkNotNull(container, "container");
       checkNotNull(blob, "blob");
-      return cleanRequest(processor.createRequest(createMethod,
-            ImmutableList.<Object> of(container, blobToObject.apply(blob))));
+      return cleanRequest(processor.apply(Invocation.create(createMethod,
+            ImmutableList.<Object> of(container, blobToObject.apply(blob)))));
    }
 
    @Override
@@ -93,7 +95,8 @@ public HttpRequest signPutBlob(String container, Blob blob, long timeInSeconds)
    public HttpRequest signRemoveBlob(String container, String name) {
       checkNotNull(container, "container");
       checkNotNull(name, "name");
-      return cleanRequest(processor.createRequest(deleteMethod, ImmutableList.<Object> of(getPath(container, name))));
+      return cleanRequest(processor.apply(Invocation.create(deleteMethod,
+            ImmutableList.<Object> of(getPath(container, name)))));
    }
 
    private String getPath(String container, String name) {
@@ -104,8 +107,8 @@ private String getPath(String container, String name) {
    public HttpRequest signGetBlob(String container, String name, org.jclouds.blobstore.options.GetOptions options) {
       checkNotNull(container, "container");
       checkNotNull(name, "name");
-      return cleanRequest(processor.createRequest(getMethod,
-            ImmutableList.of(getPath(container, name), blob2ObjectGetOptions.apply(checkNotNull(options, "options")))));
+      return cleanRequest(processor.apply(Invocation.create(getMethod,
+            ImmutableList.of(getPath(container, name), blob2ObjectGetOptions.apply(checkNotNull(options, "options"))))));
    }
 
 }

apis/atmos/src/main/java/org/jclouds/atmos/blobstore/strategy/FindMD5InUserMetadata.java

@@ -26,11 +26,9 @@
 import java.util.Map;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Future;
 import java.util.concurrent.SynchronousQueue;
-import java.util.concurrent.TimeoutException;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
 import javax.annotation.Resource;
 import javax.inject.Named;
@@ -45,12 +43,12 @@
 import org.jclouds.blobstore.options.ListContainerOptions;
 import org.jclouds.blobstore.strategy.ContainsValueInListStrategy;
 import org.jclouds.blobstore.strategy.ListBlobsInContainer;
-import org.jclouds.concurrent.Futures;
 import org.jclouds.logging.Logger;
 
 import com.google.common.base.Throwables;
 import com.google.common.collect.Maps;
 import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.ListeningExecutorService;
 import com.google.inject.Inject;
 
 /**
@@ -65,7 +63,7 @@ public class FindMD5InUserMetadata implements ContainsValueInListStrategy {
    protected final ObjectMD5 objectMD5;
    protected final ListBlobsInContainer getAllBlobMetadata;
    private final AtmosAsyncClient client;
-   private final ExecutorService userExecutor;
+   private final ListeningExecutorService userExecutor;
    /**
     * maximum duration of an blob Request
     */
@@ -74,7 +72,7 @@ public class FindMD5InUserMetadata implements ContainsValueInListStrategy {
    protected Long maxTime;
 
    @Inject
-   FindMD5InUserMetadata(@Named(Constants.PROPERTY_USER_THREADS) ExecutorService userExecutor, ObjectMD5 objectMD5,
+   FindMD5InUserMetadata(@Named(Constants.PROPERTY_USER_THREADS) ListeningExecutorService userExecutor, ObjectMD5 objectMD5,
          ListBlobsInContainer getAllBlobMetadata, AtmosAsyncClient client) {
       this.objectMD5 = objectMD5;
       this.getAllBlobMetadata = getAllBlobMetadata;
@@ -86,10 +84,9 @@ public class FindMD5InUserMetadata implements ContainsValueInListStrategy {
    public boolean execute(final String containerName, Object value, ListContainerOptions options) {
       final byte[] toSearch = objectMD5.apply(value);
       final BlockingQueue<Boolean> queue = new SynchronousQueue<Boolean>();
-      Map<String, Future<?>> responses = Maps.newHashMap();
+      Map<String, ListenableFuture<?>> responses = Maps.newHashMap();
       for (BlobMetadata md : getAllBlobMetadata.execute(containerName, options)) {
-         final ListenableFuture<AtmosObject> future = Futures.makeListenable(
-               client.headFile(containerName + "/" + md.getName()), userExecutor);
+         final ListenableFuture<AtmosObject> future = client.headFile(containerName + "/" + md.getName());
          future.addListener(new Runnable() {
             public void run() {
                try {

apis/atmos/src/main/java/org/jclouds/atmos/functions/ParseObjectFromHeadersAndHttpContent.java

@@ -79,7 +79,7 @@ public AtmosObject apply(HttpResponse from) {
    @Override
    public ParseObjectFromHeadersAndHttpContent setContext(HttpRequest request) {
       this.uri = request.getEndpoint();
-      return setPath(GeneratedHttpRequest.class.cast(request).getArgs().get(0).toString());
+      return setPath(GeneratedHttpRequest.class.cast(request).getInvocation().getArgs().get(0).toString());
    }
 
    private ParseObjectFromHeadersAndHttpContent setPath(String path) {

apis/atmos/src/main/java/org/jclouds/atmos/util/AtmosUtils.java

@@ -17,6 +17,8 @@
  * under the License.
  */
 package org.jclouds.atmos.util;
+import static com.google.common.base.Preconditions.checkState;
+import static org.jclouds.util.Predicates2.retry;
 
 import java.io.ByteArrayInputStream;
 import java.io.InputStream;
@@ -31,16 +33,16 @@
 import org.jclouds.atmos.filters.SignRequest;
 import org.jclouds.atmos.options.PutOptions;
 import org.jclouds.atmos.xml.ErrorHandler;
+import org.jclouds.blobstore.ContainerNotFoundException;
 import org.jclouds.blobstore.KeyAlreadyExistsException;
 import org.jclouds.blobstore.domain.Blob;
 import org.jclouds.crypto.Crypto;
 import org.jclouds.http.HttpCommand;
 import org.jclouds.http.HttpException;
 import org.jclouds.http.HttpResponse;
 import org.jclouds.http.functions.ParseSax;
-import org.jclouds.util.Assertions;
 
-import com.google.common.base.Supplier;
+import com.google.common.base.Predicate;
 
 /**
  * Encryption, Hashing, and IO Utilities needed to sign and verify Atmos Storage requests and
@@ -78,25 +80,23 @@ public static String putBlob(final AtmosClient sync, Crypto crypto, BlobToObject
          sync.createFile(container, object, options);
 
       } catch(KeyAlreadyExistsException e) {
-         deleteAndEnsureGone(sync, path);
+         deletePathAndEnsureGone(sync, path);
          sync.createFile(container, object, options);
       }
       return path;
    }
-
-   public static void deleteAndEnsureGone(final AtmosClient sync, final String path) {
-      try {
-         if (!Assertions.eventuallyTrue(new Supplier<Boolean>() {
-            public Boolean get() {
-               sync.deletePath(path);
-               return !sync.pathExists(path);
+
+   public static void deletePathAndEnsureGone(final AtmosClient sync, String path) {
+      checkState(retry(new Predicate<String>() {
+         public boolean apply(String in) {
+            try {
+               sync.deletePath(in);
+               return !sync.pathExists(in);
+            } catch (ContainerNotFoundException e) {
+               return true;
             }
-         }, 3000)) {
-            throw new IllegalStateException(path + " still exists after deleting!");
          }
-      } catch (InterruptedException e) {
-         throw new IllegalStateException(path + " interrupted during deletion!", e);
-      }
+      }, 3000).apply(path), "%s still exists after deleting!", path);
    }
 
    public AtmosError parseAtmosErrorFromContent(HttpCommand command, HttpResponse response, String content)