googleapis
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/RestartableInputStream.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/RestartableInputStream.java
@@ -0,0 +1,48 @@
+/*
+ * Copyright 2015 Google Inc. All Rights Reserved.
+ *
+ * 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 com.google.cloud.storage;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+
+/**
+ *
+ */
+public class RestartableInputStream {
+
+    InputStream inputStream;
+
+    public RestartableInputStream(InputStream inputStream) {
+        this.inputStream = inputStream;
+        this.inputStream.mark(Integer.MAX_VALUE);
+    }
+
+    public InputStream getInputStream() {
+        return this.inputStream;
+    }
+
+    public void restart() {
+        try {
+            this.inputStream.reset();
+            this.inputStream.mark(Integer.MAX_VALUE);
+        }
+        catch (IOException e) {
+            throw new IOExpection();
+        }
+    }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/RetryableInputStreamContent.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/RetryableInputStreamContent.java
@@ -0,0 +1,49 @@
+package com.google.cloud.storage;
+
+import com.google.api.client.http.AbstractInputStreamContent;
+import sun.nio.ch.IOUtil;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+
+public final class RetryableInputStreamContent extends AbstractInputStreamContent {
+
+    private final RestartableInputStream restartableInputStream;
+    private boolean retrySupported = true;
+
+    public RetryableInputStreamContent(String type, RestartableInputStream restartableInputStream) {
+        super(type);
+        this.restartableInputStream = restartableInputStream;
+    }
+
+    @Override
+    public InputStream getInputStream() throws IOException {
+        return restartableInputStream.getInputStream();
+    }
+
+    @Override
+    public long getLength() {
+        return this.length;
+    }
+
+    @Override
+    public boolean retrySupported() {
+        return this.retrySupported;
+    }
+
+    public RetryableInputStreamContent setCloseInputStream(boolean closeInputStream) {
+        super.setCloseInputStream(closeInputStream);
+        return this;
+    }
+
+    public RetryableInputStreamContent setRetrySupported(boolean retrySupported) {
+        this.retrySupported = retrySupported;
+        return this;
+    }
+
+    public RetryableInputStreamContent setLength(long length) {
+        this.length = length;
+        return this;
+    }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java
@@ -1493,6 +1493,8 @@ public static Builder newBuilder() {
   @Deprecated
   Blob create(BlobInfo blobInfo, InputStream content, BlobWriteOption... options);
 
+  Blob create(BlobInfo blobInfo, RestartableInputStream content, BlobWriteOption... options);
+
   /**
    * Returns the requested bucket or {@code null} if not found.
    *

diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java
@@ -147,6 +147,24 @@ public Blob create(BlobInfo blobInfo, InputStream content, BlobWriteOption... op
     return Blob.fromPb(this, storageRpc.create(blobPb, inputStreamParam, optionsMap));
   }
 
+  public Blob create(BlobInfo blobInfo, RestartableInputStream content, BlobWriteOption... options) {
+    Tuple<BlobInfo, BlobTargetOption[]> targetOptions = BlobTargetOption.convert(blobInfo, options);
+    StorageObject blobPb = targetOptions.x().toPb();
+    Map<StorageRpc.Option, ?> optionsMap = optionMap(targetOptions.x(), targetOptions.y());
+    RestartableInputStream inputStreamParam = firstNonNull(content
+            , new RestartableInputStream(new ByteArrayInputStream(EMPTY_BYTE_ARRAY)));
+    try {
+      return Blob.fromPb(this, runWithRetries(new Callable<StorageObject>() {
+        @Override
+        public StorageObject call() {
+          content.restart();
+          return storageRpc.create(blobPb, content, optionsMap);}
+      }, getOptions().getRetrySettings(), EXCEPTION_HANDLER, getOptions().getClock()));
+    } catch (RetryHelperException e) {
+      throw StorageException.translateAndThrow(e);
+    }
+  }
+
   private Blob internalCreate(BlobInfo info, final byte[] content, BlobTargetOption... options) {
     Preconditions.checkNotNull(content);
     final StorageObject blobPb = info.toPb();

diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/HttpStorageRpc.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/HttpStorageRpc.java
@@ -58,6 +58,8 @@
 import com.google.cloud.http.HttpTransportOptions;
 import com.google.cloud.storage.StorageException;
 import com.google.cloud.storage.StorageOptions;
+import com.google.cloud.storage.RestartableInputStream;
+import com.google.cloud.storage.RetryableInputStreamContent;
 import com.google.common.base.Function;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Iterables;
@@ -244,6 +246,29 @@ public StorageObject create(StorageObject storageObject, final InputStream conte
     }
   }
 
+  @Override
+  public StorageObject create(StorageObject storageObject, final RestartableInputStream content,
+                              Map<Option, ?> options) {
+    try {
+      Storage.Objects.Insert insert = storage.objects()
+              .insert(storageObject.getBucket(), storageObject,
+                      new RetryableInputStreamContent(storageObject.getContentType(), content));
+      insert.getMediaHttpUploader().setDirectUploadEnabled(true);
+      setEncryptionHeaders(insert.getRequestHeaders(), ENCRYPTION_KEY_PREFIX, options);
+      return insert.setProjection(DEFAULT_PROJECTION)
+              .setPredefinedAcl(Option.PREDEFINED_ACL.getString(options))
+              .setIfMetagenerationMatch(Option.IF_METAGENERATION_MATCH.getLong(options))
+              .setIfMetagenerationNotMatch(Option.IF_METAGENERATION_NOT_MATCH.getLong(options))
+              .setIfGenerationMatch(Option.IF_GENERATION_MATCH.getLong(options))
+              .setIfGenerationNotMatch(Option.IF_GENERATION_NOT_MATCH.getLong(options))
+              .setUserProject(Option.USER_PROJECT.getString(options))
+              .execute();
+    } catch (IOException ex) {
+      throw translate(ex);
+    }
+  }
+
+
   @Override
   public Tuple<String, Iterable<Bucket>> list(Map<Option, ?> options) {
     try {

diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/StorageRpc.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/StorageRpc.java
@@ -27,6 +27,7 @@
 import com.google.api.services.storage.model.TestIamPermissionsResponse;
 import com.google.cloud.ServiceRpc;
 import com.google.cloud.Tuple;
+import com.google.cloud.storage.RestartableInputStream;
 import com.google.cloud.storage.StorageException;
 import java.io.InputStream;
 import java.util.List;
@@ -186,6 +187,13 @@ public int hashCode() {
    */
   StorageObject create(StorageObject object, InputStream content, Map<Option, ?> options);
 
+    /**
+   * Creates a new storage object.
+   *
+   * @throws StorageException upon failure
+   */
+  StorageObject create(StorageObject object, RestartableInputStream content, Map<Option, ?> options);
+
   /**
    * Lists the project's buckets.
    *