diff --git a/README.md b/README.md
index 2887f5a1a9a1..1b9867fd198f 100644
--- a/README.md
+++ b/README.md
@@ -61,14 +61,13 @@ Here is a code snippet showing a simple usage example from within Compute/App En
 
 ```java
 import com.google.gcloud.datastore.Datastore;
-import com.google.gcloud.datastore.DatastoreFactory;
 import com.google.gcloud.datastore.DatastoreOptions;
 import com.google.gcloud.datastore.DateTime;
 import com.google.gcloud.datastore.Entity;
 import com.google.gcloud.datastore.Key;
 import com.google.gcloud.datastore.KeyFactory;
 
-Datastore datastore = DatastoreFactory.instance().get(DatastoreOptions.getDefaultInstance());
+Datastore datastore = DatastoreOptions.getDefaultInstance().service();
 KeyFactory keyFactory = datastore.newKeyFactory().kind(KIND);
 Key key = keyFactory.newKey(keyName);
 Entity entity = datastore.get(key);
@@ -106,14 +105,13 @@ import static java.nio.charset.StandardCharsets.UTF_8;
 import com.google.gcloud.storage.Blob;
 import com.google.gcloud.storage.BlobId;
 import com.google.gcloud.storage.Storage;
-import com.google.gcloud.storage.StorageFactory;
 import com.google.gcloud.storage.StorageOptions;
 
 import java.nio.ByteBuffer;
 import java.nio.channels.WritableByteChannel;
 
 StorageOptions options = StorageOptions.builder().projectId("project").build();
-Storage storage = StorageFactory.instance().get(options);
+Storage storage = options.service();
 BlobId blobId = BlobId.of("bucket", "blob_name");
 Blob blob = Blob.load(storage, blobId);
 if (blob == null) {
diff --git a/TESTING.md b/TESTING.md
index 158d71be2e91..02a3d14ab0bf 100644
--- a/TESTING.md
+++ b/TESTING.md
@@ -18,7 +18,7 @@ You can test against a temporary local datastore by following these steps:
       .projectId(PROJECT_ID)
       .host("http://localhost:8080")
       .build();
-  Datastore localDatastore = DatastoreFactory.instance().get(options);
+  Datastore localDatastore = options.service();
   ```
 3. Run your tests.
 
@@ -35,7 +35,7 @@ You can test against a remote datastore emulator as well.  To do this, set the `
       .projectId(PROJECT_ID)
       .host("http://<hostname of machine>:<port>")
       .build();
-  Datastore localDatastore = DatastoreFactory.instance().get(options);
+  Datastore localDatastore = options.service();
   ```
 
 Note that the remote datastore must be running before your tests are run.
@@ -52,7 +52,7 @@ Currently, there isn't an emulator for Google Cloud Storage, so an alternative i
 Here is an example that uses the `RemoteGcsHelper` to create a bucket.
   ```java
   RemoteGcsHelper gcsHelper = RemoteGcsHelper.create(PROJECT_ID, "/path/to/my/JSON/key.json");
-  Storage storage = StorageFactory.instance().get(gcsHelper.options());
+  Storage storage = gcsHelper.options().service();
   String bucket = RemoteGcsHelper.generateBucketName();
   storage.create(BucketInfo.of(bucket));
   ```
diff --git a/gcloud-java-core/src/main/java/com/google/gcloud/AuthCredentials.java b/gcloud-java-core/src/main/java/com/google/gcloud/AuthCredentials.java
index ffc54df77a90..73c66279ea53 100644
--- a/gcloud-java-core/src/main/java/com/google/gcloud/AuthCredentials.java
+++ b/gcloud-java-core/src/main/java/com/google/gcloud/AuthCredentials.java
@@ -31,8 +31,6 @@
 
 import java.io.IOException;
 import java.io.InputStream;
-import java.io.ObjectInputStream;
-import java.io.ObjectStreamException;
 import java.io.Serializable;
 import java.security.GeneralSecurityException;
 import java.security.PrivateKey;
@@ -42,15 +40,34 @@
 /**
  * Credentials for accessing Google Cloud services.
  */
-public abstract class AuthCredentials implements Serializable {
-
-  private static final long serialVersionUID = 236297804453464604L;
+public abstract class AuthCredentials implements Restorable<AuthCredentials> {
 
   private static class AppEngineAuthCredentials extends AuthCredentials {
 
-    private static final long serialVersionUID = 7931300552744202954L;
-
     private static final AuthCredentials INSTANCE = new AppEngineAuthCredentials();
+    private static final AppEngineAuthCredentialsState STATE =
+        new AppEngineAuthCredentialsState();
+
+    private static class AppEngineAuthCredentialsState
+        implements RestorableState<AuthCredentials>, Serializable {
+
+      private static final long serialVersionUID = 3558563960848658928L;
+
+      @Override
+      public AuthCredentials restore() {
+        return INSTANCE;
+      }
+
+      @Override
+      public int hashCode() {
+        return getClass().getName().hashCode();
+      }
+
+      @Override
+      public boolean equals(Object obj) {
+        return obj instanceof AppEngineAuthCredentialsState;
+      }
+    }
 
     @Override
     protected HttpRequestInitializer httpRequestInitializer(HttpTransport transport,
@@ -58,19 +75,56 @@ protected HttpRequestInitializer httpRequestInitializer(HttpTransport transport,
       return new AppIdentityCredential(scopes);
     }
 
-    private Object readResolve() throws ObjectStreamException {
-      return INSTANCE;
+    @Override
+    public RestorableState<AuthCredentials> capture() {
+      return STATE;
     }
   }
 
   public static class ServiceAccountAuthCredentials extends AuthCredentials {
 
-    private static final long serialVersionUID = 8007708734318445901L;
     private final String account;
     private final PrivateKey privateKey;
 
     private static final AuthCredentials NO_CREDENTIALS = new ServiceAccountAuthCredentials();
 
+    private static class ServiceAccountAuthCredentialsState
+        implements RestorableState<AuthCredentials>, Serializable {
+
+      private static final long serialVersionUID = -7302180782414633639L;
+
+      private final String account;
+      private final PrivateKey privateKey;
+
+      private ServiceAccountAuthCredentialsState(String account, PrivateKey privateKey) {
+        this.account = account;
+        this.privateKey = privateKey;
+      }
+
+      @Override
+      public AuthCredentials restore() {
+        if (account == null && privateKey == null) {
+          return NO_CREDENTIALS;
+        }
+        return new ServiceAccountAuthCredentials(account, privateKey);
+      }
+
+      @Override
+      public int hashCode() {
+        return Objects.hash(account, privateKey);
+      }
+
+      @Override
+      public boolean equals(Object obj) {
+        if (!(obj instanceof ServiceAccountAuthCredentialsState)) {
+          return false;
+        }
+        ServiceAccountAuthCredentialsState other = (ServiceAccountAuthCredentialsState) obj;
+        return Objects.equals(account, other.account)
+            && Objects.equals(privateKey, other.privateKey);
+      }
+    }
+
     ServiceAccountAuthCredentials(String account, PrivateKey privateKey) {
       this.account = checkNotNull(account);
       this.privateKey = checkNotNull(privateKey);
@@ -104,59 +158,94 @@ public PrivateKey privateKey() {
     }
 
     @Override
-    public int hashCode() {
-      return Objects.hash(account, privateKey);
-    }
-
-    @Override
-    public boolean equals(Object obj) {
-      if (!(obj instanceof ServiceAccountAuthCredentials)) {
-        return false;
-      }
-      ServiceAccountAuthCredentials other = (ServiceAccountAuthCredentials) obj;
-      return Objects.equals(account, other.account)
-          && Objects.equals(privateKey, other.privateKey);
+    public RestorableState<AuthCredentials> capture() {
+      return new ServiceAccountAuthCredentialsState(account, privateKey);
     }
   }
 
   private static class ComputeEngineAuthCredentials extends AuthCredentials {
 
-    private static final long serialVersionUID = -5217355402127260144L;
+    private ComputeCredential computeCredential;
 
-    private transient ComputeCredential computeCredential;
+    private static final ComputeEngineAuthCredentialsState STATE =
+        new ComputeEngineAuthCredentialsState();
 
-    ComputeEngineAuthCredentials() throws IOException, GeneralSecurityException {
-      computeCredential = getComputeCredential();
-    }
+    private static class ComputeEngineAuthCredentialsState
+        implements RestorableState<AuthCredentials>, Serializable {
+
+      private static final long serialVersionUID = -6168594072854417404L;
+
+      @Override
+      public AuthCredentials restore() {
+        try {
+          return new ComputeEngineAuthCredentials();
+        } catch (IOException | GeneralSecurityException e) {
+          throw new IllegalStateException(
+              "Could not restore " + ComputeEngineAuthCredentials.class.getSimpleName(), e);
+        }
+      }
+
+      @Override
+      public int hashCode() {
+        return getClass().getName().hashCode();
+      }
 
-    private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
-      in.defaultReadObject();
-      try {
-        computeCredential = getComputeCredential();
-      } catch (GeneralSecurityException e) {
-        throw new IOException(e);
+      @Override
+      public boolean equals(Object obj) {
+        return obj instanceof ComputeEngineAuthCredentialsState;
       }
     }
 
+    ComputeEngineAuthCredentials() throws IOException, GeneralSecurityException {
+      computeCredential = getComputeCredential();
+    }
+
     @Override
     protected HttpRequestInitializer httpRequestInitializer(HttpTransport transport,
         Set<String> scopes) {
       return computeCredential;
     }
+
+    @Override
+    public RestorableState<AuthCredentials> capture() {
+      return STATE;
+    }
   }
 
   private static class ApplicationDefaultAuthCredentials extends AuthCredentials {
 
-    private static final long serialVersionUID = -8306873864136099893L;
+    private GoogleCredentials googleCredentials;
 
-    private transient GoogleCredentials googleCredentials;
+    private static final ApplicationDefaultAuthCredentialsState STATE =
+        new ApplicationDefaultAuthCredentialsState();
 
-    ApplicationDefaultAuthCredentials() throws IOException {
-      googleCredentials = GoogleCredentials.getApplicationDefault();
+    private static class ApplicationDefaultAuthCredentialsState
+        implements RestorableState<AuthCredentials>, Serializable {
+
+      private static final long serialVersionUID = -8839085552021212257L;
+
+      @Override
+      public AuthCredentials restore() {
+        try {
+          return new ApplicationDefaultAuthCredentials();
+        } catch (IOException e) {
+          throw new IllegalStateException(
+              "Could not restore " + ApplicationDefaultAuthCredentials.class.getSimpleName(), e);
+        }
+      }
+
+      @Override
+      public int hashCode() {
+        return getClass().getName().hashCode();
+      }
+
+      @Override
+      public boolean equals(Object obj) {
+        return obj instanceof ApplicationDefaultAuthCredentialsState;
+      }
     }
 
-    private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
-      in.defaultReadObject();
+    ApplicationDefaultAuthCredentials() throws IOException {
       googleCredentials = GoogleCredentials.getApplicationDefault();
     }
 
@@ -165,6 +254,11 @@ protected HttpRequestInitializer httpRequestInitializer(HttpTransport transport,
         Set<String> scopes) {
       return new HttpCredentialsAdapter(googleCredentials);
     }
+
+    @Override
+    public RestorableState<AuthCredentials> capture() {
+      return STATE;
+    }
   }
 
   protected abstract HttpRequestInitializer httpRequestInitializer(HttpTransport transport,
diff --git a/gcloud-java-core/src/main/java/com/google/gcloud/BaseService.java b/gcloud-java-core/src/main/java/com/google/gcloud/BaseService.java
index 982d3058295c..7600d25411fd 100644
--- a/gcloud-java-core/src/main/java/com/google/gcloud/BaseService.java
+++ b/gcloud-java-core/src/main/java/com/google/gcloud/BaseService.java
@@ -16,7 +16,7 @@
 
 package com.google.gcloud;
 
-public abstract class BaseService<OptionsT extends ServiceOptions<?, OptionsT>>
+public abstract class BaseService<OptionsT extends ServiceOptions<?, ?, OptionsT>>
     implements Service<OptionsT> {
 
   private final OptionsT options;
diff --git a/gcloud-java-core/src/main/java/com/google/gcloud/Restorable.java b/gcloud-java-core/src/main/java/com/google/gcloud/Restorable.java
new file mode 100644
index 000000000000..51391e33bd7d
--- /dev/null
+++ b/gcloud-java-core/src/main/java/com/google/gcloud/Restorable.java
@@ -0,0 +1,46 @@
+/*
+ * 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.gcloud;
+
+/**
+ * Implementation of this interface can persist their state and restore from it.
+ *
+ * <p>
+ * A typical capture usage:
+ * <pre>  {@code
+ * X restorableObj; // X instanceof Restorable<X>
+ * RestorableState<X> state = restorableObj.capture();
+ * .. persist state
+ * }</pre>
+ *
+ * A typical restore usage:
+ * <pre>  {@code
+ * RestorableState<X> state = ... // read from persistence
+ * X restorableObj = state.restore();
+ * ...
+ * }</pre>
+ */
+public interface Restorable<T extends Restorable<T>> {
+
+  /**
+   * Captures the state of this object.
+   *
+   * @return a {@link RestorableState} instance that contains the state for this object and can
+   *     restore it afterwards.
+   */
+  RestorableState<T> capture();
+}
diff --git a/gcloud-java-core/src/main/java/com/google/gcloud/RestorableState.java b/gcloud-java-core/src/main/java/com/google/gcloud/RestorableState.java
index 9cd3ee5c3c4c..0c60411cb285 100644
--- a/gcloud-java-core/src/main/java/com/google/gcloud/RestorableState.java
+++ b/gcloud-java-core/src/main/java/com/google/gcloud/RestorableState.java
@@ -23,7 +23,7 @@
  * Implementations of this class must implement {@link java.io.Serializable} to ensure that the
  * state of a the object can be correctly serialized.
  */
-public interface RestorableState<T> {
+public interface RestorableState<T extends Restorable<T>> {
 
   /**
    * Returns an object whose internal state reflects the one saved in the invocation object.
diff --git a/gcloud-java-core/src/main/java/com/google/gcloud/Service.java b/gcloud-java-core/src/main/java/com/google/gcloud/Service.java
index 19759fb20e21..2748c55058b4 100644
--- a/gcloud-java-core/src/main/java/com/google/gcloud/Service.java
+++ b/gcloud-java-core/src/main/java/com/google/gcloud/Service.java
@@ -16,6 +16,6 @@
 
 package com.google.gcloud;
 
-public interface Service<OptionsT extends ServiceOptions<?, OptionsT>> {
+public interface Service<OptionsT extends ServiceOptions<?, ?, OptionsT>> {
   OptionsT options();
 }
diff --git a/gcloud-java-core/src/main/java/com/google/gcloud/ServiceFactory.java b/gcloud-java-core/src/main/java/com/google/gcloud/ServiceFactory.java
new file mode 100644
index 000000000000..b59fc1e9a10e
--- /dev/null
+++ b/gcloud-java-core/src/main/java/com/google/gcloud/ServiceFactory.java
@@ -0,0 +1,28 @@
+/*
+ * 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.gcloud;
+
+/**
+ * A base interface for all service factories.
+ *
+ * Implementation must provide a public no-arg constructor.
+ * Loading of a factory implementation is done via {@link java.util.ServiceLoader}.
+ */
+public interface ServiceFactory<ServiceT extends Service, ServiceOptionsT extends ServiceOptions> {
+
+  ServiceT create(ServiceOptionsT serviceOptions);
+}
diff --git a/gcloud-java-core/src/main/java/com/google/gcloud/ServiceOptions.java b/gcloud-java-core/src/main/java/com/google/gcloud/ServiceOptions.java
index b1510ae6a9fd..f22c4d51e371 100644
--- a/gcloud-java-core/src/main/java/com/google/gcloud/ServiceOptions.java
+++ b/gcloud-java-core/src/main/java/com/google/gcloud/ServiceOptions.java
@@ -34,6 +34,7 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.InputStreamReader;
+import java.io.ObjectInputStream;
 import java.io.ObjectStreamException;
 import java.io.Serializable;
 import java.lang.reflect.Method;
@@ -47,8 +48,9 @@
 import java.util.regex.Pattern;
 
 public abstract class ServiceOptions<
+        ServiceT extends Service,
         ServiceRpcT,
-        OptionsT extends ServiceOptions<ServiceRpcT, OptionsT>>
+        OptionsT extends ServiceOptions<ServiceT, ServiceRpcT, OptionsT>>
     implements Serializable {
 
   private static final String DEFAULT_HOST = "https://www.googleapis.com";
@@ -57,21 +59,35 @@ public abstract class ServiceOptions<
 
   private final String projectId;
   private final String host;
-  private final HttpTransportFactory httpTransportFactory;
-  private final AuthCredentials authCredentials;
+  private final String httpTransportFactoryClassName;
+  private final RestorableState<AuthCredentials> authCredentialsState;
   private final RetryParams retryParams;
-  private final ServiceRpcFactory<ServiceRpcT, OptionsT> serviceRpcFactory;
+  private final String serviceRpcFactoryClassName;
+  private final String serviceFactoryClassName;
   private final int connectTimeout;
   private final int readTimeout;
   private final Clock clock;
 
-  public interface HttpTransportFactory extends Serializable {
+  private transient HttpTransportFactory httpTransportFactory;
+  private transient AuthCredentials authCredentials;
+  private transient ServiceRpcFactory<ServiceRpcT, OptionsT> serviceRpcFactory;
+  private transient ServiceFactory<ServiceT, OptionsT> serviceFactory;
+  private transient ServiceT service;
+  private transient ServiceRpcT rpc;
+
+  /**
+   * A base interface for all {@link HttpTransport} factories.
+   *
+   * Implementation must provide a public no-arg constructor. Loading of a factory implementation is
+   * done via {@link java.util.ServiceLoader}.
+   */
+  public interface HttpTransportFactory {
     HttpTransport create();
   }
 
-  private enum DefaultHttpTransportFactory implements HttpTransportFactory {
+  public static class DefaultHttpTransportFactory implements HttpTransportFactory {
 
-    INSTANCE;
+    private static final HttpTransportFactory INSTANCE = new DefaultHttpTransportFactory();
 
     @Override
     public HttpTransport create() {
@@ -133,15 +149,17 @@ private Object readResolve() throws ObjectStreamException {
   }
 
   protected abstract static class Builder<
+      ServiceT extends Service,
       ServiceRpcT,
-      OptionsT extends ServiceOptions<ServiceRpcT, OptionsT>,
-      B extends Builder<ServiceRpcT, OptionsT, B>> {
+      OptionsT extends ServiceOptions<ServiceT, ServiceRpcT, OptionsT>,
+      B extends Builder<ServiceT, ServiceRpcT, OptionsT, B>> {
 
     private String projectId;
     private String host;
     private HttpTransportFactory httpTransportFactory;
     private AuthCredentials authCredentials;
     private RetryParams retryParams;
+    private ServiceFactory<ServiceT, OptionsT> serviceFactory;
     private ServiceRpcFactory<ServiceRpcT, OptionsT> serviceRpcFactory;
     private int connectTimeout = -1;
     private int readTimeout = -1;
@@ -149,22 +167,34 @@ protected abstract static class Builder<
 
     protected Builder() {}
 
-    protected Builder(ServiceOptions<ServiceRpcT, OptionsT> options) {
+    protected Builder(ServiceOptions<ServiceT, ServiceRpcT, OptionsT> options) {
       projectId = options.projectId;
       host = options.host;
       httpTransportFactory = options.httpTransportFactory;
       authCredentials = options.authCredentials;
       retryParams = options.retryParams;
+      serviceFactory = options.serviceFactory;
       serviceRpcFactory = options.serviceRpcFactory;
+      connectTimeout = options.connectTimeout;
+      readTimeout = options.readTimeout;
+      clock = options.clock;
     }
 
-    protected abstract ServiceOptions<ServiceRpcT, OptionsT> build();
+    protected abstract ServiceOptions<ServiceT, ServiceRpcT, OptionsT> build();
 
     @SuppressWarnings("unchecked")
     protected B self() {
       return (B) this;
     }
 
+    /**
+     * Sets the service factory.
+     */
+    public B serviceFactory(ServiceFactory<ServiceT, OptionsT> serviceFactory) {
+      this.serviceFactory = serviceFactory;
+      return self();
+    }
+
     /**
      * Sets the service's clock. The clock is mainly used for testing purpose. {@link Clock} will be
      * replaced by Java8's {@code java.time.Clock}.
@@ -242,7 +272,7 @@ public B serviceRpcFactory(ServiceRpcFactory<ServiceRpcT, OptionsT> serviceRpcFa
      * Sets the timeout in milliseconds to establish a connection.
      *
      * @param connectTimeout connection timeout in milliseconds. 0 for an infinite timeout, a
-     * negative number for the default value (20000).
+     *        negative number for the default value (20000).
      * @return the builder.
      */
     public B connectTimeout(int connectTimeout) {
@@ -253,8 +283,8 @@ public B connectTimeout(int connectTimeout) {
     /**
      * Sets the timeout in milliseconds to read data from an established connection.
      *
-     * @param readTimeout read timeout in milliseconds. 0 for an infinite timeout, a
-     * negative number for the default value (20000).
+     * @param readTimeout read timeout in milliseconds. 0 for an infinite timeout, a negative number
+     *        for the default value (20000).
      * @return the builder.
      */
     public B readTimeout(int readTimeout) {
@@ -263,14 +293,23 @@ public B readTimeout(int readTimeout) {
     }
   }
 
-  protected ServiceOptions(Builder<ServiceRpcT, OptionsT, ?> builder) {
+  protected ServiceOptions(Class<? extends ServiceFactory<ServiceT, OptionsT>> serviceFactoryClass,
+      Class<? extends ServiceRpcFactory<ServiceRpcT, OptionsT>> rpcFactoryClass,
+      Builder<ServiceT, ServiceRpcT, OptionsT, ?> builder) {
     projectId = checkNotNull(builder.projectId != null ? builder.projectId : defaultProject());
     host = firstNonNull(builder.host, defaultHost());
-    httpTransportFactory =
-        firstNonNull(builder.httpTransportFactory, DefaultHttpTransportFactory.INSTANCE);
+    httpTransportFactory = firstNonNull(builder.httpTransportFactory,
+        getFromServiceLoader(HttpTransportFactory.class, DefaultHttpTransportFactory.INSTANCE));
+    httpTransportFactoryClassName = httpTransportFactory.getClass().getName();
     authCredentials = firstNonNull(builder.authCredentials, defaultAuthCredentials());
+    authCredentialsState = authCredentials.capture();
     retryParams = builder.retryParams;
-    serviceRpcFactory = builder.serviceRpcFactory;
+    serviceFactory = firstNonNull(builder.serviceFactory,
+        getFromServiceLoader(serviceFactoryClass, defaultServiceFactory()));
+    serviceFactoryClassName = serviceFactory.getClass().getName();
+    serviceRpcFactory = firstNonNull(builder.serviceRpcFactory,
+        getFromServiceLoader(rpcFactoryClass, defaultRpcFactory()));
+    serviceRpcFactoryClassName = serviceRpcFactory.getClass().getName();
     connectTimeout = builder.connectTimeout;
     readTimeout = builder.readTimeout;
     clock = firstNonNull(builder.clock, Clock.defaultClock());
@@ -331,10 +370,10 @@ protected static String googleCloudProjectId() {
     } catch (IOException ignore) {
       // ignore
     }
-   File configDir;
+    File configDir;
     if (System.getenv().containsKey("CLOUDSDK_CONFIG")) {
       configDir = new File(System.getenv("CLOUDSDK_CONFIG"));
-    } else if (isWindows() &&  System.getenv().containsKey("APPDATA")) {
+    } else if (isWindows() && System.getenv().containsKey("APPDATA")) {
       configDir = new File(System.getenv("APPDATA"), "gcloud");
     } else {
       configDir = new File(System.getProperty("user.home"), ".config/gcloud");
@@ -389,10 +428,22 @@ protected static String getAppEngineProjectId() {
     }
   }
 
-  protected abstract Set<String> scopes();
+  public ServiceT service() {
+    if (service == null) {
+      service = serviceFactory.create((OptionsT) this);
+    }
+    return service;
+  }
+
+  public ServiceRpcT rpc() {
+    if (rpc == null) {
+      rpc = serviceRpcFactory.create((OptionsT) this);
+    }
+    return rpc;
+  }
 
   /**
-   * Returns the project id. 
+   * Returns the project id.
    */
   public String projectId() {
     return projectId;
@@ -427,13 +478,6 @@ public RetryParams retryParams() {
     return retryParams != null ? retryParams : RetryParams.noRetries();
   }
 
-  /**
-   * Returns the factory for rpc services.
-   */
-  public ServiceRpcFactory<ServiceRpcT, OptionsT> serviceRpcFactory() {
-    return serviceRpcFactory;
-  }
-
   /**
    * Returns a request initializer responsible for initializing requests according to service
    * options.
@@ -473,41 +517,57 @@ public int readTimeout() {
   }
 
   /**
-   * Returns the service's clock. Default time source uses {@link System#currentTimeMillis()} to
-   * get current time. 
+   * Returns the service's clock. Default time source uses {@link System#currentTimeMillis()} to get
+   * current time.
    */
   public Clock clock() {
     return clock;
   }
 
   protected int baseHashCode() {
-    return Objects.hash(projectId, host, httpTransportFactory, authCredentials, retryParams,
-        serviceRpcFactory, connectTimeout, readTimeout, clock);
+    return Objects.hash(projectId, host, httpTransportFactoryClassName, authCredentialsState,
+        retryParams, serviceFactoryClassName, serviceRpcFactoryClassName, connectTimeout,
+        readTimeout, clock);
   }
 
-  protected boolean baseEquals(ServiceOptions<?, ?> other) {
+  protected boolean baseEquals(ServiceOptions<?, ?, ?> other) {
     return Objects.equals(projectId, other.projectId)
         && Objects.equals(host, other.host)
-        && Objects.equals(httpTransportFactory, other.httpTransportFactory)
-        && Objects.equals(authCredentials, other.authCredentials)
+        && Objects.equals(httpTransportFactoryClassName, other.httpTransportFactoryClassName)
+        && Objects.equals(authCredentialsState, other.authCredentialsState)
         && Objects.equals(retryParams, other.retryParams)
-        && Objects.equals(serviceRpcFactory, other.serviceRpcFactory)
+        && Objects.equals(serviceFactoryClassName, other.serviceFactoryClassName)
+        && Objects.equals(serviceRpcFactoryClassName, other.serviceRpcFactoryClassName)
         && Objects.equals(connectTimeout, other.connectTimeout)
         && Objects.equals(readTimeout, other.readTimeout)
         && Objects.equals(clock, clock);
   }
 
-  public abstract Builder<ServiceRpcT, OptionsT, ?> toBuilder();
+  private void readObject(ObjectInputStream input) throws IOException, ClassNotFoundException {
+    input.defaultReadObject();
+    httpTransportFactory = newInstance(httpTransportFactoryClassName);
+    serviceFactory = newInstance(serviceFactoryClassName);
+    serviceRpcFactory = newInstance(serviceRpcFactoryClassName);
+    authCredentials = authCredentialsState.restore();
+  }
 
-  /**
-   * Creates a service RPC using a factory loaded by {@link ServiceLoader}.
-   */
-  protected static
-      <ServiceRpcT, OptionsT extends ServiceOptions<ServiceRpcT, OptionsT>>
-      ServiceRpcT createRpc(OptionsT options,
-          Class<? extends ServiceRpcFactory<ServiceRpcT, OptionsT>> factoryClass) {
-    ServiceRpcFactory<ServiceRpcT, OptionsT> factory =
-        Iterables.getFirst(ServiceLoader.load(factoryClass), null);
-    return factory == null ? null : factory.create(options);
+  private static <T> T newInstance(String className) throws IOException, ClassNotFoundException {
+    try {
+      return (T) Class.forName(className).newInstance();
+    } catch (InstantiationException | IllegalAccessException e) {
+      throw new IOException(e);
+    }
+  }
+
+  protected abstract <T extends ServiceFactory<ServiceT, OptionsT>> T defaultServiceFactory();
+
+  protected abstract <T extends ServiceRpcFactory<ServiceRpcT, OptionsT>> T defaultRpcFactory();
+
+  protected abstract Set<String> scopes();
+
+  public abstract <B extends Builder<ServiceT, ServiceRpcT, OptionsT, B>> B toBuilder();
+
+  private static <T> T getFromServiceLoader(Class<? extends T> clazz, T defaultInstance) {
+    return Iterables.getFirst(ServiceLoader.load(clazz), defaultInstance);
   }
 }
diff --git a/gcloud-java-core/src/main/java/com/google/gcloud/spi/ServiceRpcFactory.java b/gcloud-java-core/src/main/java/com/google/gcloud/spi/ServiceRpcFactory.java
index 89e08cda9eda..d20b690167a1 100644
--- a/gcloud-java-core/src/main/java/com/google/gcloud/spi/ServiceRpcFactory.java
+++ b/gcloud-java-core/src/main/java/com/google/gcloud/spi/ServiceRpcFactory.java
@@ -18,16 +18,13 @@
 
 import com.google.gcloud.ServiceOptions;
 
-import java.io.Serializable;
-
 /**
  * A base interface for all service RPC factories.
+ *
+ * Implementation must provide a public no-arg constructor.
  * Loading of a factory implementation is done via {@link java.util.ServiceLoader}.
  */
-public interface ServiceRpcFactory<
-        ServiceRpcT,
-        OptionsT extends ServiceOptions<ServiceRpcT, OptionsT>>
-    extends Serializable {
+public interface ServiceRpcFactory<ServiceRpcT, OptionsT extends ServiceOptions> {
 
   ServiceRpcT create(OptionsT options);
 }
diff --git a/gcloud-java-datastore/README.md b/gcloud-java-datastore/README.md
index f0dca1777fe8..bbcdd9d8857c 100644
--- a/gcloud-java-datastore/README.md
+++ b/gcloud-java-datastore/README.md
@@ -52,14 +52,13 @@ Here is a code snippet showing a simple usage example from within Compute/App En
 
 ```java
 import com.google.gcloud.datastore.Datastore;
-import com.google.gcloud.datastore.DatastoreFactory;
 import com.google.gcloud.datastore.DatastoreOptions;
 import com.google.gcloud.datastore.DateTime;
 import com.google.gcloud.datastore.Entity;
 import com.google.gcloud.datastore.Key;
 import com.google.gcloud.datastore.KeyFactory;
 
-Datastore datastore = DatastoreFactory.instance().get(DatastoreOptions.getDefaultInstance());
+Datastore datastore = DatastoreOptions.getDefaultInstance().service();
 KeyFactory keyFactory = datastore.newKeyFactory().kind(KIND);
 Key key = keyFactory.newKey(keyName);
 Entity entity = datastore.get(key);
diff --git a/gcloud-java-datastore/src/main/java/com/google/gcloud/datastore/DatastoreFactory.java b/gcloud-java-datastore/src/main/java/com/google/gcloud/datastore/DatastoreFactory.java
index a64fab3715f1..b1f5a026a3e5 100644
--- a/gcloud-java-datastore/src/main/java/com/google/gcloud/datastore/DatastoreFactory.java
+++ b/gcloud-java-datastore/src/main/java/com/google/gcloud/datastore/DatastoreFactory.java
@@ -17,27 +17,10 @@
 package com.google.gcloud.datastore;
 
 
+import com.google.gcloud.ServiceFactory;
+
 /**
- * A base class for Datastore factories.
+ * An interface for Datastore factories.
  */
-public abstract class DatastoreFactory {
-
-  private static final DatastoreFactory INSTANCE = new DatastoreFactory() {
-      @Override
-      public Datastore get(DatastoreOptions options) {
-        return new DatastoreImpl(options);
-      }
-    };
-
-  /**
-   * Returns the default factory instance.
-   */
-  public static DatastoreFactory instance() {
-    return INSTANCE;
-  }
-
-  /**
-   * Returns a {@code Datastore} service for the given options.
-   */
-  public abstract Datastore get(DatastoreOptions options);
+public interface DatastoreFactory extends ServiceFactory<Datastore, DatastoreOptions> {
 }
diff --git a/gcloud-java-datastore/src/main/java/com/google/gcloud/datastore/DatastoreImpl.java b/gcloud-java-datastore/src/main/java/com/google/gcloud/datastore/DatastoreImpl.java
index c12ffc8a032d..43fd75396538 100644
--- a/gcloud-java-datastore/src/main/java/com/google/gcloud/datastore/DatastoreImpl.java
+++ b/gcloud-java-datastore/src/main/java/com/google/gcloud/datastore/DatastoreImpl.java
@@ -74,7 +74,7 @@ public RetryResult beforeEval(Exception exception) {
 
   DatastoreImpl(DatastoreOptions options) {
     super(options);
-    this.datastoreRpc = options.datastoreRpc();
+    this.datastoreRpc = options.rpc();
     retryParams = MoreObjects.firstNonNull(options.retryParams(), RetryParams.noRetries());
   }
 
diff --git a/gcloud-java-datastore/src/main/java/com/google/gcloud/datastore/DatastoreOptions.java b/gcloud-java-datastore/src/main/java/com/google/gcloud/datastore/DatastoreOptions.java
index 7b0dd3a2a606..328573ac7164 100644
--- a/gcloud-java-datastore/src/main/java/com/google/gcloud/datastore/DatastoreOptions.java
+++ b/gcloud-java-datastore/src/main/java/com/google/gcloud/datastore/DatastoreOptions.java
@@ -34,9 +34,9 @@
 import java.util.Objects;
 import java.util.Set;
 
-public class DatastoreOptions extends ServiceOptions<DatastoreRpc, DatastoreOptions> {
+public class DatastoreOptions extends ServiceOptions<Datastore, DatastoreRpc, DatastoreOptions> {
 
-  private static final long serialVersionUID = -8636602944160689193L;
+  private static final long serialVersionUID = 5056049000758143852L;
   private static final String DATASET_ENV_NAME = "DATASTORE_DATASET";
   private static final String HOST_ENV_NAME = "DATASTORE_HOST";
   private static final String DATASTORE_SCOPE = "https://www.googleapis.com/auth/datastore";
@@ -46,10 +46,29 @@ public class DatastoreOptions extends ServiceOptions<DatastoreRpc, DatastoreOpti
   private final String namespace;
   private final boolean force;
   private final boolean normalizeDataset;
-  private transient DatastoreRpc datastoreRpc;
+
+  public static class DefaultDatastoreFactory implements DatastoreFactory {
+
+    private static final DatastoreFactory INSTANCE = new DefaultDatastoreFactory();
+
+    @Override
+    public Datastore create(DatastoreOptions options) {
+      return new DatastoreImpl(options);
+    }
+  }
+
+  public static class DefaultDatastoreRpcFactory implements DatastoreRpcFactory {
+
+    private static final DatastoreRpcFactory INSTANCE = new DefaultDatastoreRpcFactory();
+
+    @Override
+    public DatastoreRpc create(DatastoreOptions options) {
+      return new DefaultDatastoreRpc(options);
+    }
+  }
 
   public static class Builder extends
-      ServiceOptions.Builder<DatastoreRpc, DatastoreOptions, Builder> {
+      ServiceOptions.Builder<Datastore, DatastoreRpc, DatastoreOptions, Builder> {
 
     private String namespace;
     private boolean force;
@@ -88,7 +107,7 @@ Builder normalizeDataset(boolean normalizeDataset) {
   }
 
   private DatastoreOptions(Builder builder) {
-    super(builder);
+    super(DatastoreFactory.class, DatastoreRpcFactory.class, builder);
     normalizeDataset = builder.normalizeDataset;
     namespace = builder.namespace != null ? builder.namespace : defaultNamespace();
     force = builder.force;
@@ -108,7 +127,7 @@ private DatastoreOptions normalize() {
         .build();
     requestPb.addKey(key);
     try {
-      LookupResponse responsePb = datastoreRpc().lookup(requestPb.build());
+      LookupResponse responsePb = rpc().lookup(requestPb.build());
       if (responsePb.getDeferredCount() > 0) {
         key = responsePb.getDeferred(0);
       } else {
@@ -138,6 +157,16 @@ protected String defaultProject() {
     return projectId != null ? projectId : super.defaultProject();
   }
 
+  @Override
+  protected DatastoreFactory defaultServiceFactory() {
+    return DefaultDatastoreFactory.INSTANCE;
+  }
+
+  @Override
+  protected DatastoreRpcFactory defaultRpcFactory() {
+    return DefaultDatastoreRpcFactory.INSTANCE;
+  }
+
   public String namespace() {
     return namespace;
   }
@@ -186,25 +215,6 @@ public boolean equals(Object obj) {
         && Objects.equals(normalizeDataset, other.normalizeDataset);
   }
 
-  DatastoreRpc datastoreRpc() {
-    if (datastoreRpc != null) {
-      return datastoreRpc;
-    }
-    if (serviceRpcFactory() != null) {
-      datastoreRpc = serviceRpcFactory().create(this);
-    } else {
-      datastoreRpc = createRpc(this, DatastoreRpcFactory.class);
-      if (datastoreRpc == null) {
-        datastoreRpc = new DefaultDatastoreRpc(this);
-      }
-    }
-    return datastoreRpc;
-  }
-
-  public static DatastoreOptions defaultInstance() {
-    return builder().build();
-  }
-
   public static Builder builder() {
     return new Builder();
   }
diff --git a/gcloud-java-datastore/src/main/java/com/google/gcloud/datastore/package-info.java b/gcloud-java-datastore/src/main/java/com/google/gcloud/datastore/package-info.java
index 3b402820e663..2135267d9ac4 100644
--- a/gcloud-java-datastore/src/main/java/com/google/gcloud/datastore/package-info.java
+++ b/gcloud-java-datastore/src/main/java/com/google/gcloud/datastore/package-info.java
@@ -20,7 +20,7 @@
  * <p>A simple usage example:
  * <pre> {@code
  * DatastoreOptions options = DatastoreOptions.builder().projectId(PROJECT_ID).build();
- * Datastore datastore = DatastoreFactory.instance().get(options);
+ * Datastore datastore = options.service();
  * KeyFactory keyFactory = datastore.newKeyFactory().kind(kind);
  * Key key = keyFactory.newKey(keyName);
  * Entity entity = datastore.get(key);
diff --git a/gcloud-java-datastore/src/main/java/com/google/gcloud/datastore/testing/package-info.java b/gcloud-java-datastore/src/main/java/com/google/gcloud/datastore/testing/package-info.java
index 9d20b3c7ffd8..d03c9d85cd09 100644
--- a/gcloud-java-datastore/src/main/java/com/google/gcloud/datastore/testing/package-info.java
+++ b/gcloud-java-datastore/src/main/java/com/google/gcloud/datastore/testing/package-info.java
@@ -25,9 +25,9 @@
  *     .projectId(PROJECT_ID)
  *     .host("localhost:8080")
  *     .build();
- * Datastore localDatastore = DatastoreFactory.instance().get(options);
+ * Datastore localDatastore = options.service();
  * } </pre>
- * 
+ *
  * <p>After the test:
  * <pre> {@code
  * gcdHelper.stop();
diff --git a/gcloud-java-datastore/src/main/java/com/google/gcloud/spi/DefaultDatastoreRpc.java b/gcloud-java-datastore/src/main/java/com/google/gcloud/spi/DefaultDatastoreRpc.java
index e20bd9a3f9d6..ccb89267a29e 100644
--- a/gcloud-java-datastore/src/main/java/com/google/gcloud/spi/DefaultDatastoreRpc.java
+++ b/gcloud-java-datastore/src/main/java/com/google/gcloud/spi/DefaultDatastoreRpc.java
@@ -79,7 +79,7 @@ public DefaultDatastoreRpc(DatastoreOptions options) {
   private static String normalizeHost(String host) {
     host = host.toLowerCase();
     if (includesScheme(host)) {
-      Preconditions.checkArgument(!(host.startsWith("https://") && isLocalHost(host)), 
+      Preconditions.checkArgument(!(host.startsWith("https://") && isLocalHost(host)),
           "\"https\" is not supported for localhost.  Use \"http\" instead.");
       return host;
     }
diff --git a/gcloud-java-datastore/src/test/java/com/google/gcloud/datastore/DatastoreOptionsTest.java b/gcloud-java-datastore/src/test/java/com/google/gcloud/datastore/DatastoreOptionsTest.java
index f9200b3327e1..78536e3f45cb 100644
--- a/gcloud-java-datastore/src/test/java/com/google/gcloud/datastore/DatastoreOptionsTest.java
+++ b/gcloud-java-datastore/src/test/java/com/google/gcloud/datastore/DatastoreOptionsTest.java
@@ -79,8 +79,7 @@ public void testForce() throws Exception {
 
   @Test
   public void testDatastore() throws Exception {
-    assertSame(datastoreRpcFactory, options.build().serviceRpcFactory());
-    assertSame(datastoreRpc, options.build().datastoreRpc());
+    assertSame(datastoreRpc, options.build().rpc());
   }
 
   @Test
diff --git a/gcloud-java-datastore/src/test/java/com/google/gcloud/datastore/DatastoreTest.java b/gcloud-java-datastore/src/test/java/com/google/gcloud/datastore/DatastoreTest.java
index c55820b324fb..c0d21456d28f 100644
--- a/gcloud-java-datastore/src/test/java/com/google/gcloud/datastore/DatastoreTest.java
+++ b/gcloud-java-datastore/src/test/java/com/google/gcloud/datastore/DatastoreTest.java
@@ -126,7 +126,7 @@ public void setUp() throws IOException, InterruptedException {
         .projectId(PROJECT_ID)
         .host("http://localhost:" + PORT)
         .build();
-    datastore = DatastoreFactory.instance().get(options);
+    datastore = options.service();
     StructuredQuery<Key> query = Query.keyQueryBuilder().build();
     QueryResults<Key> result = datastore.run(query);
     datastore.delete(Iterators.toArray(result, Key.class));
@@ -628,7 +628,7 @@ private Datastore createDatastoreForDeferredLookup() throws DatastoreRpcExceptio
             .retryParams(RetryParams.getDefaultInstance())
             .serviceRpcFactory(rpcFactoryMock)
             .build();
-    return DatastoreFactory.instance().get(options);
+    return options.service();
   }
 
   @Test
@@ -741,7 +741,7 @@ public void testRetryableException() throws Exception {
         .retryParams(RetryParams.getDefaultInstance())
         .serviceRpcFactory(rpcFactoryMock)
         .build();
-    Datastore datastore = DatastoreFactory.instance().get(options);
+    Datastore datastore = options.service();
     Entity entity = datastore.get(KEY1);
     assertEquals(ENTITY1, entity);
     EasyMock.verify(rpcFactoryMock, rpcMock);
@@ -764,7 +764,7 @@ public void testNonRetryableException() throws Exception {
         .retryParams(retryParams)
         .serviceRpcFactory(rpcFactoryMock)
         .build();
-    Datastore datastore = DatastoreFactory.instance().get(options);
+    Datastore datastore = options.service();
     thrown.expect(DatastoreException.class);
     thrown.expectMessage(Reason.PERMISSION_DENIED.description());
     datastore.get(KEY1);
@@ -787,7 +787,7 @@ public void testRuntimeException() throws Exception {
         .retryParams(RetryParams.getDefaultInstance())
         .serviceRpcFactory(rpcFactoryMock)
         .build();
-    Datastore datastore = DatastoreFactory.instance().get(options);
+    Datastore datastore = options.service();
     thrown.expect(DatastoreException.class);
     thrown.expectMessage(exceptionMessage);
     datastore.get(KEY1);
diff --git a/gcloud-java-examples/src/main/java/com/google/gcloud/examples/DatastoreExample.java b/gcloud-java-examples/src/main/java/com/google/gcloud/examples/DatastoreExample.java
index c707e6686707..a3b1257f943a 100644
--- a/gcloud-java-examples/src/main/java/com/google/gcloud/examples/DatastoreExample.java
+++ b/gcloud-java-examples/src/main/java/com/google/gcloud/examples/DatastoreExample.java
@@ -17,7 +17,6 @@
 package com.google.gcloud.examples;
 
 import com.google.gcloud.datastore.Datastore;
-import com.google.gcloud.datastore.DatastoreFactory;
 import com.google.gcloud.datastore.DatastoreOptions;
 import com.google.gcloud.datastore.DateTime;
 import com.google.gcloud.datastore.Entity;
@@ -183,7 +182,7 @@ public static void main(String... args) {
         .namespace(NAMESPACE)
         .build();
     String name = args.length > 1 ? args[1] : System.getProperty("user.name");
-    Datastore datastore = DatastoreFactory.instance().get(options);
+    Datastore datastore = options.service();
     KeyFactory keyFactory = datastore.newKeyFactory().kind(USER_KIND);
     Key key = keyFactory.newKey(name);
     String actionName = args.length > 2 ? args[2].toLowerCase() : DEFAULT_ACTION;
diff --git a/gcloud-java-examples/src/main/java/com/google/gcloud/examples/StorageExample.java b/gcloud-java-examples/src/main/java/com/google/gcloud/examples/StorageExample.java
index d47494b953fc..46fb94297ec7 100644
--- a/gcloud-java-examples/src/main/java/com/google/gcloud/examples/StorageExample.java
+++ b/gcloud-java-examples/src/main/java/com/google/gcloud/examples/StorageExample.java
@@ -31,7 +31,6 @@
 import com.google.gcloud.storage.Storage.ComposeRequest;
 import com.google.gcloud.storage.Storage.CopyRequest;
 import com.google.gcloud.storage.Storage.SignUrlOption;
-import com.google.gcloud.storage.StorageFactory;
 import com.google.gcloud.storage.StorageOptions;
 
 import java.io.FileOutputStream;
@@ -41,7 +40,6 @@
 import java.nio.ByteBuffer;
 import java.nio.channels.Channels;
 import java.nio.channels.WritableByteChannel;
-import static java.nio.charset.StandardCharsets.UTF_8;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
@@ -558,7 +556,7 @@ public static void main(String... args) throws Exception {
       printUsage();
       return;
     }
-    Storage storage = StorageFactory.instance().get(optionsBuilder.build());
+    Storage storage = optionsBuilder.build().service();
     Object request;
     try {
       request = action.parse(args);
diff --git a/gcloud-java-storage/README.md b/gcloud-java-storage/README.md
index bd3a9ea0da88..717fd1f1f3e4 100644
--- a/gcloud-java-storage/README.md
+++ b/gcloud-java-storage/README.md
@@ -1,7 +1,7 @@
 Google Cloud Java Client for Storage
 ====================================
 
-Java idiomatic client for [Google Cloud Storage] (https://cloud.google.com/storage/). 
+Java idiomatic client for [Google Cloud Storage] (https://cloud.google.com/storage/).
 
 [![Build Status](https://travis-ci.org/GoogleCloudPlatform/gcloud-java.svg?branch=master)](https://travis-ci.org/GoogleCloudPlatform/gcloud-java)
 [![Coverage Status](https://coveralls.io/repos/GoogleCloudPlatform/gcloud-java/badge.svg?branch=master)](https://coveralls.io/r/GoogleCloudPlatform/gcloud-java?branch=master)
@@ -55,13 +55,12 @@ import static java.nio.charset.StandardCharsets.UTF_8;
 
 import com.google.gcloud.storage.Blob;
 import com.google.gcloud.storage.Storage;
-import com.google.gcloud.storage.StorageFactory;
 import com.google.gcloud.storage.StorageOptions;
 
 import java.nio.ByteBuffer;
 import java.nio.channels.WritableByteChannel;
 
-Storage storage = StorageFactory.instance().get(StorageOptions.getDefaultInstance());
+Storage storage = StorageOptions.getDefaultInstance().service();
 Blob blob = new Blob(storage, "bucket", "blob_name");
 if (!blob.exists()) {
   storage2.create(blob.info(), "Hello, Cloud Storage!".getBytes(UTF_8));
diff --git a/gcloud-java-storage/src/main/java/com/google/gcloud/storage/BlobReadChannel.java b/gcloud-java-storage/src/main/java/com/google/gcloud/storage/BlobReadChannel.java
index b004e3d61634..205dc4b97309 100644
--- a/gcloud-java-storage/src/main/java/com/google/gcloud/storage/BlobReadChannel.java
+++ b/gcloud-java-storage/src/main/java/com/google/gcloud/storage/BlobReadChannel.java
@@ -16,6 +16,7 @@
 
 package com.google.gcloud.storage;
 
+import com.google.gcloud.Restorable;
 import com.google.gcloud.RestorableState;
 
 import java.io.Closeable;
@@ -29,7 +30,8 @@
  *
  * This class is @{link Serializable}, which allows incremental reads.
  */
-public interface BlobReadChannel extends ReadableByteChannel, Closeable {
+public interface BlobReadChannel extends ReadableByteChannel, Closeable,
+    Restorable<BlobReadChannel> {
 
   /**
    * Overridden to remove IOException.
@@ -48,10 +50,11 @@ public interface BlobReadChannel extends ReadableByteChannel, Closeable {
   void chunkSize(int chunkSize);
 
   /**
-   * Saves the read channel state.
+   * Captures the read channel state so that it can be saved and restored afterwards.
    *
    * @return a {@link RestorableState} object that contains the read channel state and can restore
-   *     it afterwards. State object must implement {@link java.io.Serializable}.
+   *     it afterwards.
    */
-  public RestorableState<BlobReadChannel> save();
+  @Override
+  RestorableState<BlobReadChannel> capture();
 }
diff --git a/gcloud-java-storage/src/main/java/com/google/gcloud/storage/BlobReadChannelImpl.java b/gcloud-java-storage/src/main/java/com/google/gcloud/storage/BlobReadChannelImpl.java
index 7731d04837a6..09047a642218 100644
--- a/gcloud-java-storage/src/main/java/com/google/gcloud/storage/BlobReadChannelImpl.java
+++ b/gcloud-java-storage/src/main/java/com/google/gcloud/storage/BlobReadChannelImpl.java
@@ -57,12 +57,12 @@ class BlobReadChannelImpl implements BlobReadChannel {
     this.blob = blob;
     this.requestOptions = requestOptions;
     isOpen = true;
-    storageRpc = serviceOptions.storageRpc();
+    storageRpc = serviceOptions.rpc();
     storageObject = blob.toPb();
   }
 
   @Override
-  public RestorableState<BlobReadChannel> save() {
+  public RestorableState<BlobReadChannel> capture() {
     StateImpl.Builder builder = StateImpl.builder(serviceOptions, blob, requestOptions)
         .position(position)
         .isOpen(isOpen)
diff --git a/gcloud-java-storage/src/main/java/com/google/gcloud/storage/BlobWriteChannel.java b/gcloud-java-storage/src/main/java/com/google/gcloud/storage/BlobWriteChannel.java
index be3ef2293ec3..a6208e5020ae 100644
--- a/gcloud-java-storage/src/main/java/com/google/gcloud/storage/BlobWriteChannel.java
+++ b/gcloud-java-storage/src/main/java/com/google/gcloud/storage/BlobWriteChannel.java
@@ -16,6 +16,7 @@
 
 package com.google.gcloud.storage;
 
+import com.google.gcloud.Restorable;
 import com.google.gcloud.RestorableState;
 
 import java.io.Closeable;
@@ -28,7 +29,8 @@
  * data will only be visible after calling {@link #close()}. This class is serializable, to allow
  * incremental writes.
  */
-public interface BlobWriteChannel extends WritableByteChannel, Closeable {
+public interface BlobWriteChannel extends WritableByteChannel, Closeable,
+    Restorable<BlobWriteChannel> {
 
   /**
    * Sets the minimum size that will be written by a single RPC.
@@ -37,12 +39,13 @@ public interface BlobWriteChannel extends WritableByteChannel, Closeable {
   void chunkSize(int chunkSize);
 
   /**
-   * Saves the write channel state so that it can be restored afterwards. The original
+   * Captures the write channel state so that it can be saved and restored afterwards. The original
    * {@code BlobWriteChannel} and the restored one should not both be used. Closing one channel
    * causes the other channel to close, subsequent writes will fail.
    *
    * @return a {@link RestorableState} object that contains the write channel state and can restore
-   *     it afterwards. State object must implement {@link java.io.Serializable}.
+   *     it afterwards.
    */
-  public RestorableState<BlobWriteChannel> save();
+  @Override
+  RestorableState<BlobWriteChannel> capture();
 }
diff --git a/gcloud-java-storage/src/main/java/com/google/gcloud/storage/BlobWriteChannelImpl.java b/gcloud-java-storage/src/main/java/com/google/gcloud/storage/BlobWriteChannelImpl.java
index 1c841d1dfc6a..8c3254a28d44 100644
--- a/gcloud-java-storage/src/main/java/com/google/gcloud/storage/BlobWriteChannelImpl.java
+++ b/gcloud-java-storage/src/main/java/com/google/gcloud/storage/BlobWriteChannelImpl.java
@@ -57,7 +57,7 @@ class BlobWriteChannelImpl implements BlobWriteChannel {
       Map<StorageRpc.Option, ?> optionsMap) {
     this.options = options;
     this.blobInfo = blobInfo;
-    storageRpc = options.storageRpc();
+    storageRpc = options.rpc();
     storageObject = blobInfo.toPb();
     uploadId = storageRpc.open(storageObject, optionsMap);
   }
@@ -66,12 +66,12 @@ class BlobWriteChannelImpl implements BlobWriteChannel {
     this.options = options;
     this.blobInfo = blobInfo;
     this.uploadId = uploadId;
-    storageRpc = options.storageRpc();
+    storageRpc = options.rpc();
     storageObject = blobInfo.toPb();
   }
 
   @Override
-  public RestorableState<BlobWriteChannel> save() {
+  public RestorableState<BlobWriteChannel> capture() {
     byte[] bufferToSave = null;
     if (isOpen) {
       flush();
diff --git a/gcloud-java-storage/src/main/java/com/google/gcloud/storage/StorageFactory.java b/gcloud-java-storage/src/main/java/com/google/gcloud/storage/StorageFactory.java
index e269f0c9d92b..fbce5559464c 100644
--- a/gcloud-java-storage/src/main/java/com/google/gcloud/storage/StorageFactory.java
+++ b/gcloud-java-storage/src/main/java/com/google/gcloud/storage/StorageFactory.java
@@ -17,27 +17,10 @@
 package com.google.gcloud.storage;
 
 
+import com.google.gcloud.ServiceFactory;
+
 /**
- * A base class for Storage factories.
+ * An interface for Storage factories.
  */
-public abstract class StorageFactory {
-
-  private static final StorageFactory INSTANCE = new StorageFactory() {
-    @Override
-    public Storage get(StorageOptions options) {
-      return new StorageImpl(options);
-    }
-  };
-
-  /**
-   * Returns the default factory instance.
-   */
-  public static StorageFactory instance() {
-    return INSTANCE;
-  }
-
-  /**
-   * Returns a {@code Storage} service for the given options.
-   */
-  public abstract Storage get(StorageOptions options);
+public interface StorageFactory extends ServiceFactory<Storage, StorageOptions> {
 }
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 b6292e1cdf87..6856ee27eec7 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
@@ -101,7 +101,7 @@ public RetryResult beforeEval(Exception exception) {
 
   StorageImpl(StorageOptions options) {
     super(options);
-    storageRpc = options.storageRpc();
+    storageRpc = options.rpc();
   }
 
   @Override
@@ -289,7 +289,7 @@ private static ListResult<BucketInfo> listBuckets(final StorageOptions serviceOp
           new Callable<Tuple<String, Iterable<com.google.api.services.storage.model.Bucket>>>() {
             @Override
             public Tuple<String, Iterable<com.google.api.services.storage.model.Bucket>> call() {
-              return serviceOptions.storageRpc().list(optionsMap);
+              return serviceOptions.rpc().list(optionsMap);
             }
           }, serviceOptions.retryParams(), EXCEPTION_HANDLER);
       String cursor = result.x();
@@ -320,7 +320,7 @@ private static ListResult<BlobInfo> listBlobs(final String bucket,
           new Callable<Tuple<String, Iterable<StorageObject>>>() {
             @Override
             public Tuple<String, Iterable<StorageObject>> call() {
-              return serviceOptions.storageRpc().list(bucket, optionsMap);
+              return serviceOptions.rpc().list(bucket, optionsMap);
             }
           }, serviceOptions.retryParams(), EXCEPTION_HANDLER);
       String cursor = result.x();
diff --git a/gcloud-java-storage/src/main/java/com/google/gcloud/storage/StorageOptions.java b/gcloud-java-storage/src/main/java/com/google/gcloud/storage/StorageOptions.java
index a439e3c8ae49..16c17c3e8d98 100644
--- a/gcloud-java-storage/src/main/java/com/google/gcloud/storage/StorageOptions.java
+++ b/gcloud-java-storage/src/main/java/com/google/gcloud/storage/StorageOptions.java
@@ -26,7 +26,7 @@
 import java.util.Objects;
 import java.util.Set;
 
-public class StorageOptions extends ServiceOptions<StorageRpc, StorageOptions> {
+public class StorageOptions extends ServiceOptions<Storage, StorageRpc, StorageOptions> {
 
   private static final long serialVersionUID = -7804860602287801084L;
   private static final String GCS_SCOPE = "https://www.googleapis.com/auth/devstorage.full_control";
@@ -34,10 +34,29 @@ public class StorageOptions extends ServiceOptions<StorageRpc, StorageOptions> {
   private static final String DEFAULT_PATH_DELIMITER = "/";
 
   private final String pathDelimiter;
-  private transient StorageRpc storageRpc;
+
+  public static class DefaultStorageFactory implements StorageFactory {
+
+    private static final StorageFactory INSTANCE = new DefaultStorageFactory();
+
+    @Override
+    public Storage create(StorageOptions options) {
+      return new StorageImpl(options);
+    }
+  }
+
+  public static class DefaultStorageRpcFactory implements StorageRpcFactory {
+
+    private static final StorageRpcFactory INSTANCE = new DefaultStorageRpcFactory();
+
+    @Override
+    public StorageRpc create(StorageOptions options) {
+      return new DefaultStorageRpc(options);
+    }
+  }
 
   public static class Builder extends
-      ServiceOptions.Builder<StorageRpc, StorageOptions, Builder> {
+      ServiceOptions.Builder<Storage, StorageRpc, StorageOptions, Builder> {
 
     private String pathDelimiter;
 
@@ -45,6 +64,7 @@ private Builder() {}
 
     private Builder(StorageOptions options) {
       super(options);
+      pathDelimiter = options.pathDelimiter;
     }
 
     /**
@@ -65,28 +85,23 @@ public StorageOptions build() {
   }
 
   private StorageOptions(Builder builder) {
-    super(builder);
+    super(StorageFactory.class, StorageRpcFactory.class, builder);
     pathDelimiter = MoreObjects.firstNonNull(builder.pathDelimiter, DEFAULT_PATH_DELIMITER);
   }
 
   @Override
-  protected Set<String> scopes() {
-    return SCOPES;
+  protected StorageFactory defaultServiceFactory() {
+    return DefaultStorageFactory.INSTANCE;
   }
 
-  StorageRpc storageRpc() {
-    if (storageRpc != null) {
-      return storageRpc;
-    }
-    if (serviceRpcFactory() != null) {
-      storageRpc = serviceRpcFactory().create(this);
-    } else {
-      storageRpc = createRpc(this, StorageRpcFactory.class);
-      if (storageRpc == null) {
-        storageRpc = new DefaultStorageRpc(this);
-      }
-    }
-    return storageRpc;
+  @Override
+  protected StorageRpcFactory defaultRpcFactory() {
+    return DefaultStorageRpcFactory.INSTANCE;
+  }
+
+  @Override
+  protected Set<String> scopes() {
+    return SCOPES;
   }
 
   /**
@@ -115,10 +130,6 @@ public boolean equals(Object obj) {
     return baseEquals(other) && Objects.equals(pathDelimiter, other.pathDelimiter);
   }
 
-  public static StorageOptions defaultInstance() {
-    return builder().build();
-  }
-
   public static Builder builder() {
     return new Builder();
   }
diff --git a/gcloud-java-storage/src/main/java/com/google/gcloud/storage/package-info.java b/gcloud-java-storage/src/main/java/com/google/gcloud/storage/package-info.java
index b4a701fde840..2a09631be40a 100644
--- a/gcloud-java-storage/src/main/java/com/google/gcloud/storage/package-info.java
+++ b/gcloud-java-storage/src/main/java/com/google/gcloud/storage/package-info.java
@@ -20,7 +20,7 @@
  * <p>A simple usage example:
  * <pre>{@code
  * StorageOptions options = StorageOptions.builder().projectId("project").build();
- * Storage storage = StorageFactory.instance().get(options);
+ * Storage storage = options.service();
  * BlobId blobId = BlobId.of("bucket", "blob_name");
  * Blob blob = Blob.load(storage, blobId);
  * if (blob == null) {
diff --git a/gcloud-java-storage/src/main/java/com/google/gcloud/storage/testing/package-info.java b/gcloud-java-storage/src/main/java/com/google/gcloud/storage/testing/package-info.java
index eca45b4b6306..82b3578284dc 100644
--- a/gcloud-java-storage/src/main/java/com/google/gcloud/storage/testing/package-info.java
+++ b/gcloud-java-storage/src/main/java/com/google/gcloud/storage/testing/package-info.java
@@ -21,7 +21,7 @@
  * <p>Before the test:
  * <pre> {@code
  * RemoteGcsHelper gcsHelper = RemoteGcsHelper.create(PROJECT_ID, "/path/to/JSON/key.json");
- * Storage storage = StorageFactory.instance().get(gcsHelper.options());
+ * Storage storage = gcsHelper.options().service();
  * String bucket = RemoteGcsHelper.generateBucketName();
  * storage.create(BucketInfo.of(bucket));
  * } </pre>
diff --git a/gcloud-java-storage/src/test/java/com/google/gcloud/storage/BlobReadChannelImplTest.java b/gcloud-java-storage/src/test/java/com/google/gcloud/storage/BlobReadChannelImplTest.java
index e8ed915581b8..e1f904bf72fe 100644
--- a/gcloud-java-storage/src/test/java/com/google/gcloud/storage/BlobReadChannelImplTest.java
+++ b/gcloud-java-storage/src/test/java/com/google/gcloud/storage/BlobReadChannelImplTest.java
@@ -16,27 +16,31 @@
 
 package com.google.gcloud.storage;
 
+import static org.easymock.EasyMock.anyObject;
+import static org.easymock.EasyMock.createMock;
+import static org.easymock.EasyMock.expect;
+import static org.easymock.EasyMock.expectLastCall;
+import static org.easymock.EasyMock.replay;
 import static org.easymock.EasyMock.verify;
 import static org.junit.Assert.assertArrayEquals;
-import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
 import com.google.common.collect.ImmutableMap;
 import com.google.gcloud.RestorableState;
-import com.google.gcloud.RetryParams;
 import com.google.gcloud.spi.StorageRpc;
+import com.google.gcloud.spi.StorageRpcFactory;
 
-import org.easymock.EasyMock;
-import org.junit.Test;
+import org.junit.After;
 import org.junit.Before;
+import org.junit.Test;
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.Arrays;
 import java.util.Map;
 import java.util.Random;
-import org.junit.After;
 
 public class BlobReadChannelImplTest {
 
@@ -48,44 +52,44 @@ public class BlobReadChannelImplTest {
   private static final int CUSTOM_CHUNK_SIZE = 2 * 1024 * 1024;
   private static final Random RANDOM = new Random();
 
-  private StorageOptions optionsMock;
+  private StorageOptions options;
+  private StorageRpcFactory rpcFactoryMock;
   private StorageRpc storageRpcMock;
   private BlobReadChannelImpl reader;
 
   @Before
   public void setUp() throws IOException, InterruptedException {
-    optionsMock = EasyMock.createMock(StorageOptions.class);
-    storageRpcMock = EasyMock.createMock(StorageRpc.class);
+    rpcFactoryMock = createMock(StorageRpcFactory.class);
+    storageRpcMock = createMock(StorageRpc.class);
+    expect(rpcFactoryMock.create(anyObject(StorageOptions.class))).andReturn(storageRpcMock);
+    replay(rpcFactoryMock);
+    options = StorageOptions.builder()
+        .projectId("projectId")
+        .serviceRpcFactory(rpcFactoryMock)
+        .build();
   }
 
   @After
   public void tearDown() throws Exception {
-    verify(optionsMock);
-    verify(storageRpcMock);
+    verify(rpcFactoryMock, storageRpcMock);
   }
 
   @Test
   public void testCreate() {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.replay(optionsMock);
-    EasyMock.replay(storageRpcMock);
-    reader = new BlobReadChannelImpl(optionsMock, BLOB_ID, EMPTY_RPC_OPTIONS);
+    replay(storageRpcMock);
+    reader = new BlobReadChannelImpl(options, BLOB_ID, EMPTY_RPC_OPTIONS);
     assertTrue(reader.isOpen());
   }
 
   @Test
   public void testReadBuffered() throws IOException {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries());
-    EasyMock.replay(optionsMock);
-    reader = new BlobReadChannelImpl(optionsMock, BLOB_ID, EMPTY_RPC_OPTIONS);
+    reader = new BlobReadChannelImpl(options, BLOB_ID, EMPTY_RPC_OPTIONS);
     byte[] result = randomByteArray(DEFAULT_CHUNK_SIZE);
     ByteBuffer firstReadBuffer = ByteBuffer.allocate(42);
     ByteBuffer secondReadBuffer = ByteBuffer.allocate(42);
-    EasyMock
-        .expect(storageRpcMock.read(BLOB_ID.toPb(), EMPTY_RPC_OPTIONS, 0, DEFAULT_CHUNK_SIZE))
+    expect(storageRpcMock.read(BLOB_ID.toPb(), EMPTY_RPC_OPTIONS, 0, DEFAULT_CHUNK_SIZE))
         .andReturn(result);
-    EasyMock.replay(storageRpcMock);
+    replay(storageRpcMock);
     reader.read(firstReadBuffer);
     reader.read(secondReadBuffer);
     assertArrayEquals(Arrays.copyOf(result, firstReadBuffer.capacity()), firstReadBuffer.array());
@@ -97,24 +101,17 @@ public void testReadBuffered() throws IOException {
 
   @Test
   public void testReadBig() throws IOException {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries()).times(2);
-    EasyMock.replay(optionsMock);
-    reader = new BlobReadChannelImpl(optionsMock, BLOB_ID, EMPTY_RPC_OPTIONS);
+    reader = new BlobReadChannelImpl(options, BLOB_ID, EMPTY_RPC_OPTIONS);
     reader.chunkSize(CUSTOM_CHUNK_SIZE);
     byte[] firstResult = randomByteArray(DEFAULT_CHUNK_SIZE);
     byte[] secondResult = randomByteArray(DEFAULT_CHUNK_SIZE);
     ByteBuffer firstReadBuffer = ByteBuffer.allocate(DEFAULT_CHUNK_SIZE);
     ByteBuffer secondReadBuffer = ByteBuffer.allocate(42);
-    EasyMock
-        .expect(storageRpcMock.read(BLOB_ID.toPb(), EMPTY_RPC_OPTIONS, 0, DEFAULT_CHUNK_SIZE))
-        .andReturn(firstResult);
-    EasyMock
-        .expect(
-            storageRpcMock.read(BLOB_ID.toPb(), EMPTY_RPC_OPTIONS, DEFAULT_CHUNK_SIZE,
-                CUSTOM_CHUNK_SIZE))
-        .andReturn(secondResult);
-    EasyMock.replay(storageRpcMock);
+    storageRpcMock.read(BLOB_ID.toPb(), EMPTY_RPC_OPTIONS, 0, DEFAULT_CHUNK_SIZE);
+    expectLastCall().andReturn(firstResult);
+    storageRpcMock.read(BLOB_ID.toPb(), EMPTY_RPC_OPTIONS, DEFAULT_CHUNK_SIZE, CUSTOM_CHUNK_SIZE);
+    expectLastCall().andReturn(secondResult);
+    replay(storageRpcMock);
     reader.read(firstReadBuffer);
     reader.read(secondReadBuffer);
     assertArrayEquals(firstResult, firstReadBuffer.array());
@@ -124,42 +121,32 @@ public void testReadBig() throws IOException {
 
   @Test
   public void testReadFinish() throws IOException {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries());
-    EasyMock.replay(optionsMock);
-    reader = new BlobReadChannelImpl(optionsMock, BLOB_ID, EMPTY_RPC_OPTIONS);
+    reader = new BlobReadChannelImpl(options, BLOB_ID, EMPTY_RPC_OPTIONS);
     byte[] result = {};
     ByteBuffer readBuffer = ByteBuffer.allocate(DEFAULT_CHUNK_SIZE);
-    EasyMock
-        .expect(storageRpcMock.read(BLOB_ID.toPb(), EMPTY_RPC_OPTIONS, 0, DEFAULT_CHUNK_SIZE))
+    expect(storageRpcMock.read(BLOB_ID.toPb(), EMPTY_RPC_OPTIONS, 0, DEFAULT_CHUNK_SIZE))
         .andReturn(result);
-    EasyMock.replay(storageRpcMock);
+    replay(storageRpcMock);
     assertEquals(-1, reader.read(readBuffer));
   }
 
   @Test
   public void testSeek() throws IOException {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries());
-    EasyMock.replay(optionsMock);
-    reader = new BlobReadChannelImpl(optionsMock, BLOB_ID, EMPTY_RPC_OPTIONS);
+    reader = new BlobReadChannelImpl(options, BLOB_ID, EMPTY_RPC_OPTIONS);
     reader.seek(42);
     byte[] result = randomByteArray(DEFAULT_CHUNK_SIZE);
     ByteBuffer readBuffer = ByteBuffer.allocate(DEFAULT_CHUNK_SIZE);
-    EasyMock
-        .expect(storageRpcMock.read(BLOB_ID.toPb(), EMPTY_RPC_OPTIONS, 42, DEFAULT_CHUNK_SIZE))
+    expect(storageRpcMock.read(BLOB_ID.toPb(), EMPTY_RPC_OPTIONS, 42, DEFAULT_CHUNK_SIZE))
         .andReturn(result);
-    EasyMock.replay(storageRpcMock);
+    replay(storageRpcMock);
     reader.read(readBuffer);
     assertArrayEquals(result, readBuffer.array());
   }
 
   @Test
   public void testClose() throws IOException {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.replay(optionsMock);
-    EasyMock.replay(storageRpcMock);
-    reader = new BlobReadChannelImpl(optionsMock, BLOB_ID, EMPTY_RPC_OPTIONS);
+    replay(storageRpcMock);
+    reader = new BlobReadChannelImpl(options, BLOB_ID, EMPTY_RPC_OPTIONS);
     assertTrue(reader.isOpen());
     reader.close();
     assertTrue(!reader.isOpen());
@@ -167,10 +154,8 @@ public void testClose() throws IOException {
 
   @Test
   public void testReadClosed() {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.replay(optionsMock);
-    EasyMock.replay(storageRpcMock);
-    reader = new BlobReadChannelImpl(optionsMock, BLOB_ID, EMPTY_RPC_OPTIONS);
+    replay(storageRpcMock);
+    reader = new BlobReadChannelImpl(options, BLOB_ID, EMPTY_RPC_OPTIONS);
     reader.close();
     try {
       ByteBuffer readBuffer = ByteBuffer.allocate(DEFAULT_CHUNK_SIZE);
@@ -183,23 +168,18 @@ public void testReadClosed() {
 
   @Test
   public void testSaveAndRestore() throws IOException, ClassNotFoundException {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock).times(2);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries()).times(2);
-    EasyMock.replay(optionsMock);
     byte[] firstResult = randomByteArray(DEFAULT_CHUNK_SIZE);
     byte[] secondResult = randomByteArray(DEFAULT_CHUNK_SIZE);
     ByteBuffer firstReadBuffer = ByteBuffer.allocate(42);
     ByteBuffer secondReadBuffer = ByteBuffer.allocate(DEFAULT_CHUNK_SIZE);
-    EasyMock
-        .expect(storageRpcMock.read(BLOB_ID.toPb(), EMPTY_RPC_OPTIONS, 0, DEFAULT_CHUNK_SIZE))
+    expect(storageRpcMock.read(BLOB_ID.toPb(), EMPTY_RPC_OPTIONS, 0, DEFAULT_CHUNK_SIZE))
         .andReturn(firstResult);
-    EasyMock
-        .expect(storageRpcMock.read(BLOB_ID.toPb(), EMPTY_RPC_OPTIONS, 42, DEFAULT_CHUNK_SIZE))
+    expect(storageRpcMock.read(BLOB_ID.toPb(), EMPTY_RPC_OPTIONS, 42, DEFAULT_CHUNK_SIZE))
         .andReturn(secondResult);
-    EasyMock.replay(storageRpcMock);
-    reader = new BlobReadChannelImpl(optionsMock, BLOB_ID, EMPTY_RPC_OPTIONS);
+    replay(storageRpcMock);
+    reader = new BlobReadChannelImpl(options, BLOB_ID, EMPTY_RPC_OPTIONS);
     reader.read(firstReadBuffer);
-    RestorableState<BlobReadChannel> readerState = reader.save();
+    RestorableState<BlobReadChannel> readerState = reader.capture();
     BlobReadChannel restoredReader = readerState.restore();
     restoredReader.read(secondReadBuffer);
     assertArrayEquals(Arrays.copyOf(firstResult, firstReadBuffer.capacity()),
@@ -209,13 +189,11 @@ public void testSaveAndRestore() throws IOException, ClassNotFoundException {
 
   @Test
   public void testStateEquals() {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock).times(2);
-    EasyMock.replay(optionsMock);
-    EasyMock.replay(storageRpcMock);
-    reader = new BlobReadChannelImpl(optionsMock, BLOB_ID, EMPTY_RPC_OPTIONS);
-    BlobReadChannel secondReader = new BlobReadChannelImpl(optionsMock, BLOB_ID, EMPTY_RPC_OPTIONS);
-    RestorableState<BlobReadChannel> state = reader.save();
-    RestorableState<BlobReadChannel> secondState = secondReader.save();
+    replay(storageRpcMock);
+    reader = new BlobReadChannelImpl(options, BLOB_ID, EMPTY_RPC_OPTIONS);
+    BlobReadChannel secondReader = new BlobReadChannelImpl(options, BLOB_ID, EMPTY_RPC_OPTIONS);
+    RestorableState<BlobReadChannel> state = reader.capture();
+    RestorableState<BlobReadChannel> secondState = secondReader.capture();
     assertEquals(state, secondState);
     assertEquals(state.hashCode(), secondState.hashCode());
     assertEquals(state.toString(), secondState.toString());
diff --git a/gcloud-java-storage/src/test/java/com/google/gcloud/storage/BlobWriteChannelImplTest.java b/gcloud-java-storage/src/test/java/com/google/gcloud/storage/BlobWriteChannelImplTest.java
index ab3f7a000d90..6faa36173ab9 100644
--- a/gcloud-java-storage/src/test/java/com/google/gcloud/storage/BlobWriteChannelImplTest.java
+++ b/gcloud-java-storage/src/test/java/com/google/gcloud/storage/BlobWriteChannelImplTest.java
@@ -16,23 +16,30 @@
 
 package com.google.gcloud.storage;
 
+import static org.easymock.EasyMock.anyObject;
+import static org.easymock.EasyMock.capture;
+import static org.easymock.EasyMock.captureLong;
+import static org.easymock.EasyMock.createMock;
+import static org.easymock.EasyMock.eq;
+import static org.easymock.EasyMock.expect;
+import static org.easymock.EasyMock.expectLastCall;
+import static org.easymock.EasyMock.replay;
 import static org.easymock.EasyMock.verify;
-import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
 import com.google.common.collect.ImmutableMap;
 import com.google.gcloud.RestorableState;
-import com.google.gcloud.RetryParams;
 import com.google.gcloud.spi.StorageRpc;
+import com.google.gcloud.spi.StorageRpcFactory;
 
 import org.easymock.Capture;
 import org.easymock.CaptureType;
-import org.easymock.EasyMock;
 import org.junit.After;
-import org.junit.Test;
 import org.junit.Before;
+import org.junit.Test;
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
@@ -52,55 +59,53 @@ public class BlobWriteChannelImplTest {
   private static final int CUSTOM_CHUNK_SIZE = 4 * MIN_CHUNK_SIZE;
   private static final Random RANDOM = new Random();
 
-  private StorageOptions optionsMock;
+  private StorageOptions options;
+  private StorageRpcFactory rpcFactoryMock;
   private StorageRpc storageRpcMock;
   private BlobWriteChannelImpl writer;
 
   @Before
   public void setUp() throws IOException, InterruptedException {
-    optionsMock = EasyMock.createMock(StorageOptions.class);
-    storageRpcMock = EasyMock.createMock(StorageRpc.class);
+    rpcFactoryMock = createMock(StorageRpcFactory.class);
+    storageRpcMock = createMock(StorageRpc.class);
+    expect(rpcFactoryMock.create(anyObject(StorageOptions.class)))
+        .andReturn(storageRpcMock);
+    replay(rpcFactoryMock);
+    options = StorageOptions.builder()
+        .projectId("projectid")
+        .serviceRpcFactory(rpcFactoryMock)
+        .build();
   }
 
   @After
   public void tearDown() throws Exception {
-    verify(optionsMock);
-    verify(storageRpcMock);
+    verify(rpcFactoryMock, storageRpcMock);
   }
 
   @Test
   public void testCreate() {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.replay(optionsMock);
-    EasyMock.expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
-    EasyMock.replay(storageRpcMock);
-    writer = new BlobWriteChannelImpl(optionsMock, BLOB_INFO, EMPTY_RPC_OPTIONS);
+    expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
+    replay(storageRpcMock);
+    writer = new BlobWriteChannelImpl(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
     assertTrue(writer.isOpen());
   }
 
   @Test
   public void testWriteWithoutFlush() throws IOException {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.replay(optionsMock);
-    EasyMock.expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
-    EasyMock.replay(storageRpcMock);
-    writer = new BlobWriteChannelImpl(optionsMock, BLOB_INFO, EMPTY_RPC_OPTIONS);
+    expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
+    replay(storageRpcMock);
+    writer = new BlobWriteChannelImpl(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
     assertEquals(MIN_CHUNK_SIZE, writer.write(ByteBuffer.allocate(MIN_CHUNK_SIZE)));
   }
 
   @Test
   public void testWriteWithFlush() throws IOException {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries());
-    EasyMock.replay(optionsMock);
-    EasyMock.expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
+    expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
     Capture<byte[]> capturedBuffer = Capture.newInstance();
-    storageRpcMock.write(EasyMock.eq(UPLOAD_ID), EasyMock.capture(capturedBuffer), EasyMock.eq(0),
-        EasyMock.eq(BLOB_INFO.toPb()), EasyMock.eq(0L), EasyMock.eq(CUSTOM_CHUNK_SIZE),
-        EasyMock.eq(false));
-    EasyMock.expectLastCall();
-    EasyMock.replay(storageRpcMock);
-    writer = new BlobWriteChannelImpl(optionsMock, BLOB_INFO, EMPTY_RPC_OPTIONS);
+    storageRpcMock.write(eq(UPLOAD_ID), capture(capturedBuffer), eq(0),
+        eq(BLOB_INFO.toPb()), eq(0L), eq(CUSTOM_CHUNK_SIZE), eq(false));
+    replay(storageRpcMock);
+    writer = new BlobWriteChannelImpl(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
     writer.chunkSize(CUSTOM_CHUNK_SIZE);
     ByteBuffer buffer = randomBuffer(CUSTOM_CHUNK_SIZE);
     assertEquals(CUSTOM_CHUNK_SIZE, writer.write(buffer));
@@ -109,17 +114,13 @@ public void testWriteWithFlush() throws IOException {
 
   @Test
   public void testWritesAndFlush() throws IOException {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries());
-    EasyMock.replay(optionsMock);
-    EasyMock.expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
+    expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
     Capture<byte[]> capturedBuffer = Capture.newInstance();
-    storageRpcMock.write(EasyMock.eq(UPLOAD_ID), EasyMock.capture(capturedBuffer), EasyMock.eq(0),
-        EasyMock.eq(BLOB_INFO.toPb()), EasyMock.eq(0L), EasyMock.eq(DEFAULT_CHUNK_SIZE),
-        EasyMock.eq(false));
-    EasyMock.expectLastCall();
-    EasyMock.replay(storageRpcMock);
-    writer = new BlobWriteChannelImpl(optionsMock, BLOB_INFO, EMPTY_RPC_OPTIONS);
+    storageRpcMock.write(eq(UPLOAD_ID), capture(capturedBuffer), eq(0),
+        eq(BLOB_INFO.toPb()), eq(0L), eq(DEFAULT_CHUNK_SIZE),
+        eq(false));
+    replay(storageRpcMock);
+    writer = new BlobWriteChannelImpl(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
     ByteBuffer[] buffers = new ByteBuffer[DEFAULT_CHUNK_SIZE / MIN_CHUNK_SIZE];
     for (int i = 0; i < buffers.length; i++) {
       buffers[i] = randomBuffer(MIN_CHUNK_SIZE);
@@ -135,16 +136,12 @@ public void testWritesAndFlush() throws IOException {
 
   @Test
   public void testCloseWithoutFlush() throws IOException {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries());
-    EasyMock.replay(optionsMock);
-    EasyMock.expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
+    expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
     Capture<byte[]> capturedBuffer = Capture.newInstance();
-    storageRpcMock.write(EasyMock.eq(UPLOAD_ID), EasyMock.capture(capturedBuffer), EasyMock.eq(0),
-        EasyMock.eq(BLOB_INFO.toPb()), EasyMock.eq(0L), EasyMock.eq(0), EasyMock.eq(true));
-    EasyMock.expectLastCall();
-    EasyMock.replay(storageRpcMock);
-    writer = new BlobWriteChannelImpl(optionsMock, BLOB_INFO, EMPTY_RPC_OPTIONS);
+    storageRpcMock.write(eq(UPLOAD_ID), capture(capturedBuffer), eq(0),
+        eq(BLOB_INFO.toPb()), eq(0L), eq(0), eq(true));
+    replay(storageRpcMock);
+    writer = new BlobWriteChannelImpl(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
     assertTrue(writer.isOpen());
     writer.close();
     assertArrayEquals(new byte[0], capturedBuffer.getValue());
@@ -153,18 +150,14 @@ public void testCloseWithoutFlush() throws IOException {
 
   @Test
   public void testCloseWithFlush() throws IOException {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries());
-    EasyMock.replay(optionsMock);
-    EasyMock.expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
+    expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
     Capture<byte[]> capturedBuffer = Capture.newInstance();
     ByteBuffer buffer = randomBuffer(MIN_CHUNK_SIZE);
-    storageRpcMock.write(EasyMock.eq(UPLOAD_ID), EasyMock.capture(capturedBuffer), EasyMock.eq(0),
-        EasyMock.eq(BLOB_INFO.toPb()), EasyMock.eq(0L), EasyMock.eq(MIN_CHUNK_SIZE),
-        EasyMock.eq(true));
-    EasyMock.expectLastCall();
-    EasyMock.replay(storageRpcMock);
-    writer = new BlobWriteChannelImpl(optionsMock, BLOB_INFO, EMPTY_RPC_OPTIONS);
+    storageRpcMock.write(eq(UPLOAD_ID), capture(capturedBuffer), eq(0),
+        eq(BLOB_INFO.toPb()), eq(0L), eq(MIN_CHUNK_SIZE),
+        eq(true));
+    replay(storageRpcMock);
+    writer = new BlobWriteChannelImpl(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
     assertTrue(writer.isOpen());
     writer.write(buffer);
     writer.close();
@@ -175,16 +168,12 @@ public void testCloseWithFlush() throws IOException {
 
   @Test
   public void testWriteClosed() throws IOException {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries());
-    EasyMock.replay(optionsMock);
-    EasyMock.expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
+    expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
     Capture<byte[]> capturedBuffer = Capture.newInstance();
-    storageRpcMock.write(EasyMock.eq(UPLOAD_ID), EasyMock.capture(capturedBuffer), EasyMock.eq(0),
-        EasyMock.eq(BLOB_INFO.toPb()), EasyMock.eq(0L), EasyMock.eq(0), EasyMock.eq(true));
-    EasyMock.expectLastCall();
-    EasyMock.replay(storageRpcMock);
-    writer = new BlobWriteChannelImpl(optionsMock, BLOB_INFO, EMPTY_RPC_OPTIONS);
+    storageRpcMock.write(eq(UPLOAD_ID), capture(capturedBuffer), eq(0),
+        eq(BLOB_INFO.toPb()), eq(0L), eq(0), eq(true));
+    replay(storageRpcMock);
+    writer = new BlobWriteChannelImpl(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
     writer.close();
     try {
       writer.write(ByteBuffer.allocate(MIN_CHUNK_SIZE));
@@ -196,24 +185,21 @@ public void testWriteClosed() throws IOException {
 
   @Test
   public void testSaveAndRestore() throws IOException {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock).times(2);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries()).times(2);
-    EasyMock.replay(optionsMock);
-    EasyMock.expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
+    expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
     Capture<byte[]> capturedBuffer = Capture.newInstance(CaptureType.ALL);
     Capture<Long> capturedPosition = Capture.newInstance(CaptureType.ALL);
-    storageRpcMock.write(EasyMock.eq(UPLOAD_ID), EasyMock.capture(capturedBuffer), EasyMock.eq(0),
-        EasyMock.eq(BLOB_INFO.toPb()), EasyMock.captureLong(capturedPosition),
-        EasyMock.eq(DEFAULT_CHUNK_SIZE), EasyMock.eq(false));
-    EasyMock.expectLastCall().times(2);
-    EasyMock.replay(storageRpcMock);
+    storageRpcMock.write(eq(UPLOAD_ID), capture(capturedBuffer), eq(0),
+        eq(BLOB_INFO.toPb()), captureLong(capturedPosition),
+        eq(DEFAULT_CHUNK_SIZE), eq(false));
+    expectLastCall().times(2);
+    replay(storageRpcMock);
     ByteBuffer buffer1 = randomBuffer(DEFAULT_CHUNK_SIZE);
     ByteBuffer buffer2 = randomBuffer(DEFAULT_CHUNK_SIZE);
-    writer = new BlobWriteChannelImpl(optionsMock, BLOB_INFO, EMPTY_RPC_OPTIONS);
+    writer = new BlobWriteChannelImpl(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
     assertEquals(DEFAULT_CHUNK_SIZE, writer.write(buffer1));
     assertArrayEquals(buffer1.array(), capturedBuffer.getValues().get(0));
     assertEquals(new Long(0L), capturedPosition.getValues().get(0));
-    RestorableState<BlobWriteChannel> writerState = writer.save();
+    RestorableState<BlobWriteChannel> writerState = writer.capture();
     BlobWriteChannel restoredWriter = writerState.restore();
     assertEquals(DEFAULT_CHUNK_SIZE, restoredWriter.write(buffer2));
     assertArrayEquals(buffer2.array(), capturedBuffer.getValues().get(1));
@@ -222,20 +208,16 @@ public void testSaveAndRestore() throws IOException {
 
   @Test
   public void testSaveAndRestoreClosed() throws IOException {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock).times(2);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries());
-    EasyMock.replay(optionsMock);
-    EasyMock.expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
+    expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
     Capture<byte[]> capturedBuffer = Capture.newInstance();
-    storageRpcMock.write(EasyMock.eq(UPLOAD_ID), EasyMock.capture(capturedBuffer), EasyMock.eq(0),
-        EasyMock.eq(BLOB_INFO.toPb()), EasyMock.eq(0L), EasyMock.eq(0), EasyMock.eq(true));
-    EasyMock.expectLastCall();
-    EasyMock.replay(storageRpcMock);
-    writer = new BlobWriteChannelImpl(optionsMock, BLOB_INFO, EMPTY_RPC_OPTIONS);
+    storageRpcMock.write(eq(UPLOAD_ID), capture(capturedBuffer), eq(0),
+        eq(BLOB_INFO.toPb()), eq(0L), eq(0), eq(true));
+    replay(storageRpcMock);
+    writer = new BlobWriteChannelImpl(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
     writer.close();
-    RestorableState<BlobWriteChannel> writerState = writer.save();
+    RestorableState<BlobWriteChannel> writerState = writer.capture();
     RestorableState<BlobWriteChannel> expectedWriterState =
-        BlobWriteChannelImpl.StateImpl.builder(optionsMock, BLOB_INFO, UPLOAD_ID)
+        BlobWriteChannelImpl.StateImpl.builder(options, BLOB_INFO, UPLOAD_ID)
             .buffer(null)
             .chunkSize(DEFAULT_CHUNK_SIZE)
             .isOpen(false)
@@ -243,20 +225,18 @@ public void testSaveAndRestoreClosed() throws IOException {
             .build();
     BlobWriteChannel restoredWriter = writerState.restore();
     assertArrayEquals(new byte[0], capturedBuffer.getValue());
-    assertEquals(expectedWriterState, restoredWriter.save());
+    assertEquals(expectedWriterState, restoredWriter.capture());
   }
 
   @Test
   public void testStateEquals() {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock).times(2);
-    EasyMock.replay(optionsMock);
-    EasyMock.expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID)
+    expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID)
         .times(2);
-    EasyMock.replay(storageRpcMock);
-    writer = new BlobWriteChannelImpl(optionsMock, BLOB_INFO, EMPTY_RPC_OPTIONS);
-    BlobWriteChannel writer2 = new BlobWriteChannelImpl(optionsMock, BLOB_INFO, EMPTY_RPC_OPTIONS);
-    RestorableState<BlobWriteChannel> state = writer.save();
-    RestorableState<BlobWriteChannel> state2 = writer2.save();
+    replay(storageRpcMock);
+    writer = new BlobWriteChannelImpl(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
+    BlobWriteChannel writer2 = new BlobWriteChannelImpl(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
+    RestorableState<BlobWriteChannel> state = writer.capture();
+    RestorableState<BlobWriteChannel> state2 = writer2.capture();
     assertEquals(state, state2);
     assertEquals(state.hashCode(), state2.hashCode());
     assertEquals(state.toString(), state2.toString());
diff --git a/gcloud-java-storage/src/test/java/com/google/gcloud/storage/ITStorageTest.java b/gcloud-java-storage/src/test/java/com/google/gcloud/storage/ITStorageTest.java
index 3acb09a18080..a0bae0e0e29e 100644
--- a/gcloud-java-storage/src/test/java/com/google/gcloud/storage/ITStorageTest.java
+++ b/gcloud-java-storage/src/test/java/com/google/gcloud/storage/ITStorageTest.java
@@ -29,6 +29,10 @@
 import com.google.gcloud.RestorableState;
 import com.google.gcloud.storage.testing.RemoteGcsHelper;
 
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.io.InputStream;
@@ -47,14 +51,9 @@
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
-import org.junit.Test;
-
 public class ITStorageTest {
 
   private static Storage storage;
-  private static RemoteGcsHelper gcsHelper;
 
   private static final Logger log = Logger.getLogger(ITStorageTest.class.getName());
   private static final String bucket = RemoteGcsHelper.generateBucketName();
@@ -64,8 +63,8 @@ public class ITStorageTest {
 
   @BeforeClass
   public static void beforeClass() {
-    gcsHelper = RemoteGcsHelper.create();
-    storage = StorageFactory.instance().get(gcsHelper.options());
+    RemoteGcsHelper gcsHelper = RemoteGcsHelper.create();
+    storage = gcsHelper.options().service();
     storage.create(BucketInfo.of(bucket));
   }
 
@@ -445,7 +444,7 @@ public void testBatchRequestFail() {
   }
 
   @Test
-  public void testReadAndWriteChannels() throws UnsupportedEncodingException, IOException {
+  public void testReadAndWriteChannels() throws IOException {
     String blobName = "test-read-and-write-channels-blob";
     BlobInfo blob = BlobInfo.builder(bucket, blobName).build();
     byte[] stringBytes;
@@ -468,14 +467,14 @@ public void testReadAndWriteChannels() throws UnsupportedEncodingException, IOEx
   }
 
   @Test
-  public void testReadAndWriteSaveChannels() throws UnsupportedEncodingException, IOException {
-    String blobName = "test-read-and-write-save-channels-blob";
+  public void testReadAndWriteCaptureChannels() throws IOException {
+    String blobName = "test-read-and-write-capture-channels-blob";
     BlobInfo blob = BlobInfo.builder(bucket, blobName).build();
     byte[] stringBytes;
     BlobWriteChannel writer = storage.writer(blob);
     stringBytes = BLOB_STRING_CONTENT.getBytes(UTF_8);
     writer.write(ByteBuffer.wrap(BLOB_BYTE_CONTENT));
-    RestorableState<BlobWriteChannel> writerState = writer.save();
+    RestorableState<BlobWriteChannel> writerState = writer.capture();
     BlobWriteChannel secondWriter = writerState.restore();
     secondWriter.write(ByteBuffer.wrap(stringBytes));
     secondWriter.close();
@@ -485,7 +484,7 @@ public void testReadAndWriteSaveChannels() throws UnsupportedEncodingException,
     reader.chunkSize(BLOB_BYTE_CONTENT.length);
     readBytes = ByteBuffer.allocate(BLOB_BYTE_CONTENT.length);
     reader.read(readBytes);
-    RestorableState<BlobReadChannel> readerState = reader.save();
+    RestorableState<BlobReadChannel> readerState = reader.capture();
     BlobReadChannel secondReader = readerState.restore();
     readStringBytes = ByteBuffer.allocate(stringBytes.length);
     secondReader.read(readStringBytes);
@@ -497,7 +496,7 @@ public void testReadAndWriteSaveChannels() throws UnsupportedEncodingException,
   }
 
   @Test
-  public void testReadChannelFail() throws UnsupportedEncodingException, IOException {
+  public void testReadChannelFail() throws IOException {
     String blobName = "test-read-channel-blob-fail";
     BlobInfo blob = BlobInfo.builder(bucket, blobName).build();
     assertNotNull(storage.create(blob));
@@ -512,7 +511,7 @@ public void testReadChannelFail() throws UnsupportedEncodingException, IOExcepti
   }
 
   @Test
-  public void testWriteChannelFail() throws UnsupportedEncodingException, IOException {
+  public void testWriteChannelFail() throws IOException {
     String blobName = "test-write-channel-blob-fail";
     BlobInfo blob = BlobInfo.builder(bucket, blobName).generation(-1L).build();
     try {
@@ -527,7 +526,7 @@ public void testWriteChannelFail() throws UnsupportedEncodingException, IOExcept
   }
 
   @Test
-  public void testWriteChannelExistingBlob() throws UnsupportedEncodingException, IOException {
+  public void testWriteChannelExistingBlob() throws IOException {
     String blobName = "test-write-channel-existing-blob";
     BlobInfo blob = BlobInfo.builder(bucket, blobName).build();
     BlobInfo remoteBlob = storage.create(blob);
diff --git a/gcloud-java-storage/src/test/java/com/google/gcloud/storage/SerializationTest.java b/gcloud-java-storage/src/test/java/com/google/gcloud/storage/SerializationTest.java
index edda4ed17e25..50cbdbf9abd5 100644
--- a/gcloud-java-storage/src/test/java/com/google/gcloud/storage/SerializationTest.java
+++ b/gcloud-java-storage/src/test/java/com/google/gcloud/storage/SerializationTest.java
@@ -114,7 +114,7 @@ public void testReadChannelState() throws IOException, ClassNotFoundException {
         .build();
     BlobReadChannel reader =
         new BlobReadChannelImpl(options, BlobId.of("b", "n"), EMPTY_RPC_OPTIONS);
-    RestorableState<BlobReadChannel> state = reader.save();
+    RestorableState<BlobReadChannel> state = reader.capture();
     RestorableState<BlobReadChannel> deserializedState = serializeAndDeserialize(state);
     assertEquals(state, deserializedState);
     assertEquals(state.hashCode(), deserializedState.hashCode());
@@ -130,7 +130,7 @@ public void testWriteChannelState() throws IOException, ClassNotFoundException {
         .build();
     BlobWriteChannelImpl writer = new BlobWriteChannelImpl(
         options, BlobInfo.builder(BlobId.of("b", "n")).build(), "upload-id");
-    RestorableState<BlobWriteChannel> state = writer.save();
+    RestorableState<BlobWriteChannel> state = writer.capture();
     RestorableState<BlobWriteChannel> deserializedState = serializeAndDeserialize(state);
     assertEquals(state, deserializedState);
     assertEquals(state.hashCode(), deserializedState.hashCode());
diff --git a/gcloud-java-storage/src/test/java/com/google/gcloud/storage/StorageImplTest.java b/gcloud-java-storage/src/test/java/com/google/gcloud/storage/StorageImplTest.java
index b3a6fe36859e..9c80d43396c0 100644
--- a/gcloud-java-storage/src/test/java/com/google/gcloud/storage/StorageImplTest.java
+++ b/gcloud-java-storage/src/test/java/com/google/gcloud/storage/StorageImplTest.java
@@ -17,8 +17,8 @@
 package com.google.gcloud.storage;
 
 import static java.nio.charset.StandardCharsets.UTF_8;
-import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertSame;
@@ -31,21 +31,22 @@
 import com.google.common.collect.Iterables;
 import com.google.common.collect.Maps;
 import com.google.common.io.BaseEncoding;
+import com.google.gcloud.AuthCredentials;
 import com.google.gcloud.AuthCredentials.ServiceAccountAuthCredentials;
 import com.google.gcloud.RetryParams;
 import com.google.gcloud.ServiceOptions;
 import com.google.gcloud.spi.StorageRpc;
 import com.google.gcloud.spi.StorageRpc.Tuple;
+import com.google.gcloud.spi.StorageRpcFactory;
 
 import org.easymock.Capture;
 import org.easymock.EasyMock;
-
 import org.junit.After;
 import org.junit.Before;
 import org.junit.BeforeClass;
 import org.junit.Rule;
-import org.junit.rules.ExpectedException;
 import org.junit.Test;
+import org.junit.rules.ExpectedException;
 
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
@@ -194,10 +195,12 @@ public long millis() {
     }
   };
 
+  private static final String ACCOUNT = "account";
   private static PrivateKey privateKey;
   private static PublicKey publicKey;
 
-  private StorageOptions optionsMock;
+  private StorageOptions options;
+  private StorageRpcFactory rpcFactoryMock;
   private StorageRpc storageRpcMock;
   private Storage storage;
 
@@ -217,43 +220,47 @@ public static void beforeClass() throws NoSuchAlgorithmException, InvalidKeySpec
 
   @Before
   public void setUp() throws IOException, InterruptedException {
-    optionsMock = EasyMock.createMock(StorageOptions.class);
+    rpcFactoryMock = EasyMock.createMock(StorageRpcFactory.class);
     storageRpcMock = EasyMock.createMock(StorageRpc.class);
+    EasyMock.expect(rpcFactoryMock.create(EasyMock.anyObject(StorageOptions.class)))
+        .andReturn(storageRpcMock);
+    EasyMock.replay(rpcFactoryMock);
+    options = StorageOptions.builder()
+        .projectId("projectId")
+        .authCredentials(AuthCredentials.noCredentials())
+        .clock(TIME_SOURCE)
+        .serviceRpcFactory(rpcFactoryMock)
+        .build();
   }
 
   @After
   public void tearDown() throws Exception {
-    EasyMock.verify(optionsMock, storageRpcMock);
+    EasyMock.verify(rpcFactoryMock, storageRpcMock);
   }
 
   @Test
   public void testGetOptions() {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.replay(optionsMock, storageRpcMock);
-    storage = StorageFactory.instance().get(optionsMock);
-    assertSame(optionsMock, storage.options());
+    EasyMock.replay(storageRpcMock);
+    storage = options.service();
+    assertSame(options, storage.options());
   }
 
   @Test
   public void testCreateBucket() {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries());
     EasyMock.expect(storageRpcMock.create(BUCKET_INFO1.toPb(), EMPTY_RPC_OPTIONS))
         .andReturn(BUCKET_INFO1.toPb());
-    EasyMock.replay(optionsMock, storageRpcMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    storage = options.service();
     BucketInfo bucket = storage.create(BUCKET_INFO1);
     assertEquals(BUCKET_INFO1.toPb(), bucket.toPb());
   }
 
   @Test
   public void testCreateBucketWithOptions() {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries());
     EasyMock.expect(storageRpcMock.create(BUCKET_INFO1.toPb(), BUCKET_TARGET_OPTIONS))
         .andReturn(BUCKET_INFO1.toPb());
-    EasyMock.replay(optionsMock, storageRpcMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    storage = options.service();
     BucketInfo bucket =
         storage.create(BUCKET_INFO1, BUCKET_TARGET_METAGENERATION, BUCKET_TARGET_PREDEFINED_ACL);
     assertEquals(BUCKET_INFO1, bucket);
@@ -261,16 +268,14 @@ public void testCreateBucketWithOptions() {
 
   @Test
   public void testCreateBlob() throws IOException {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries());
     Capture<ByteArrayInputStream> capturedStream = Capture.newInstance();
     EasyMock.expect(storageRpcMock.create(
         EasyMock.eq(BLOB_INFO1.toBuilder().md5(CONTENT_MD5).crc32c(CONTENT_CRC32C).build().toPb()),
         EasyMock.capture(capturedStream),
         EasyMock.eq(EMPTY_RPC_OPTIONS)))
         .andReturn(BLOB_INFO1.toPb());
-    EasyMock.replay(optionsMock, storageRpcMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    storage = options.service();
     BlobInfo blob = storage.create(BLOB_INFO1, BLOB_CONTENT);
     assertEquals(BLOB_INFO1, blob);
     ByteArrayInputStream byteStream = capturedStream.getValue();
@@ -282,8 +287,6 @@ public void testCreateBlob() throws IOException {
 
   @Test
   public void testCreateEmptyBlob() throws IOException {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries());
     Capture<ByteArrayInputStream> capturedStream = Capture.newInstance();
     EasyMock.expect(storageRpcMock.create(
         EasyMock.eq(BLOB_INFO1.toBuilder()
@@ -294,8 +297,8 @@ public void testCreateEmptyBlob() throws IOException {
         EasyMock.capture(capturedStream),
         EasyMock.eq(EMPTY_RPC_OPTIONS)))
         .andReturn(BLOB_INFO1.toPb());
-    EasyMock.replay(optionsMock, storageRpcMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    storage = options.service();
     BlobInfo blob = storage.create(BLOB_INFO1);
     assertEquals(BLOB_INFO1, blob);
     ByteArrayInputStream byteStream = capturedStream.getValue();
@@ -305,8 +308,6 @@ public void testCreateEmptyBlob() throws IOException {
 
   @Test
   public void testCreateBlobWithOptions() throws IOException {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries());
     Capture<ByteArrayInputStream> capturedStream = Capture.newInstance();
     EasyMock.expect(storageRpcMock.create(
         EasyMock.eq(BLOB_INFO1.toBuilder()
@@ -317,8 +318,8 @@ public void testCreateBlobWithOptions() throws IOException {
         EasyMock.capture(capturedStream),
         EasyMock.eq(BLOB_TARGET_OPTIONS_CREATE)))
         .andReturn(BLOB_INFO1.toPb());
-    EasyMock.replay(optionsMock, storageRpcMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    storage = options.service();
     BlobInfo blob =
         storage.create(BLOB_INFO1, BLOB_CONTENT, BLOB_TARGET_METAGENERATION, BLOB_TARGET_NOT_EXIST,
             BLOB_TARGET_PREDEFINED_ACL);
@@ -332,40 +333,34 @@ public void testCreateBlobWithOptions() throws IOException {
 
   @Test
   public void testCreateBlobFromStream() throws IOException {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries());
     ByteArrayInputStream fileStream = new ByteArrayInputStream(BLOB_CONTENT);
     BlobInfo.Builder infoBuilder = BLOB_INFO1.toBuilder();
     BlobInfo infoWithHashes = infoBuilder.md5(CONTENT_MD5).crc32c(CONTENT_CRC32C).build();
     BlobInfo infoWithoutHashes = infoBuilder.md5(null).crc32c(null).build();
     EasyMock.expect(storageRpcMock.create(infoWithoutHashes.toPb(), fileStream, EMPTY_RPC_OPTIONS))
         .andReturn(BLOB_INFO1.toPb());
-    EasyMock.replay(optionsMock, storageRpcMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    storage = options.service();
     BlobInfo blob = storage.create(infoWithHashes, fileStream);
     assertEquals(BLOB_INFO1, blob);
   }
 
   @Test
   public void testGetBucket() {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries());
     EasyMock.expect(storageRpcMock.get(BucketInfo.of(BUCKET_NAME1).toPb(), EMPTY_RPC_OPTIONS))
         .andReturn(BUCKET_INFO1.toPb());
-    EasyMock.replay(optionsMock, storageRpcMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    storage = options.service();
     BucketInfo bucket = storage.get(BUCKET_NAME1);
     assertEquals(BUCKET_INFO1, bucket);
   }
 
   @Test
   public void testGetBucketWithOptions() {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries());
     EasyMock.expect(storageRpcMock.get(BucketInfo.of(BUCKET_NAME1).toPb(), BUCKET_SOURCE_OPTIONS))
         .andReturn(BUCKET_INFO1.toPb());
-    EasyMock.replay(optionsMock, storageRpcMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    storage = options.service();
     BucketInfo bucket =
         storage.get(BUCKET_NAME1,
             Storage.BucketSourceOption.metagenerationMatch(BUCKET_INFO1.metageneration()));
@@ -374,26 +369,22 @@ public void testGetBucketWithOptions() {
 
   @Test
   public void testGetBlob() {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries());
     EasyMock.expect(
         storageRpcMock.get(BlobId.of(BUCKET_NAME1, BLOB_NAME1).toPb(), EMPTY_RPC_OPTIONS))
         .andReturn(BLOB_INFO1.toPb());
-    EasyMock.replay(optionsMock, storageRpcMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    storage = options.service();
     BlobInfo blob = storage.get(BUCKET_NAME1, BLOB_NAME1);
     assertEquals(BLOB_INFO1, blob);
   }
 
   @Test
   public void testGetBlobWithOptions() {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries());
     EasyMock.expect(
         storageRpcMock.get(BlobId.of(BUCKET_NAME1, BLOB_NAME1).toPb(), BLOB_SOURCE_OPTIONS))
         .andReturn(BLOB_INFO1.toPb());
-    EasyMock.replay(optionsMock, storageRpcMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    storage = options.service();
     BlobInfo blob =
         storage.get(BUCKET_NAME1, BLOB_NAME1, BLOB_SOURCE_METAGENERATION, BLOB_SOURCE_GENERATION);
     assertEquals(BLOB_INFO1, blob);
@@ -401,15 +392,13 @@ public void testGetBlobWithOptions() {
 
   @Test
   public void testListBuckets() {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock).times(2);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries());
     String cursor = "cursor";
     ImmutableList<BucketInfo> bucketList = ImmutableList.of(BUCKET_INFO1, BUCKET_INFO2);
     Tuple<String, Iterable<com.google.api.services.storage.model.Bucket>> result =
         Tuple.of(cursor, Iterables.transform(bucketList, BucketInfo.TO_PB_FUNCTION));
     EasyMock.expect(storageRpcMock.list(EMPTY_RPC_OPTIONS)).andReturn(result);
-    EasyMock.replay(optionsMock, storageRpcMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    storage = options.service();
     ListResult<BucketInfo> listResult = storage.list();
     assertEquals(cursor, listResult.nextPageCursor());
     assertArrayEquals(bucketList.toArray(), Iterables.toArray(listResult, BucketInfo.class));
@@ -417,12 +406,10 @@ public void testListBuckets() {
 
   @Test
   public void testListBucketsEmpty() {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock).times(2);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries());
     EasyMock.expect(storageRpcMock.list(EMPTY_RPC_OPTIONS)).andReturn(
         Tuple.<String, Iterable<com.google.api.services.storage.model.Bucket>>of(null, null));
-    EasyMock.replay(optionsMock, storageRpcMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    storage = options.service();
     ListResult<BucketInfo> listResult = storage.list();
     assertNull(listResult.nextPageCursor());
     assertArrayEquals(ImmutableList.of().toArray(),
@@ -431,15 +418,13 @@ public void testListBucketsEmpty() {
 
   @Test
   public void testListBucketsWithOptions() {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock).times(2);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries());
     String cursor = "cursor";
     ImmutableList<BucketInfo> bucketList = ImmutableList.of(BUCKET_INFO1, BUCKET_INFO2);
     Tuple<String, Iterable<com.google.api.services.storage.model.Bucket>> result =
         Tuple.of(cursor, Iterables.transform(bucketList, BucketInfo.TO_PB_FUNCTION));
     EasyMock.expect(storageRpcMock.list(BUCKET_LIST_OPTIONS)).andReturn(result);
-    EasyMock.replay(optionsMock, storageRpcMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    storage = options.service();
     ListResult<BucketInfo> listResult = storage.list(BUCKET_LIST_MAX_RESULT, BUCKET_LIST_PREFIX);
     assertEquals(cursor, listResult.nextPageCursor());
     assertArrayEquals(bucketList.toArray(), Iterables.toArray(listResult, BucketInfo.class));
@@ -447,15 +432,13 @@ public void testListBucketsWithOptions() {
 
   @Test
   public void testListBlobs() {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock).times(2);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries());
     String cursor = "cursor";
     ImmutableList<BlobInfo> blobList = ImmutableList.of(BLOB_INFO1, BLOB_INFO2);
     Tuple<String, Iterable<com.google.api.services.storage.model.StorageObject>> result =
         Tuple.of(cursor, Iterables.transform(blobList, BlobInfo.TO_PB_FUNCTION));
     EasyMock.expect(storageRpcMock.list(BUCKET_NAME1, EMPTY_RPC_OPTIONS)).andReturn(result);
-    EasyMock.replay(optionsMock, storageRpcMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    storage = options.service();
     ListResult<BlobInfo> listResult = storage.list(BUCKET_NAME1);
     assertEquals(cursor, listResult.nextPageCursor());
     assertArrayEquals(blobList.toArray(), Iterables.toArray(listResult, BlobInfo.class));
@@ -463,14 +446,12 @@ public void testListBlobs() {
 
   @Test
   public void testListBlobsEmpty() {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock).times(2);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries());
     EasyMock.expect(storageRpcMock.list(BUCKET_NAME1, EMPTY_RPC_OPTIONS))
         .andReturn(
             Tuple.<String, Iterable<com.google.api.services.storage.model.StorageObject>>of(null,
                 null));
-    EasyMock.replay(optionsMock, storageRpcMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    storage = options.service();
     ListResult<BlobInfo> listResult = storage.list(BUCKET_NAME1);
     assertNull(listResult.nextPageCursor());
     assertArrayEquals(ImmutableList.of().toArray(), Iterables.toArray(listResult, BlobInfo.class));
@@ -478,15 +459,13 @@ public void testListBlobsEmpty() {
 
   @Test
   public void testListBlobsWithOptions() {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock).times(2);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries());
     String cursor = "cursor";
     ImmutableList<BlobInfo> blobList = ImmutableList.of(BLOB_INFO1, BLOB_INFO2);
     Tuple<String, Iterable<com.google.api.services.storage.model.StorageObject>> result =
         Tuple.of(cursor, Iterables.transform(blobList, BlobInfo.TO_PB_FUNCTION));
     EasyMock.expect(storageRpcMock.list(BUCKET_NAME1, BLOB_LIST_OPTIONS)).andReturn(result);
-    EasyMock.replay(optionsMock, storageRpcMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    storage = options.service();
     ListResult<BlobInfo> listResult =
         storage.list(BUCKET_NAME1, BLOB_LIST_MAX_RESULT, BLOB_LIST_PREFIX);
     assertEquals(cursor, listResult.nextPageCursor());
@@ -496,12 +475,10 @@ public void testListBlobsWithOptions() {
   @Test
   public void testUpdateBucket() {
     BucketInfo updatedBucketInfo = BUCKET_INFO1.toBuilder().indexPage("some-page").build();
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries());
     EasyMock.expect(storageRpcMock.patch(updatedBucketInfo.toPb(), EMPTY_RPC_OPTIONS))
         .andReturn(updatedBucketInfo.toPb());
-    EasyMock.replay(optionsMock, storageRpcMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    storage = options.service();
     BucketInfo bucket = storage.update(updatedBucketInfo);
     assertEquals(updatedBucketInfo, bucket);
   }
@@ -509,12 +486,10 @@ public void testUpdateBucket() {
   @Test
   public void testUpdateBucketWithOptions() {
     BucketInfo updatedBucketInfo = BUCKET_INFO1.toBuilder().indexPage("some-page").build();
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries());
     EasyMock.expect(storageRpcMock.patch(updatedBucketInfo.toPb(), BUCKET_TARGET_OPTIONS))
         .andReturn(updatedBucketInfo.toPb());
-    EasyMock.replay(optionsMock, storageRpcMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    storage = options.service();
     BucketInfo bucket =
         storage.update(updatedBucketInfo, BUCKET_TARGET_METAGENERATION,
             BUCKET_TARGET_PREDEFINED_ACL);
@@ -524,12 +499,10 @@ public void testUpdateBucketWithOptions() {
   @Test
   public void testUpdateBlob() {
     BlobInfo updatedBlobInfo = BLOB_INFO1.toBuilder().contentType("some-content-type").build();
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries());
     EasyMock.expect(storageRpcMock.patch(updatedBlobInfo.toPb(), EMPTY_RPC_OPTIONS))
         .andReturn(updatedBlobInfo.toPb());
-    EasyMock.replay(optionsMock, storageRpcMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    storage = options.service();
     BlobInfo blob = storage.update(updatedBlobInfo);
     assertEquals(updatedBlobInfo, blob);
   }
@@ -537,12 +510,10 @@ public void testUpdateBlob() {
   @Test
   public void testUpdateBlobWithOptions() {
     BlobInfo updatedBlobInfo = BLOB_INFO1.toBuilder().contentType("some-content-type").build();
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries());
     EasyMock.expect(storageRpcMock.patch(updatedBlobInfo.toPb(), BLOB_TARGET_OPTIONS_UPDATE))
         .andReturn(updatedBlobInfo.toPb());
-    EasyMock.replay(optionsMock, storageRpcMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    storage = options.service();
     BlobInfo blob =
         storage.update(updatedBlobInfo, BLOB_TARGET_METAGENERATION, BLOB_TARGET_PREDEFINED_ACL);
     assertEquals(updatedBlobInfo, blob);
@@ -550,48 +521,40 @@ public void testUpdateBlobWithOptions() {
 
   @Test
   public void testDeleteBucket() {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries());
     EasyMock.expect(storageRpcMock.delete(BucketInfo.of(BUCKET_NAME1).toPb(), EMPTY_RPC_OPTIONS))
         .andReturn(true);
-    EasyMock.replay(optionsMock, storageRpcMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    storage = options.service();
     assertTrue(storage.delete(BUCKET_NAME1));
   }
 
   @Test
   public void testDeleteBucketWithOptions() {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries());
     EasyMock
         .expect(storageRpcMock.delete(BucketInfo.of(BUCKET_NAME1).toPb(), BUCKET_SOURCE_OPTIONS))
         .andReturn(true);
-    EasyMock.replay(optionsMock, storageRpcMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    storage = options.service();
     assertTrue(storage.delete(BUCKET_NAME1, BUCKET_SOURCE_METAGENERATION));
   }
 
   @Test
   public void testDeleteBlob() {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries());
     EasyMock.expect(
         storageRpcMock.delete(BlobId.of(BUCKET_NAME1, BLOB_NAME1).toPb(), EMPTY_RPC_OPTIONS))
         .andReturn(true);
-    EasyMock.replay(optionsMock, storageRpcMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    storage = options.service();
     assertTrue(storage.delete(BUCKET_NAME1, BLOB_NAME1));
   }
 
   @Test
   public void testDeleteBlobWithOptions() {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries());
     EasyMock.expect(
         storageRpcMock.delete(BlobId.of(BUCKET_NAME1, BLOB_NAME1).toPb(), BLOB_SOURCE_OPTIONS))
         .andReturn(true);
-    EasyMock.replay(optionsMock, storageRpcMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    storage = options.service();
     assertTrue(storage.delete(BUCKET_NAME1, BLOB_NAME1, BLOB_SOURCE_GENERATION,
         BLOB_SOURCE_METAGENERATION));
   }
@@ -602,12 +565,10 @@ public void testCompose() {
         .addSource(BLOB_NAME2, BLOB_NAME3)
         .target(BLOB_INFO1)
         .build();
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries());
     EasyMock.expect(storageRpcMock.compose(ImmutableList.of(BLOB_INFO2.toPb(), BLOB_INFO3.toPb()),
         BLOB_INFO1.toPb(), EMPTY_RPC_OPTIONS)).andReturn(BLOB_INFO1.toPb());
-    EasyMock.replay(optionsMock, storageRpcMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    storage = options.service();
     BlobInfo blob = storage.compose(req);
     assertEquals(BLOB_INFO1, blob);
   }
@@ -619,12 +580,10 @@ public void testComposeWithOptions() {
         .target(BLOB_INFO1)
         .targetOptions(BLOB_TARGET_GENERATION, BLOB_TARGET_METAGENERATION)
         .build();
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries());
     EasyMock.expect(storageRpcMock.compose(ImmutableList.of(BLOB_INFO2.toPb(), BLOB_INFO3.toPb()),
         BLOB_INFO1.toPb(), BLOB_TARGET_OPTIONS_COMPOSE)).andReturn(BLOB_INFO1.toPb());
-    EasyMock.replay(optionsMock, storageRpcMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    storage = options.service();
     BlobInfo blob = storage.compose(req);
     assertEquals(BLOB_INFO1, blob);
   }
@@ -635,12 +594,10 @@ public void testCopy() {
         .source(BUCKET_NAME1, BLOB_NAME2)
         .target(BLOB_INFO1)
         .build();
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries());
     EasyMock.expect(storageRpcMock.copy(BLOB_INFO2.toPb(), EMPTY_RPC_OPTIONS, BLOB_INFO1.toPb(),
         EMPTY_RPC_OPTIONS)).andReturn(BLOB_INFO1.toPb());
-    EasyMock.replay(optionsMock, storageRpcMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    storage = options.service();
     BlobInfo blob = storage.copy(req);
     assertEquals(BLOB_INFO1, blob);
   }
@@ -653,39 +610,33 @@ public void testCopyWithOptions() {
         .target(BLOB_INFO1)
         .targetOptions(BLOB_TARGET_GENERATION, BLOB_TARGET_METAGENERATION)
         .build();
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries());
     EasyMock.expect(
         storageRpcMock.copy(BLOB_INFO2.toPb(), BLOB_SOURCE_OPTIONS_COPY, BLOB_INFO1.toPb(),
             BLOB_TARGET_OPTIONS_COMPOSE)).andReturn(BLOB_INFO1.toPb());
-    EasyMock.replay(optionsMock, storageRpcMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    storage = options.service();
     BlobInfo blob = storage.copy(req);
     assertEquals(BLOB_INFO1, blob);
   }
 
   @Test
   public void testReadAllBytes() {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries());
     EasyMock.expect(
         storageRpcMock.load(BlobId.of(BUCKET_NAME1, BLOB_NAME1).toPb(), EMPTY_RPC_OPTIONS))
         .andReturn(BLOB_CONTENT);
-    EasyMock.replay(optionsMock, storageRpcMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    storage = options.service();
     byte[] readBytes = storage.readAllBytes(BUCKET_NAME1, BLOB_NAME1);
     assertArrayEquals(BLOB_CONTENT, readBytes);
   }
 
   @Test
   public void testReadAllBytesWithOptions() {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries());
     EasyMock.expect(
         storageRpcMock.load(BlobId.of(BUCKET_NAME1, BLOB_NAME1).toPb(), BLOB_SOURCE_OPTIONS))
         .andReturn(BLOB_CONTENT);
-    EasyMock.replay(optionsMock, storageRpcMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    storage = options.service();
     byte[] readBytes = storage.readAllBytes(BUCKET_NAME1, BLOB_NAME1, BLOB_SOURCE_GENERATION,
         BLOB_SOURCE_METAGENERATION);
     assertArrayEquals(BLOB_CONTENT, readBytes);
@@ -732,11 +683,11 @@ public Tuple<StorageObject, StorageException> apply(StorageObject f) {
     StorageRpc.BatchResponse res =
         new StorageRpc.BatchResponse(deleteResult, updateResult, getResult);
 
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
+
     Capture<StorageRpc.BatchRequest> capturedBatchRequest = Capture.newInstance();
     EasyMock.expect(storageRpcMock.batch(EasyMock.capture(capturedBatchRequest))).andReturn(res);
-    EasyMock.replay(optionsMock, storageRpcMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    storage = options.service();
     BatchResponse batchResponse = storage.apply(req);
 
     // Verify captured StorageRpc.BatchRequest
@@ -773,9 +724,8 @@ public Tuple<StorageObject, StorageException> apply(StorageObject f) {
 
   @Test
   public void testReader() {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock).times(2);
-    EasyMock.replay(optionsMock, storageRpcMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    storage = options.service();
     BlobReadChannel channel = storage.reader(BUCKET_NAME1, BLOB_NAME1);
     assertNotNull(channel);
     assertTrue(channel.isOpen());
@@ -784,13 +734,11 @@ public void testReader() {
   @Test
   public void testReaderWithOptions() throws IOException {
     byte[] result = new byte[DEFAULT_CHUNK_SIZE];
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.noRetries());
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock).times(2);
     EasyMock.expect(
         storageRpcMock.read(BLOB_INFO2.toPb(), BLOB_SOURCE_OPTIONS, 0, DEFAULT_CHUNK_SIZE))
         .andReturn(result);
-    EasyMock.replay(optionsMock, storageRpcMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    storage = options.service();
     BlobReadChannel channel = storage.reader(BUCKET_NAME1, BLOB_NAME2, BLOB_SOURCE_GENERATION,
         BLOB_SOURCE_METAGENERATION);
     assertNotNull(channel);
@@ -800,14 +748,13 @@ public void testReaderWithOptions() throws IOException {
 
   @Test
   public void testWriter() {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock).times(2);
     BlobInfo.Builder infoBuilder = BLOB_INFO1.toBuilder();
     BlobInfo infoWithHashes = infoBuilder.md5(CONTENT_MD5).crc32c(CONTENT_CRC32C).build();
     BlobInfo infoWithoutHashes = infoBuilder.md5(null).crc32c(null).build();
     EasyMock.expect(storageRpcMock.open(infoWithoutHashes.toPb(), EMPTY_RPC_OPTIONS))
         .andReturn("upload-id");
-    EasyMock.replay(optionsMock, storageRpcMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    storage = options.service();
     BlobWriteChannel channel = storage.writer(infoWithHashes);
     assertNotNull(channel);
     assertTrue(channel.isOpen());
@@ -815,12 +762,11 @@ public void testWriter() {
 
   @Test
   public void testWriterWithOptions() {
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock).times(2);
     BlobInfo info = BLOB_INFO1.toBuilder().md5(CONTENT_MD5).crc32c(CONTENT_CRC32C).build();
     EasyMock.expect(storageRpcMock.open(info.toPb(), BLOB_TARGET_OPTIONS_CREATE))
         .andReturn("upload-id");
-    EasyMock.replay(optionsMock, storageRpcMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    storage = options.service();
     BlobWriteChannel channel = storage.writer(info, BLOB_WRITE_METAGENERATION, BLOB_WRITE_NOT_EXIST,
         BLOB_WRITE_PREDEFINED_ACL, BLOB_WRITE_CRC2C, BLOB_WRITE_MD5_HASH);
     assertNotNull(channel);
@@ -830,21 +776,15 @@ public void testWriterWithOptions() {
   @Test
   public void testSignUrl() throws NoSuchAlgorithmException, InvalidKeyException,
       SignatureException, UnsupportedEncodingException {
-    String account = "account";
-    ServiceAccountAuthCredentials credentialsMock =
-        EasyMock.createMock(ServiceAccountAuthCredentials.class);
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.expect(optionsMock.authCredentials()).andReturn(credentialsMock).times(2);
-    EasyMock.expect(optionsMock.clock()).andReturn(TIME_SOURCE);
-    EasyMock.expect(credentialsMock.privateKey()).andReturn(privateKey);
-    EasyMock.expect(credentialsMock.account()).andReturn(account);
-    EasyMock.replay(optionsMock, storageRpcMock, credentialsMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    ServiceAccountAuthCredentials authCredentials =
+        ServiceAccountAuthCredentials.createFor(ACCOUNT, privateKey);
+    storage = options.toBuilder().authCredentials(authCredentials).build().service();
     URL url = storage.signUrl(BLOB_INFO1, 14, TimeUnit.DAYS);
     String stringUrl = url.toString();
     String expectedUrl =
         new StringBuilder("https://storage.googleapis.com/").append(BUCKET_NAME1).append("/")
-            .append(BLOB_NAME1).append("?GoogleAccessId=").append(account).append("&Expires=")
+            .append(BLOB_NAME1).append("?GoogleAccessId=").append(ACCOUNT).append("&Expires=")
             .append(42L + 1209600).append("&Signature=").toString();
     assertTrue(stringUrl.startsWith(expectedUrl));
     String signature = stringUrl.substring(expectedUrl.length());
@@ -859,22 +799,15 @@ public void testSignUrl() throws NoSuchAlgorithmException, InvalidKeyException,
     signer.update(signedMessageBuilder.toString().getBytes(UTF_8));
     assertTrue(signer.verify(BaseEncoding.base64().decode(
         URLDecoder.decode(signature, UTF_8.name()))));
-    EasyMock.verify(credentialsMock);
   }
 
   @Test
   public void testSignUrlWithOptions() throws NoSuchAlgorithmException, InvalidKeyException,
       SignatureException, UnsupportedEncodingException {
-    String account = "account";
-    ServiceAccountAuthCredentials credentialsMock =
-        EasyMock.createMock(ServiceAccountAuthCredentials.class);
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.expect(optionsMock.authCredentials()).andReturn(credentialsMock).times(2);
-    EasyMock.expect(optionsMock.clock()).andReturn(TIME_SOURCE);
-    EasyMock.expect(credentialsMock.privateKey()).andReturn(privateKey);
-    EasyMock.expect(credentialsMock.account()).andReturn(account);
-    EasyMock.replay(optionsMock, storageRpcMock, credentialsMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    ServiceAccountAuthCredentials authCredentials =
+        ServiceAccountAuthCredentials.createFor(ACCOUNT, privateKey);
+    storage = options.toBuilder().authCredentials(authCredentials).build().service();
     URL url =
         storage.signUrl(BLOB_INFO1, 14, TimeUnit.DAYS,
             Storage.SignUrlOption.httpMethod(HttpMethod.POST),
@@ -882,7 +815,7 @@ public void testSignUrlWithOptions() throws NoSuchAlgorithmException, InvalidKey
     String stringUrl = url.toString();
     String expectedUrl =
         new StringBuilder("https://storage.googleapis.com/").append(BUCKET_NAME1).append("/")
-            .append(BLOB_NAME1).append("?GoogleAccessId=").append(account).append("&Expires=")
+            .append(BLOB_NAME1).append("?GoogleAccessId=").append(ACCOUNT).append("&Expires=")
             .append(42L + 1209600).append("&Signature=").toString();
     assertTrue(stringUrl.startsWith(expectedUrl));
     String signature = stringUrl.substring(expectedUrl.length());
@@ -897,7 +830,6 @@ public void testSignUrlWithOptions() throws NoSuchAlgorithmException, InvalidKey
     signer.update(signedMessageBuilder.toString().getBytes(UTF_8));
     assertTrue(signer.verify(BaseEncoding.base64().decode(
         URLDecoder.decode(signature, UTF_8.name()))));
-    EasyMock.verify(credentialsMock);
   }
 
   @Test
@@ -920,11 +852,11 @@ public Tuple<StorageObject, StorageException> apply(StorageObject f) {
     StorageRpc.BatchResponse res =
         new StorageRpc.BatchResponse(deleteResult, updateResult, getResult);
 
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
+
     Capture<StorageRpc.BatchRequest> capturedBatchRequest = Capture.newInstance();
     EasyMock.expect(storageRpcMock.batch(EasyMock.capture(capturedBatchRequest))).andReturn(res);
-    EasyMock.replay(optionsMock, storageRpcMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    storage = options.service();
     List<BlobInfo> resultBlobs = storage.get(blobId1, blobId2);
 
     // Verify captured StorageRpc.BatchRequest
@@ -963,11 +895,11 @@ public Tuple<StorageObject, StorageException> apply(StorageObject f) {
     StorageRpc.BatchResponse res =
         new StorageRpc.BatchResponse(deleteResult, updateResult, getResult);
 
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
+
     Capture<StorageRpc.BatchRequest> capturedBatchRequest = Capture.newInstance();
     EasyMock.expect(storageRpcMock.batch(EasyMock.capture(capturedBatchRequest))).andReturn(res);
-    EasyMock.replay(optionsMock, storageRpcMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    storage = options.service();
     List<BlobInfo> resultBlobs = storage.update(blobInfo1, blobInfo2);
 
     // Verify captured StorageRpc.BatchRequest
@@ -1006,11 +938,10 @@ public Tuple<Boolean, StorageException> apply(StorageObject f) {
     StorageRpc.BatchResponse res =
         new StorageRpc.BatchResponse(deleteResult, updateResult, getResult);
 
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
     Capture<StorageRpc.BatchRequest> capturedBatchRequest = Capture.newInstance();
     EasyMock.expect(storageRpcMock.batch(EasyMock.capture(capturedBatchRequest))).andReturn(res);
-    EasyMock.replay(optionsMock, storageRpcMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    storage = options.service();
     List<Boolean> deleteResults = storage.delete(blobInfo1.blobId(), blobInfo2.blobId());
 
     // Verify captured StorageRpc.BatchRequest
@@ -1032,13 +963,11 @@ public Tuple<Boolean, StorageException> apply(StorageObject f) {
   @Test
   public void testRetryableException() {
     BlobId blob = BlobId.of(BUCKET_NAME1, BLOB_NAME1);
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.getDefaultInstance());
     EasyMock.expect(storageRpcMock.get(blob.toPb(), EMPTY_RPC_OPTIONS))
         .andThrow(new StorageException(500, "InternalError", true))
         .andReturn(BLOB_INFO1.toPb());
-    EasyMock.replay(optionsMock, storageRpcMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    storage = options.toBuilder().retryParams(RetryParams.getDefaultInstance()).build().service();
     BlobInfo readBlob = storage.get(blob);
     assertEquals(BLOB_INFO1, readBlob);
   }
@@ -1047,12 +976,10 @@ public void testRetryableException() {
   public void testNonRetryableException() {
     BlobId blob = BlobId.of(BUCKET_NAME1, BLOB_NAME1);
     String exceptionMessage = "Not Implemented";
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.getDefaultInstance());
     EasyMock.expect(storageRpcMock.get(blob.toPb(), EMPTY_RPC_OPTIONS))
         .andThrow(new StorageException(501, exceptionMessage, false));
-    EasyMock.replay(optionsMock, storageRpcMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    storage = options.toBuilder().retryParams(RetryParams.getDefaultInstance()).build().service();
     thrown.expect(StorageException.class);
     thrown.expectMessage(exceptionMessage);
     storage.get(blob);
@@ -1062,12 +989,10 @@ public void testNonRetryableException() {
   public void testRuntimeException() {
     BlobId blob = BlobId.of(BUCKET_NAME1, BLOB_NAME1);
     String exceptionMessage = "Artificial runtime exception";
-    EasyMock.expect(optionsMock.storageRpc()).andReturn(storageRpcMock);
-    EasyMock.expect(optionsMock.retryParams()).andReturn(RetryParams.getDefaultInstance());
     EasyMock.expect(storageRpcMock.get(blob.toPb(), EMPTY_RPC_OPTIONS))
         .andThrow(new RuntimeException(exceptionMessage));
-    EasyMock.replay(optionsMock, storageRpcMock);
-    storage = StorageFactory.instance().get(optionsMock);
+    EasyMock.replay(storageRpcMock);
+    storage = options.toBuilder().retryParams(RetryParams.getDefaultInstance()).build().service();
     thrown.expect(StorageException.class);
     thrown.expectMessage(exceptionMessage);
     storage.get(blob);
diff --git a/src/site/resources/index.html b/src/site/resources/index.html
index 4daf4fa596f4..25c769db2fbf 100644
--- a/src/site/resources/index.html
+++ b/src/site/resources/index.html
@@ -118,7 +118,7 @@ <h3 class="block-title">What is it?</h3>
 		    integrates better with your codebase.
 		    All this means you spend more time creating code that matters
 		    to you.</p>
-		  
+
 		  <p><code>gcloud</code> is configured to access Google Cloud Platform
 		    services and authorize (OAuth 2.0) automatically on your behalf.
 		    Add the <code>gcloud</code> dependency to your project and get a private key to be
@@ -126,7 +126,7 @@ <h3 class="block-title">What is it?</h3>
 		    Compute Engine instance, the private key is automatically detected.
                 </div>
 
-                <div class="quote-box--supplementary">		  
+                <div class="quote-box--supplementary">
                   <h4>Example: Retrieve Datastore Entries</h4>
 
                     <div class="code-sample-tabs" ng-init="selected = 'compute engine'">
@@ -136,7 +136,6 @@ <h4>Example: Retrieve Datastore Entries</h4>
 
                     <div hljs="" language="java" ng-show="selected == 'compute engine'" class=""><pre>
 <code class="hljs java">import com.google.gcloud.datastore.Datastore;
-import com.google.gcloud.datastore.DatastoreFactory;
 import com.google.gcloud.datastore.DatastoreOptions;
 import com.google.gcloud.datastore.Entity;
 import com.google.gcloud.datastore.Key;
@@ -145,7 +144,7 @@ <h4>Example: Retrieve Datastore Entries</h4>
 // Authentication is automatic inside Google Compute Engine
 // and Google App Engine.
 DatastoreOptions options = DatastoreOptions.builder().build();
-Datastore datastore = DatastoreFactory.instance().get(options);
+Datastore datastore = options.service();
 KeyFactory keyFactory = datastore.newKeyFactory().kind(<span class="hljs-string">KIND</span>);
 Key key = keyFactory.newKey(keyName);
 Entity entity = datastore.get(key);
@@ -156,7 +155,6 @@ <h4>Example: Retrieve Datastore Entries</h4>
                     <div hljs="" language="java" ng-show="selected == 'elsewhere'" class=""><pre>
 <code class="hljs java">import com.google.gcloud.AuthCredentials;
 import com.google.gcloud.datastore.Datastore;
-import com.google.gcloud.datastore.DatastoreFactory;
 import com.google.gcloud.datastore.DatastoreOptions;
 import com.google.gcloud.datastore.Entity;
 import com.google.gcloud.datastore.Key;
@@ -166,7 +164,7 @@ <h4>Example: Retrieve Datastore Entries</h4>
   .projectId(<span class="hljs-string">PROJECT_ID</span>)
   .authCredentials(AuthCredentials.createForJson(
     new FileInputStream(<span class="hljs-string">PATH_TO_JSON_KEY</span>))).build();
-Datastore datastore = DatastoreFactory.instance().get(options);
+Datastore datastore = options.service();
 KeyFactory keyFactory = datastore.newKeyFactory().kind(<span class="hljs-string">KIND</span>);
 Key key = keyFactory.newKey(keyName);
 Entity entity = datastore.get(key);