diff --git a/gcloud-java-storage/src/main/java/com/google/gcloud/spi/DefaultStorageRpc.java b/gcloud-java-storage/src/main/java/com/google/gcloud/spi/DefaultStorageRpc.java
index 7c2d3a7be21a..a3e79333085d 100644
--- a/gcloud-java-storage/src/main/java/com/google/gcloud/spi/DefaultStorageRpc.java
+++ b/gcloud-java-storage/src/main/java/com/google/gcloud/spi/DefaultStorageRpc.java
@@ -43,6 +43,7 @@
 import com.google.api.client.http.HttpResponse;
 import com.google.api.client.http.HttpResponseException;
 import com.google.api.client.http.HttpTransport;
+import com.google.api.client.http.InputStreamContent;
 import com.google.api.client.http.json.JsonHttpContent;
 import com.google.api.client.json.JsonFactory;
 import com.google.api.client.json.jackson.JacksonFactory;
@@ -63,6 +64,7 @@
 
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
+import java.io.InputStream;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
@@ -119,13 +121,14 @@ public Bucket create(Bucket bucket, Map<Option, ?> options) throws StorageExcept
   }
 
   @Override
-  public StorageObject create(StorageObject storageObject, final byte[] content,
+  public StorageObject create(StorageObject storageObject, final InputStream content,
       Map<Option, ?> options) throws StorageException {
     try {
-      return storage.objects()
+      Storage.Objects.Insert insert = storage.objects()
           .insert(storageObject.getBucket(), storageObject,
-              new ByteArrayContent(storageObject.getContentType(), content))
-          .setProjection(DEFAULT_PROJECTION)
+              new InputStreamContent(storageObject.getContentType(), content));
+      insert.getMediaHttpUploader().setDirectUploadEnabled(true);
+      return insert.setProjection(DEFAULT_PROJECTION)
           .setPredefinedAcl(PREDEFINED_ACL.getString(options))
           .setIfMetagenerationMatch(IF_METAGENERATION_MATCH.getLong(options))
           .setIfMetagenerationNotMatch(IF_METAGENERATION_NOT_MATCH.getLong(options))
@@ -521,4 +524,3 @@ public String open(StorageObject object, Map<Option, ?> options)
     }
   }
 }
-
diff --git a/gcloud-java-storage/src/main/java/com/google/gcloud/spi/StorageRpc.java b/gcloud-java-storage/src/main/java/com/google/gcloud/spi/StorageRpc.java
index 7448fec9d32f..423929adff08 100644
--- a/gcloud-java-storage/src/main/java/com/google/gcloud/spi/StorageRpc.java
+++ b/gcloud-java-storage/src/main/java/com/google/gcloud/spi/StorageRpc.java
@@ -22,6 +22,7 @@
 import com.google.common.collect.ImmutableMap;
 import com.google.gcloud.storage.StorageException;
 
+import java.io.InputStream;
 import java.util.List;
 import java.util.Map;
 
@@ -128,7 +129,7 @@ public BatchResponse(Map<StorageObject, Tuple<Boolean, StorageException>> delete
 
   Bucket create(Bucket bucket, Map<Option, ?> options) throws StorageException;
 
-  StorageObject create(StorageObject object, byte[] content, Map<Option, ?> options)
+  StorageObject create(StorageObject object, InputStream content, Map<Option, ?> options)
       throws StorageException;
 
   Tuple<String, Iterable<Bucket>> list(Map<Option, ?> options) throws StorageException;
diff --git a/gcloud-java-storage/src/main/java/com/google/gcloud/storage/Storage.java b/gcloud-java-storage/src/main/java/com/google/gcloud/storage/Storage.java
index 590c8d9d74d4..2ce25a8190f3 100644
--- a/gcloud-java-storage/src/main/java/com/google/gcloud/storage/Storage.java
+++ b/gcloud-java-storage/src/main/java/com/google/gcloud/storage/Storage.java
@@ -25,6 +25,7 @@
 import com.google.gcloud.Service;
 import com.google.gcloud.spi.StorageRpc;
 
+import java.io.InputStream;
 import java.io.Serializable;
 import java.net.URL;
 import java.util.Arrays;
@@ -493,13 +494,31 @@ public static Builder builder() {
   BucketInfo create(BucketInfo bucketInfo, BucketTargetOption... options);
 
   /**
-   * Create a new blob.
+   * Create a new blob with no content.
+   *
+   * @return a complete blob information.
+   * @throws StorageException upon failure
+   */
+  BlobInfo create(BlobInfo blobInfo, BlobTargetOption... options);
+
+  /**
+   * Create a new blob. Direct upload is used to upload {@code content}. For large content,
+   * {@link #writer} is recommended as it uses resumable upload.
    *
    * @return a complete blob information.
    * @throws StorageException upon failure
    */
   BlobInfo create(BlobInfo blobInfo, byte[] content, BlobTargetOption... options);
 
+  /**
+   * Create a new blob. Direct upload is used to upload {@code content}. For large content,
+   * {@link #writer} is recommended as it uses resumable upload.
+   *
+   * @return a complete blob information.
+   * @throws StorageException upon failure
+   */
+  BlobInfo create(BlobInfo blobInfo, InputStream content, BlobTargetOption... options);
+
   /**
    * Return the requested bucket or {@code null} if not found.
    *
diff --git a/gcloud-java-storage/src/main/java/com/google/gcloud/storage/StorageImpl.java b/gcloud-java-storage/src/main/java/com/google/gcloud/storage/StorageImpl.java
index 4c80c6741559..d7dbafdc8b9e 100644
--- a/gcloud-java-storage/src/main/java/com/google/gcloud/storage/StorageImpl.java
+++ b/gcloud-java-storage/src/main/java/com/google/gcloud/storage/StorageImpl.java
@@ -49,6 +49,8 @@
 import com.google.gcloud.spi.StorageRpc;
 import com.google.gcloud.spi.StorageRpc.Tuple;
 
+import java.io.ByteArrayInputStream;
+import java.io.InputStream;
 import java.io.Serializable;
 import java.io.UnsupportedEncodingException;
 import java.net.MalformedURLException;
@@ -112,14 +114,27 @@ public com.google.api.services.storage.model.Bucket call() {
         }, options().retryParams(), EXCEPTION_HANDLER));
   }
 
+  @Override
+  public BlobInfo create(BlobInfo blobInfo, BlobTargetOption... options) {
+    return create(blobInfo, new ByteArrayInputStream(EMPTY_BYTE_ARRAY), options);
+  }
+
   @Override
   public BlobInfo create(BlobInfo blobInfo, final byte[] content, BlobTargetOption... options) {
+    return create(blobInfo,
+        new ByteArrayInputStream(firstNonNull(content, EMPTY_BYTE_ARRAY)), options);
+  }
+
+  @Override
+  public BlobInfo create(BlobInfo blobInfo, final InputStream content,
+      BlobTargetOption... options) {
     final StorageObject blobPb = blobInfo.toPb();
     final Map<StorageRpc.Option, ?> optionsMap = optionMap(blobInfo, options);
     return BlobInfo.fromPb(runWithRetries(new Callable<StorageObject>() {
       @Override
       public StorageObject call() {
-        return storageRpc.create(blobPb, firstNonNull(content, EMPTY_BYTE_ARRAY), optionsMap);
+        return storageRpc.create(blobPb,
+            firstNonNull(content, new ByteArrayInputStream(EMPTY_BYTE_ARRAY)), optionsMap);
       }
     }, options().retryParams(), EXCEPTION_HANDLER));
   }