added WriteTo interface for streaming puts

blobstore/src/main/java/org/jclouds/blobstore/TransientAsyncBlobStore.java

@@ -38,7 +38,6 @@ import static com.google.common.util.concurrent.Futures.immediateFuture;
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
-import java.io.InputStream;
 import java.io.ObjectInput;
 import java.io.ObjectInputStream;
 import java.io.ObjectOutput;
@@ -95,13 +94,13 @@ import com.google.common.base.Predicate;
 import com.google.common.base.Throwables;
 import com.google.common.collect.Iterables;
 import com.google.common.collect.Multimaps;
-import com.google.common.io.Closeables;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.inject.internal.Nullable;
 
 /**
- * Implementation of {@link BaseAsyncBlobStore} which keeps all data in a local Map object.
+ * Implementation of {@link BaseAsyncBlobStore} which keeps all data in a local
+ * Map object.
  * 
  * @author Adrian Cole
  * @author James Murty
@@ -119,8 +118,7 @@ public class TransientAsyncBlobStore extends BaseAsyncBlobStore {
    @Inject
    protected TransientAsyncBlobStore(BlobStoreContext context, DateService dateService, Crypto crypto,
          ConcurrentMap<String, ConcurrentMap<String, Blob>> containerToBlobs,
-            ConcurrentMap<String, Location> containerToLocation,
+         ConcurrentMap<String, Location> containerToLocation, HttpGetOptionsListToGetOptions httpGetOptionsConverter,
-            HttpGetOptionsListToGetOptions httpGetOptionsConverter,
          IfDirectoryReturnNameStrategy ifDirectoryReturnName, Blob.Factory blobFactory, BlobUtils blobUtils,
          @Named(Constants.PROPERTY_USER_THREADS) ExecutorService service, Location defaultLocation,
          Set<Location> locations) {
@@ -469,14 +467,11 @@ public class TransientAsyncBlobStore extends BaseAsyncBlobStore {
                .cast(DelegatingPayload.class.cast(object.getPayload()).getDelegate()) : null : null;
       try {
          if (payload == null || !(payload instanceof ByteArrayPayload)) {
-            InputStream input = object.getPayload().getInput();
-            try {
             String oldContentType = object.getPayload().getContentType();
-               payload = (ByteArrayPayload) Payloads.calculateMD5(Payloads.newPayload(object.getPayload().getInput()));
+            ByteArrayOutputStream out = new ByteArrayOutputStream();
+            object.getPayload().writeTo(out);
+            payload = (ByteArrayPayload) Payloads.calculateMD5(Payloads.newPayload(out.toByteArray()));
             payload.setContentType(oldContentType);
-            } finally {
-               Closeables.closeQuietly(input);
-            }
          } else {
             if (payload.getContentMD5() == null)
                Payloads.calculateMD5(object, crypto.md5());

blobstore/src/test/java/org/jclouds/blobstore/integration/internal/BaseBlobIntegrationTest.java

@@ -34,11 +34,13 @@ import java.io.File;
 import java.io.FileInputStream;
 import java.io.IOException;
 import java.io.InputStream;
+import java.io.OutputStream;
 import java.io.UnsupportedEncodingException;
 import java.security.NoSuchAlgorithmException;
 import java.security.cert.CertificateException;
 import java.util.Date;
 import java.util.Map;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 import java.util.zip.GZIPInputStream;
 
@@ -58,6 +60,8 @@ import org.jclouds.http.BaseJettyTest;
 import org.jclouds.http.HttpResponseException;
 import org.jclouds.io.InputSuppliers;
 import org.jclouds.io.Payloads;
+import org.jclouds.io.WriteTo;
+import org.jclouds.io.payloads.StreamingPayload;
 import org.jclouds.logging.Logger;
 import org.jclouds.util.Utils;
 import org.testng.ITestContext;
@@ -353,8 +357,8 @@ public class BaseBlobIntegrationTest extends BaseBlobStoreIntegrationTest {
 
    @DataProvider(name = "delete")
    public Object[][] createData() {
-      return new Object[][] { { "normal" }, { "sp ace" }, { "qu?stion" }, { "unic₪de" }, { "path/foo" }, { "colon:" },
+      return new Object[][] { { "normal" }, { "sp ace" }, { "qu?stion" }, { "unic₪de" }, { "path/foo" },
-               { "asteri*k" }, { "quote\"" }, { "{great<r}" }, { "lesst>en" }, { "p|pe" } };
+            { "colon:" }, { "asteri*k" }, { "quote\"" }, { "{great<r}" }, { "lesst>en" }, { "p|pe" } };
    }
 
    @Test(groups = { "integration", "live" }, dataProvider = "delete")
@@ -380,8 +384,8 @@ public class BaseBlobIntegrationTest extends BaseBlobStoreIntegrationTest {
 
             });
       assertEquals(Iterables.size(listing), 0, String.format(
-               "deleting %s, we still have %s blobs left in container %s, using encoding %s", key, Iterables
+            "deleting %s, we still have %s blobs left in container %s, using encoding %s", key,
-                        .size(listing), containerName, LOCAL_ENCODING));
+            Iterables.size(listing), containerName, LOCAL_ENCODING));
    }
 
    @Test(groups = { "integration", "live" })
@@ -427,6 +431,33 @@ public class BaseBlobIntegrationTest extends BaseBlobStoreIntegrationTest {
       }
    }
 
+   @Test(groups = { "integration", "live" })
+   public void testPutObjectStream() throws InterruptedException, IOException, ExecutionException {
+      Blob blob = context.getBlobStore().newBlob("streaming");
+      blob.setPayload(new StreamingPayload(new WriteTo() {
+         @Override
+         public void writeTo(OutputStream outstream) throws IOException {
+            outstream.write("foo".getBytes());
+         }
+      }));
+      blob.getMetadata().setContentType("text/csv");
+
+      String containerName = getContainerName();
+      try {
+
+         assertNotNull(context.getBlobStore().putBlob(containerName, blob));
+
+         blob = context.getBlobStore().getBlob(containerName, blob.getMetadata().getName());
+         String returnedString = getContentAsStringOrNullAndClose(blob);
+         assertEquals(returnedString, "foo");
+         assertEquals(blob.getPayload().getContentType(), "text/csv");
+         PageSet<? extends StorageMetadata> set = context.getBlobStore().list(containerName);
+         assert set.size() == 1 : set;
+      } finally {
+         returnContainer(containerName);
+      }
+   }
+
    protected volatile static Crypto crypto;
    static {
       try {

core/src/main/java/org/jclouds/io/Payload.java

@@ -19,9 +19,7 @@
 package org.jclouds.io;
 
 import java.io.Closeable;
-import java.io.IOException;
 import java.io.InputStream;
-import java.io.OutputStream;
 
 import javax.annotation.Nullable;
 
@@ -30,7 +28,7 @@ import com.google.common.io.InputSupplier;
 /**
  * @author Adrian Cole
  */
-public interface Payload extends InputSupplier<InputStream>, Closeable{
+public interface Payload extends InputSupplier<InputStream>, WriteTo, Closeable {
 
    /**
     * Creates a new InputStream object of the payload.
@@ -47,22 +45,16 @@ public interface Payload extends InputSupplier<InputStream>, Closeable{
     */
    boolean isRepeatable();
 
-   /**
-    * Writes the payload content to the output stream.
-    * 
-    * @throws IOException
-    */
-   void writeTo(OutputStream outstream) throws IOException;
-
    void setContentLength(@Nullable Long contentLength);
 
    /**
     * Returns the total size of the payload, or the chunk that's available.
     * <p/>
-    * Chunking is only used when {@link org.jclouds.http.GetOptions} is called with options like
+    * Chunking is only used when {@link org.jclouds.http.GetOptions} is called
-    * tail, range, or startAt.
+    * with options like tail, range, or startAt.
     * 
-    * @return the length in bytes that can be be obtained from {@link #getInput()}
+    * @return the length in bytes that can be be obtained from
+    *         {@link #getInput()}
     * @see javax.ws.rs.core.HttpHeaders#CONTENT_LENGTH
     * @see org.jclouds.http.options.GetOptions
     */
@@ -80,7 +72,8 @@ public interface Payload extends InputSupplier<InputStream>, Closeable{
    String getContentType();
 
    /**
-    * release resources used by this entity. This should be called when data is discarded.
+    * release resources used by this entity. This should be called when data is
+    * discarded.
     */
    void release();
 }

core/src/main/java/org/jclouds/io/WriteTo.java

@@ -0,0 +1,36 @@
+/**
+ *
+ * Copyright (C) 2010 Cloud Conscious, LLC. <info@cloudconscious.com>
+ *
+ * ====================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ====================================================================
+ */
+package org.jclouds.io;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+/**
+ * @author Adrian Cole
+ */
+public interface WriteTo {
+
+   /**
+    * Writes the payload content to the output stream.
+    * 
+    * @throws IOException
+    */
+   void writeTo(OutputStream outstream) throws IOException;
+
+}

core/src/main/java/org/jclouds/io/payloads/StreamingPayload.java

@@ -0,0 +1,174 @@
+/**
+ *
+ * Copyright (C) 2010 Cloud Conscious, LLC. <info@cloudconscious.com>
+ *
+ * ====================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ====================================================================
+ */
+package org.jclouds.io.payloads;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+
+import javax.annotation.Nullable;
+
+import org.jclouds.io.Payload;
+import org.jclouds.io.WriteTo;
+
+/**
+ * Note that not all services accept streaming payloads. For example, Rackspace
+ * CloudFiles accepts streaming while Amazon S3 does not.
+ * 
+ * @author Adrian Cole
+ */
+public class StreamingPayload implements Payload {
+   protected String contentType;
+   protected transient volatile boolean written;
+   protected final WriteTo writeTo;
+
+   public StreamingPayload(WriteTo writeTo) {
+      this.writeTo = checkNotNull(writeTo, "writeTo");
+      this.contentType = "application/unknown";
+   }
+
+   /**
+    * @throws UnsupportedOperationException
+    *            this payload is for streaming writes only
+    */
+   @Override
+   public Object getRawContent() {
+      throw new UnsupportedOperationException("this payload is for streaming writes only");
+   }
+
+   /**
+    * @throws UnsupportedOperationException
+    *            this payload is for streaming writes only
+    */
+   @Override
+   public InputStream getInput() {
+      throw new UnsupportedOperationException("this payload is for streaming writes only");
+   }
+
+   /**
+    * {@inheritDoc}
+    */
+   @Override
+   public Long getContentLength() {
+      return null;
+   }
+
+   /**
+    * {@inheritDoc}
+    */
+   @Override
+   public void setContentLength(@Nullable Long contentLength) {
+      throw new UnsupportedOperationException("this payload is for streaming writes only");
+   }
+
+   /**
+    * {@inheritDoc}
+    */
+   @Override
+   public byte[] getContentMD5() {
+      return null;
+   }
+
+   /**
+    * {@inheritDoc}
+    */
+   @Override
+   public void setContentMD5(byte[] md5) {
+      throw new UnsupportedOperationException("this payload is for streaming writes only");
+   }
+
+   /**
+    * {@inheritDoc}
+    */
+   @Override
+   public String getContentType() {
+      return contentType;
+   }
+
+   /**
+    * {@inheritDoc}
+    */
+   @Override
+   public void setContentType(@Nullable String contentType) {
+      this.contentType = contentType;
+   }
+
+   /**
+    * {@inheritDoc}
+    */
+   @Override
+   public void writeTo(OutputStream outstream) throws IOException {
+      writeTo.writeTo(outstream);
+   }
+
+   @Override
+   public String toString() {
+      return "[contentType=" + contentType + ", written=" + written + "]";
+   }
+
+   /**
+    * By default we are not repeatable.
+    */
+   @Override
+   public boolean isRepeatable() {
+      return true;
+   }
+
+   @Override
+   public int hashCode() {
+      final int prime = 31;
+      int result = 1;
+      result = prime * result + ((contentType == null) ? 0 : contentType.hashCode());
+      return result;
+   }
+
+   @Override
+   public boolean equals(Object obj) {
+      if (this == obj)
+         return true;
+      if (obj == null)
+         return false;
+      if (getClass() != obj.getClass())
+         return false;
+      StreamingPayload other = (StreamingPayload) obj;
+      if (contentType == null) {
+         if (other.contentType != null)
+            return false;
+      } else if (!contentType.equals(other.contentType))
+         return false;
+      return true;
+   }
+
+   /**
+    * By default there are no resources to release.
+    */
+   @Override
+   public void release() {
+   }
+
+   /**
+    * Delegates to release()
+    */
+   @Override
+   public void close() {
+      release();
+   }
+}

extensions/gae/src/main/java/org/jclouds/gae/ConvertToGaeRequest.java

@@ -20,9 +20,9 @@ package org.jclouds.gae;
 
 import static com.google.appengine.api.urlfetch.FetchOptions.Builder.disallowTruncate;
 import static com.google.common.base.Preconditions.checkNotNull;
-import static com.google.common.io.ByteStreams.toByteArray;
 import static com.google.common.io.Closeables.closeQuietly;
 
+import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.net.MalformedURLException;
@@ -51,8 +51,8 @@ public class ConvertToGaeRequest implements Function<HttpRequest, HTTPRequest> {
    public static final String USER_AGENT = "jclouds/1.0 urlfetch/1.3.5";
 
    /**
-    * byte [] content is replayable and the only content type supportable by GAE. As such, we
+    * byte [] content is replayable and the only content type supportable by
-    * convert the original request content to a byte array.
+    * GAE. As such, we convert the original request content to a byte array.
     */
    @Override
    public HTTPRequest apply(HttpRequest request) {
@@ -76,13 +76,15 @@ public class ConvertToGaeRequest implements Function<HttpRequest, HTTPRequest> {
       }
       gaeRequest.addHeader(new HTTPHeader(HttpHeaders.USER_AGENT, USER_AGENT));
       /**
-       * byte [] content is replayable and the only content type supportable by GAE. As such, we
+       * byte [] content is replayable and the only content type supportable by
-       * convert the original request content to a byte array.
+       * GAE. As such, we convert the original request content to a byte array.
        */
       if (request.getPayload() != null) {
          InputStream input = request.getPayload().getInput();
          try {
-            byte[] array = toByteArray(input);
+            ByteArrayOutputStream out = new ByteArrayOutputStream();
+            request.getPayload().writeTo(out);
+            byte[] array = out.toByteArray();
             if (!request.getPayload().isRepeatable()) {
                Payload oldPayload = request.getPayload();
                request.setPayload(array);

rackspace/src/test/resources/log4j.xml

@@ -161,13 +161,13 @@
     <category name="jclouds.ssh">
         <priority value="DEBUG" />
         <appender-ref ref="ASYNCSSH" />
-    </category>
+    </category><!--
 
     <category name="jclouds.wire">
         <priority value="DEBUG" />
         <appender-ref ref="ASYNCWIRE" />
     </category>
-    <category name="jclouds.blobstore">
+    --><category name="jclouds.blobstore">
         <priority value="DEBUG" />
         <appender-ref ref="ASYNCBLOBSTORE" />
     </category>