Repository: googleapis/java-storage
Pull request #1546 (2022-08-04 22:28:27)

chore: introduce new UnifiedOpt internal api
Attempt to create a single unified api for the various types of options
to remove the need for cross translation, duplication and extraneous
conflation.

*NOTE* For now this api is internal only, however when a binary
compatibility boundary is able to be crossed we could make parts of it
public and replace the existing Option classes.

## New Approach

The general idea of the new approach is one of encapsulated
transformation to specific types of containers.

When handling an option, we need to take into consideration at least
each of the following:
1. Apiary client
2. com.google.cloud.storage.* Model classes
3. gRPC client
4. Maintainability of Options
5. Cognitive load keeping track of options
6. Lifecycle of instances of Options

In particular, 4 and 5 above have been non-trivial to us and customers
both in the past. Specifically, there are three different classes
 related to reading a blob (`Storage.BlobSourceOption`,
`Storage.BlobGetOption`, `Blob.BlobSourceOption`) and three different
classes related to writing/updating a blob. Often highly duplicated, but
not actually commonly defined/maintained. Additionally, each kind of
list operation has it's own Option, duplicated again.

This new change leverages the fact that multiple inheritance of java
interfaces is possible. As such, we can define a set of interfaces which
relate to different resource types (Bucket, Blob, HmacKey) and interfaces
which relate to operation types (Source, Target, List) and applies these
interfaces to those options for which they apply.

Then, we can define the necessary transformations an Option needs to
provide in order to operate in the above circumstances.

### Implementation
Each parameter from https://cloud.google.com/storage/docs/json_api/v1/parameters
has a corresponding concrete class definition in UnifiedOpts. Each class
binds a provided value to the corresponding StorageRpc.Option when applicable.

Some "Options" are actually mutations of BlobInfo before it is translated
to its transport specific representation. Several of these are confusing,
and are present for the sake of backward compatibility but should be
removed/excluded from any future public API modification.

Some "Options" are actually value extractors from a BlobId/BlobInfo/BucketInfo
several of these are confusing, and can even be mutually exclusive with
the other behavior of the operation (performing an object get which extracts
the generation value adding `ifGenerationNotMatch` sending a request to GCS
which can never succeed).

#### Serialization
All existing Options are Serializable meaning our new Opt classes also need
to be Serializable. Each Opt property defines equals/hashCode along with
serialVersionUID in order to provide compatibility.

#### ProjectAsSource
Some options - Generation, Metageneration - have corresponding "SOURCE"
versions. These are important in the case of operations like rewrite. As such,
an interface `ProjectAsSource` has been introduced which can be applied to an
Opt to generate the corresponding "SOURCE" Opt when necessary.

#### Extractors
Parameter extractors are implemented as an Opt which produces another Opt
when provided with an instance of some Model class. For example,
`MetagenerationMatchExtractor` will produce an instance of
`MetagenerationMatch` when resolved against a `BlobInfo` or `BucketInfo`.

Each extractor is itself a singleton, as they are "pure" and always produce
a value relative to the input provided to one of its method invocations.
To ensure each extractor is a singleton, integration with Java Serialization
has been done via `readResolve()` which will return the singleton instance
when attempting to deserialize.

### Binary Compatibility
Leave all existing Option classes exposed as they are, however each class
will function as a facade for one of the Opt instance for the
corresponding new Unified Opt.

### Implementation Compatibility
Several of the existing Option implementations actually perform extractions
or mutations to instances of BucketInfo or BlobInfo between method invocation.
Some of these have been preserved, but are also marked `@Deprecated` where
these Options can lead to confusing semantics (BlobId with generation set,
then also specifying generationNotMatch()). Removing these confusing Options
will require a binary break and updated docs/samples to show customers how
they can accomplish a clear semantic operation.

There is a rough new multi-phase pattern used with the new options and flows
like this:
1. Create an new instance of Opts from the existing legacy Option classes
2. With the result from 1, resolve any extractor Opt to it's corresponding
   concrete instance
3. If applicable, apply any modification back to the "Info" class
   (This should be removed in the future, but is the current way someone can
   define md5, crc32c or automatic content type based on object name)
4. Have the result from 2 create a Map<StorageRpc.Option, ?> which can be
   passed to StorageRpc methods
5. For gRPC implementation, instead of using the map like in 4, we can
   define mapping methods to operate on the respective request builder


## Impacted Code changes

1. Completely gut com.google.cloud.storage.Option, this class is left only
   for binary compatibility reasons.
2. Move behavior of clearing md5 and crc32c from a BlobInfo out of translate
   and into each applicable method. This is done for behavioral compatibility
   and should be removed at such time as we can release breaking changes.
   * If a customer sets their md5/crc32c before attempting a create or write
     the value will be removed, only added back if a corresponding `md5Match`
     or `crc32cMatch` is provided.
3. Many many test were reaching down into Option fields. Each test doing this
   has been updated to instead provide the key/value they are validating
   explicitly.
4. In `Bucket` and `Blob` some operations were mutating the respective Info
   before passing it to `Storage`, this has been removed so that any Option
   processing logic which can modify an Info is centralized in the `Storage`
   instance.
5. In `Bucket` and `Blob` some level of Option validation was being performed
   before calling the `Storage` operation, this has been removed so that
   all Option validation can be performed in the `Storage` instance.

### Test Cleanup

1. Cleanup Bucket related tests which were hardcoded to expect modifying
   generation/metageneration of BlobInfo before passed to storage
2. Delete several test from BucketTest which were performing early validation
   unnecessarily.
3. Delete MANY unused constants from various test classes



Patch:
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/Blob.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Blob.java
index 08ff2db1..161925b7 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/Blob.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/Blob.java
@@ -16,8 +16,6 @@
 
 package com.google.cloud.storage;
 
-import static com.google.cloud.storage.Blob.BlobSourceOption.toGetOptions;
-import static com.google.cloud.storage.Blob.BlobSourceOption.toSourceOptions;
 import static com.google.common.base.Preconditions.checkNotNull;
 
 import com.google.auth.ServiceAccountSigner;
@@ -25,13 +23,14 @@ import com.google.auth.ServiceAccountSigner.SigningException;
 import com.google.cloud.ReadChannel;
 import com.google.cloud.WriteChannel;
 import com.google.cloud.storage.Acl.Entity;
+import com.google.cloud.storage.Storage.BlobGetOption;
 import com.google.cloud.storage.Storage.BlobTargetOption;
 import com.google.cloud.storage.Storage.BlobWriteOption;
 import com.google.cloud.storage.Storage.CopyRequest;
 import com.google.cloud.storage.Storage.SignUrlOption;
 import com.google.cloud.storage.TransportCompatibility.Transport;
-import com.google.cloud.storage.spi.v1.StorageRpc;
-import com.google.common.io.BaseEncoding;
+import com.google.cloud.storage.UnifiedOpts.ObjectOptExtractor;
+import com.google.cloud.storage.UnifiedOpts.ObjectSourceOpt;
 import java.io.IOException;
 import java.io.ObjectInputStream;
 import java.io.OutputStream;
@@ -75,58 +74,12 @@ public class Blob extends BlobInfo {
   private transient Storage storage;
 
   /** Class for specifying blob source options when {@code Blob} methods are used. */
-  public static class BlobSourceOption extends Option {
+  public static class BlobSourceOption extends Option<ObjectSourceOpt> {
 
     private static final long serialVersionUID = 214616862061934846L;
 
-    private BlobSourceOption(StorageRpc.Option rpcOption) {
-      super(rpcOption, null);
-    }
-
-    private BlobSourceOption(StorageRpc.Option rpcOption, Object value) {
-      super(rpcOption, value);
-    }
-
-    private Storage.BlobSourceOption toSourceOptions(BlobInfo blobInfo) {
-      switch (getRpcOption()) {
-        case IF_GENERATION_MATCH:
-          return Storage.BlobSourceOption.generationMatch(blobInfo.getGeneration());
-        case IF_GENERATION_NOT_MATCH:
-          return Storage.BlobSourceOption.generationNotMatch(blobInfo.getGeneration());
-        case IF_METAGENERATION_MATCH:
-          return Storage.BlobSourceOption.metagenerationMatch(blobInfo.getMetageneration());
-        case IF_METAGENERATION_NOT_MATCH:
-          return Storage.BlobSourceOption.metagenerationNotMatch(blobInfo.getMetageneration());
-        case CUSTOMER_SUPPLIED_KEY:
-          return Storage.BlobSourceOption.decryptionKey((String) getValue());
-        case USER_PROJECT:
-          return Storage.BlobSourceOption.userProject((String) getValue());
-        case RETURN_RAW_INPUT_STREAM:
-          return Storage.BlobSourceOption.shouldReturnRawInputStream((boolean) getValue());
-        default:
-          throw new AssertionError("Unexpected enum value");
-      }
-    }
-
-    private Storage.BlobGetOption toGetOption(BlobInfo blobInfo) {
-      switch (getRpcOption()) {
-        case IF_GENERATION_MATCH:
-          return Storage.BlobGetOption.generationMatch(blobInfo.getGeneration());
-        case IF_GENERATION_NOT_MATCH:
-          return Storage.BlobGetOption.generationNotMatch(blobInfo.getGeneration());
-        case IF_METAGENERATION_MATCH:
-          return Storage.BlobGetOption.metagenerationMatch(blobInfo.getMetageneration());
-        case IF_METAGENERATION_NOT_MATCH:
-          return Storage.BlobGetOption.metagenerationNotMatch(blobInfo.getMetageneration());
-        case USER_PROJECT:
-          return Storage.BlobGetOption.userProject((String) getValue());
-        case CUSTOMER_SUPPLIED_KEY:
-          return Storage.BlobGetOption.decryptionKey((String) getValue());
-        case RETURN_RAW_INPUT_STREAM:
-          return Storage.BlobGetOption.shouldReturnRawInputStream((boolean) getValue());
-        default:
-          throw new AssertionError("Unexpected enum value");
-      }
+    private BlobSourceOption(ObjectSourceOpt opt) {
+      super(opt);
     }
 
     /**
@@ -134,7 +87,7 @@ public class Blob extends BlobInfo {
      * if generation does not match.
      */
     public static BlobSourceOption generationMatch() {
-      return new BlobSourceOption(StorageRpc.Option.IF_GENERATION_MATCH);
+      return new BlobSourceOption(UnifiedOpts.generationMatchExtractor());
     }
 
     /**
@@ -145,7 +98,7 @@ public class Blob extends BlobInfo {
      */
     @Deprecated
     public static BlobSourceOption generationNotMatch() {
-      return new BlobSourceOption(StorageRpc.Option.IF_GENERATION_NOT_MATCH);
+      return new BlobSourceOption(UnifiedOpts.generationNotMatchExtractor());
     }
 
     /**
@@ -153,7 +106,7 @@ public class Blob extends BlobInfo {
      * fail if metageneration does not match.
      */
     public static BlobSourceOption metagenerationMatch() {
-      return new BlobSourceOption(StorageRpc.Option.IF_METAGENERATION_MATCH);
+      return new BlobSourceOption(UnifiedOpts.metagenerationMatchExtractor());
     }
 
     /**
@@ -161,7 +114,7 @@ public class Blob extends BlobInfo {
      * fail if metageneration matches.
      */
     public static BlobSourceOption metagenerationNotMatch() {
-      return new BlobSourceOption(StorageRpc.Option.IF_METAGENERATION_NOT_MATCH);
+      return new BlobSourceOption(UnifiedOpts.metagenerationNotMatchExtractor());
     }
 
     /**
@@ -169,8 +122,7 @@ public class Blob extends BlobInfo {
      * blob.
      */
     public static BlobSourceOption decryptionKey(Key key) {
-      String base64Key = BaseEncoding.base64().encode(key.getEncoded());
-      return new BlobSourceOption(StorageRpc.Option.CUSTOMER_SUPPLIED_KEY, base64Key);
+      return new BlobSourceOption(UnifiedOpts.decryptionKey(key));
     }
 
     /**
@@ -180,7 +132,7 @@ public class Blob extends BlobInfo {
      * @param key the AES256 encoded in base64
      */
     public static BlobSourceOption decryptionKey(String key) {
-      return new BlobSourceOption(StorageRpc.Option.CUSTOMER_SUPPLIED_KEY, key);
+      return new BlobSourceOption(UnifiedOpts.decryptionKey(key));
     }
 
     /**
@@ -188,7 +140,7 @@ public class Blob extends BlobInfo {
      * bucket has requester_pays flag enabled.
      */
     public static BlobSourceOption userProject(String userProject) {
-      return new BlobSourceOption(StorageRpc.Option.USER_PROJECT, userProject);
+      return new BlobSourceOption(UnifiedOpts.userProject(userProject));
     }
 
     /**
@@ -197,25 +149,36 @@ public class Blob extends BlobInfo {
      * true for ReadChannel.read().
      */
     public static BlobSourceOption shouldReturnRawInputStream(boolean shouldReturnRawInputStream) {
-      return new BlobSourceOption(
-          StorageRpc.Option.RETURN_RAW_INPUT_STREAM, shouldReturnRawInputStream);
+      return new BlobSourceOption(UnifiedOpts.returnRawInputStream(shouldReturnRawInputStream));
     }
 
     static Storage.BlobSourceOption[] toSourceOptions(
         BlobInfo blobInfo, BlobSourceOption... options) {
       Storage.BlobSourceOption[] convertedOptions = new Storage.BlobSourceOption[options.length];
-      int index = 0;
-      for (BlobSourceOption option : options) {
-        convertedOptions[index++] = option.toSourceOptions(blobInfo);
+      for (int i = 0; i < options.length; i++) {
+        ObjectSourceOpt opt = options[i].getOpt();
+        if (opt instanceof ObjectOptExtractor) {
+          ObjectOptExtractor<ObjectSourceOpt> ex = (ObjectOptExtractor<ObjectSourceOpt>) opt;
+          ObjectSourceOpt objectSourceOpt = ex.extractFromBlobInfo(blobInfo);
+          convertedOptions[i] = new Storage.BlobSourceOption(objectSourceOpt);
+        } else {
+          convertedOptions[i] = new Storage.BlobSourceOption(opt);
+        }
       }
       return convertedOptions;
     }
 
     static Storage.BlobGetOption[] toGetOptions(BlobInfo blobInfo, BlobSourceOption... options) {
       Storage.BlobGetOption[] convertedOptions = new Storage.BlobGetOption[options.length];
-      int index = 0;
-      for (BlobSourceOption option : options) {
-        convertedOptions[index++] = option.toGetOption(blobInfo);
+      for (int i = 0; i < options.length; i++) {
+        ObjectSourceOpt opt = options[i].getOpt();
+        if (opt instanceof ObjectOptExtractor) {
+          ObjectOptExtractor<ObjectSourceOpt> ex = (ObjectOptExtractor<ObjectSourceOpt>) opt;
+          ObjectSourceOpt objectSourceOpt = ex.extractFromBlobInfo(blobInfo);
+          convertedOptions[i] = new BlobGetOption(objectSourceOpt);
+        } else {
+          convertedOptions[i] = new BlobGetOption(options[i].getOpt());
+        }
       }
       return convertedOptions;
     }
@@ -231,7 +194,9 @@ public class Blob extends BlobInfo {
    */
   @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   public void downloadTo(Path path, BlobSourceOption... options) {
-    storage.downloadTo(this.getBlobId(), path, toSourceOptions(this, options));
+    // Don't use static imports of BlobSourceOption, it causes import resolution issues
+    // with the new UnifiedOpts shim interfaces
+    storage.downloadTo(this.getBlobId(), path, BlobSourceOption.toSourceOptions(this, options));
   }
 
   /**
@@ -244,7 +209,10 @@ public class Blob extends BlobInfo {
    */
   @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   public void downloadTo(OutputStream outputStream, BlobSourceOption... options) {
-    storage.downloadTo(this.getBlobId(), outputStream, toSourceOptions(this, options));
+    // Don't use static imports of BlobSourceOption, it causes import resolution issues
+    // with the new UnifiedOpts shim interfaces
+    storage.downloadTo(
+        this.getBlobId(), outputStream, BlobSourceOption.toSourceOptions(this, options));
   }
 
   /**
@@ -546,7 +514,10 @@ public class Blob extends BlobInfo {
   @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   public boolean exists(BlobSourceOption... options) {
     int length = options.length;
-    Storage.BlobGetOption[] getOptions = Arrays.copyOf(toGetOptions(this, options), length + 1);
+    // Don't use static imports of BlobSourceOption, it causes import resolution issues
+    // with the new UnifiedOpts shim interfaces
+    Storage.BlobGetOption[] getOptions =
+        Arrays.copyOf(BlobSourceOption.toGetOptions(this, options), length + 1);
     getOptions[length] = Storage.BlobGetOption.fields();
     return storage.get(getBlobId(), getOptions) != null;
   }
@@ -566,7 +537,9 @@ public class Blob extends BlobInfo {
    */
   @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   public byte[] getContent(BlobSourceOption... options) {
-    return storage.readAllBytes(getBlobId(), toSourceOptions(this, options));
+    // Don't use static imports of BlobSourceOption, it causes import resolution issues
+    // with the new UnifiedOpts shim interfaces
+    return storage.readAllBytes(getBlobId(), BlobSourceOption.toSourceOptions(this, options));
   }
 
   /**
@@ -605,7 +578,9 @@ public class Blob extends BlobInfo {
   public Blob reload(BlobSourceOption... options) {
     // BlobId with generation unset is needed to retrieve the latest version of the Blob
     BlobId idWithoutGeneration = BlobId.of(getBucket(), getName());
-    return storage.get(idWithoutGeneration, toGetOptions(this, options));
+    // Don't use static imports of BlobSourceOption, it causes import resolution issues
+    // with the new UnifiedOpts shim interfaces
+    return storage.get(idWithoutGeneration, BlobSourceOption.toGetOptions(this, options));
   }
 
   /**
@@ -657,7 +632,9 @@ public class Blob extends BlobInfo {
    */
   @TransportCompatibility({Transport.HTTP})
   public boolean delete(BlobSourceOption... options) {
-    return storage.delete(getBlobId(), toSourceOptions(this, options));
+    // Don't use static imports of BlobSourceOption, it causes import resolution issues
+    // with the new UnifiedOpts shim interfaces
+    return storage.delete(getBlobId(), BlobSourceOption.toSourceOptions(this, options));
   }
 
   /**
@@ -684,7 +661,9 @@ public class Blob extends BlobInfo {
     CopyRequest copyRequest =
         CopyRequest.newBuilder()
             .setSource(getBucket(), getName())
-            .setSourceOptions(toSourceOptions(this, options))
+            // Don't use static imports of BlobSourceOption, it causes import resolution issues
+            // with the new UnifiedOpts shim interfaces
+            .setSourceOptions(BlobSourceOption.toSourceOptions(this, options))
             .setTarget(targetBlob)
             .build();
     return storage.copy(copyRequest);
@@ -782,7 +761,9 @@ public class Blob extends BlobInfo {
    */
   @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   public ReadChannel reader(BlobSourceOption... options) {
-    return storage.reader(getBlobId(), toSourceOptions(this, options));
+    // Don't use static imports of BlobSourceOption, it causes import resolution issues
+    // with the new UnifiedOpts shim interfaces
+    return storage.reader(getBlobId(), BlobSourceOption.toSourceOptions(this, options));
   }
 
   /**
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/Bucket.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Bucket.java
index de1d801f..9ffce29f 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/Bucket.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/Bucket.java
@@ -16,24 +16,22 @@
 
 package com.google.cloud.storage;
 
-import static com.google.cloud.storage.Bucket.BucketSourceOption.toGetOptions;
-import static com.google.cloud.storage.Bucket.BucketSourceOption.toSourceOptions;
-import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkNotNull;
 
 import com.google.api.gax.paging.Page;
-import com.google.cloud.Tuple;
 import com.google.cloud.storage.Acl.Entity;
 import com.google.cloud.storage.Storage.BlobGetOption;
 import com.google.cloud.storage.Storage.BlobListOption;
+import com.google.cloud.storage.Storage.BucketGetOption;
 import com.google.cloud.storage.Storage.BucketTargetOption;
 import com.google.cloud.storage.TransportCompatibility.Transport;
-import com.google.cloud.storage.spi.v1.StorageRpc;
-import com.google.common.base.Function;
+import com.google.cloud.storage.UnifiedOpts.BucketOptExtractor;
+import com.google.cloud.storage.UnifiedOpts.BucketSourceOpt;
+import com.google.cloud.storage.UnifiedOpts.ObjectOptExtractor;
+import com.google.cloud.storage.UnifiedOpts.ObjectTargetOpt;
+import com.google.cloud.storage.UnifiedOpts.OptionShim;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Lists;
-import com.google.common.collect.Sets;
-import com.google.common.io.BaseEncoding;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.ObjectInputStream;
@@ -45,7 +43,6 @@ import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
-import java.util.Set;
 
 /**
  * A Google cloud storage bucket.
@@ -62,38 +59,12 @@ public class Bucket extends BucketInfo {
   private transient Storage storage;
 
   /** Class for specifying bucket source options when {@code Bucket} methods are used. */
-  public static class BucketSourceOption extends Option {
+  public static class BucketSourceOption extends Option<BucketSourceOpt> {
 
     private static final long serialVersionUID = 6928872234155522371L;
 
-    private BucketSourceOption(StorageRpc.Option rpcOption) {
-      super(rpcOption, null);
-    }
-
-    private BucketSourceOption(StorageRpc.Option rpcOption, Object value) {
-      super(rpcOption, value);
-    }
-
-    private Storage.BucketSourceOption toSourceOption(BucketInfo bucketInfo) {
-      switch (getRpcOption()) {
-        case IF_METAGENERATION_MATCH:
-          return Storage.BucketSourceOption.metagenerationMatch(bucketInfo.getMetageneration());
-        case IF_METAGENERATION_NOT_MATCH:
-          return Storage.BucketSourceOption.metagenerationNotMatch(bucketInfo.getMetageneration());
-        default:
-          throw new AssertionError("Unexpected enum value");
-      }
-    }
-
-    private Storage.BucketGetOption toGetOption(BucketInfo bucketInfo) {
-      switch (getRpcOption()) {
-        case IF_METAGENERATION_MATCH:
-          return Storage.BucketGetOption.metagenerationMatch(bucketInfo.getMetageneration());
-        case IF_METAGENERATION_NOT_MATCH:
-          return Storage.BucketGetOption.metagenerationNotMatch(bucketInfo.getMetageneration());
-        default:
-          throw new AssertionError("Unexpected enum value");
-      }
+    private BucketSourceOption(BucketSourceOpt opt) {
+      super(opt);
     }
 
     /**
@@ -101,7 +72,7 @@ public class Bucket extends BucketInfo {
      * fail if metageneration does not match.
      */
     public static BucketSourceOption metagenerationMatch() {
-      return new BucketSourceOption(StorageRpc.Option.IF_METAGENERATION_MATCH);
+      return new BucketSourceOption(UnifiedOpts.metagenerationMatchExtractor());
     }
 
     /**
@@ -109,7 +80,7 @@ public class Bucket extends BucketInfo {
      * will fail if metageneration matches.
      */
     public static BucketSourceOption metagenerationNotMatch() {
-      return new BucketSourceOption(StorageRpc.Option.IF_METAGENERATION_NOT_MATCH);
+      return new BucketSourceOption(UnifiedOpts.metagenerationNotMatchExtractor());
     }
 
     /**
@@ -117,16 +88,22 @@ public class Bucket extends BucketInfo {
      * with 'requester_pays' flag.
      */
     public static BucketSourceOption userProject(String userProject) {
-      return new BucketSourceOption(StorageRpc.Option.USER_PROJECT, userProject);
+      return new BucketSourceOption(UnifiedOpts.userProject(userProject));
     }
 
     static Storage.BucketSourceOption[] toSourceOptions(
         BucketInfo bucketInfo, BucketSourceOption... options) {
       Storage.BucketSourceOption[] convertedOptions =
           new Storage.BucketSourceOption[options.length];
-      int index = 0;
-      for (BucketSourceOption option : options) {
-        convertedOptions[index++] = option.toSourceOption(bucketInfo);
+      for (int i = 0; i < options.length; i++) {
+        BucketSourceOpt opt = options[i].getOpt();
+        if (opt instanceof BucketOptExtractor) {
+          BucketOptExtractor<BucketSourceOpt> ex = (BucketOptExtractor<BucketSourceOpt>) opt;
+          BucketSourceOpt bucketSourceOpt = ex.extractFromBucketInfo(bucketInfo);
+          convertedOptions[i] = new Storage.BucketSourceOption(bucketSourceOpt);
+        } else {
+          convertedOptions[i] = new Storage.BucketSourceOption(options[i].getOpt());
+        }
       }
       return convertedOptions;
     }
@@ -134,68 +111,32 @@ public class Bucket extends BucketInfo {
     static Storage.BucketGetOption[] toGetOptions(
         BucketInfo bucketInfo, BucketSourceOption... options) {
       Storage.BucketGetOption[] convertedOptions = new Storage.BucketGetOption[options.length];
-      int index = 0;
-      for (BucketSourceOption option : options) {
-        convertedOptions[index++] = option.toGetOption(bucketInfo);
+      for (int i = 0; i < options.length; i++) {
+        BucketSourceOpt opt = options[i].getOpt();
+        if (opt instanceof BucketOptExtractor) {
+          BucketOptExtractor<BucketSourceOpt> ex = (BucketOptExtractor<BucketSourceOpt>) opt;
+          BucketSourceOpt bucketSourceOpt = ex.extractFromBucketInfo(bucketInfo);
+          convertedOptions[i] = new BucketGetOption(bucketSourceOpt);
+        } else {
+          convertedOptions[i] = new BucketGetOption(options[i].getOpt());
+        }
       }
       return convertedOptions;
     }
   }
 
   /** Class for specifying blob target options when {@code Bucket} methods are used. */
-  public static class BlobTargetOption extends Option {
-
-    private static final Function<BlobTargetOption, StorageRpc.Option> TO_ENUM =
-        new Function<BlobTargetOption, StorageRpc.Option>() {
-          @Override
-          public StorageRpc.Option apply(BlobTargetOption blobTargetOption) {
-            return blobTargetOption.getRpcOption();
-          }
-        };
+  public static class BlobTargetOption extends Option<ObjectTargetOpt> {
+
     private static final long serialVersionUID = 8345296337342509425L;
 
-    private BlobTargetOption(StorageRpc.Option rpcOption, Object value) {
-      super(rpcOption, value);
-    }
-
-    private Tuple<BlobInfo, Storage.BlobTargetOption> toTargetOption(BlobInfo blobInfo) {
-      BlobId blobId = blobInfo.getBlobId();
-      switch (getRpcOption()) {
-        case PREDEFINED_ACL:
-          return Tuple.of(
-              blobInfo, Storage.BlobTargetOption.predefinedAcl((Storage.PredefinedAcl) getValue()));
-        case IF_GENERATION_MATCH:
-          blobId = BlobId.of(blobId.getBucket(), blobId.getName(), (Long) getValue());
-          return Tuple.of(
-              blobInfo.toBuilder().setBlobId(blobId).build(),
-              Storage.BlobTargetOption.generationMatch());
-        case IF_GENERATION_NOT_MATCH:
-          blobId = BlobId.of(blobId.getBucket(), blobId.getName(), (Long) getValue());
-          return Tuple.of(
-              blobInfo.toBuilder().setBlobId(blobId).build(),
-              Storage.BlobTargetOption.generationNotMatch());
-        case IF_METAGENERATION_MATCH:
-          return Tuple.of(
-              blobInfo.toBuilder().setMetageneration((Long) getValue()).build(),
-              Storage.BlobTargetOption.metagenerationMatch());
-        case IF_METAGENERATION_NOT_MATCH:
-          return Tuple.of(
-              blobInfo.toBuilder().setMetageneration((Long) getValue()).build(),
-              Storage.BlobTargetOption.metagenerationNotMatch());
-        case CUSTOMER_SUPPLIED_KEY:
-          return Tuple.of(blobInfo, Storage.BlobTargetOption.encryptionKey((String) getValue()));
-        case KMS_KEY_NAME:
-          return Tuple.of(blobInfo, Storage.BlobTargetOption.kmsKeyName((String) getValue()));
-        case USER_PROJECT:
-          return Tuple.of(blobInfo, Storage.BlobTargetOption.userProject((String) getValue()));
-        default:
-          throw new AssertionError("Unexpected enum value");
-      }
+    private BlobTargetOption(ObjectTargetOpt opt) {
+      super(opt);
     }
 
     /** Returns an option for specifying blob's predefined ACL configuration. */
     public static BlobTargetOption predefinedAcl(Storage.PredefinedAcl acl) {
-      return new BlobTargetOption(StorageRpc.Option.PREDEFINED_ACL, acl);
+      return new BlobTargetOption(UnifiedOpts.predefinedAcl(acl));
     }
 
     /**
@@ -204,7 +145,7 @@ public class Bucket extends BucketInfo {
      * #generationNotMatch(long)}.
      */
     public static BlobTargetOption doesNotExist() {
-      return new BlobTargetOption(StorageRpc.Option.IF_GENERATION_MATCH, 0L);
+      return new BlobTargetOption(UnifiedOpts.doesNotExist());
     }
 
     /**
@@ -213,7 +154,7 @@ public class Bucket extends BucketInfo {
      * together with {@link #generationNotMatch(long)} or {@link #doesNotExist()}.
      */
     public static BlobTargetOption generationMatch(long generation) {
-      return new BlobTargetOption(StorageRpc.Option.IF_GENERATION_MATCH, generation);
+      return new BlobTargetOption(UnifiedOpts.generationMatch(generation));
     }
 
     /**
@@ -222,7 +163,7 @@ public class Bucket extends BucketInfo {
      * together with {@link #generationMatch(long)} or {@link #doesNotExist()}.
      */
     public static BlobTargetOption generationNotMatch(long generation) {
-      return new BlobTargetOption(StorageRpc.Option.IF_GENERATION_NOT_MATCH, generation);
+      return new BlobTargetOption(UnifiedOpts.generationNotMatch(generation));
     }
 
     /**
@@ -231,7 +172,7 @@ public class Bucket extends BucketInfo {
      * together with {@link #metagenerationNotMatch(long)}.
      */
     public static BlobTargetOption metagenerationMatch(long metageneration) {
-      return new BlobTargetOption(StorageRpc.Option.IF_METAGENERATION_MATCH, metageneration);
+      return new BlobTargetOption(UnifiedOpts.metagenerationMatch(metageneration));
     }
 
     /**
@@ -240,7 +181,7 @@ public class Bucket extends BucketInfo {
      * with {@link #metagenerationMatch(long)}.
      */
     public static BlobTargetOption metagenerationNotMatch(long metageneration) {
-      return new BlobTargetOption(StorageRpc.Option.IF_METAGENERATION_NOT_MATCH, metageneration);
+      return new BlobTargetOption(UnifiedOpts.metagenerationNotMatch(metageneration));
     }
 
     /**
@@ -248,8 +189,7 @@ public class Bucket extends BucketInfo {
      * blob.
      */
     public static BlobTargetOption encryptionKey(Key key) {
-      String base64Key = BaseEncoding.base64().encode(key.getEncoded());
-      return new BlobTargetOption(StorageRpc.Option.CUSTOMER_SUPPLIED_KEY, base64Key);
+      return new BlobTargetOption(UnifiedOpts.encryptionKey(key));
     }
 
     /**
@@ -259,7 +199,7 @@ public class Bucket extends BucketInfo {
      * @param key the AES256 encoded in base64
      */
     public static BlobTargetOption encryptionKey(String key) {
-      return new BlobTargetOption(StorageRpc.Option.CUSTOMER_SUPPLIED_KEY, key);
+      return new BlobTargetOption(UnifiedOpts.encryptionKey(key));
     }
 
     /**
@@ -268,7 +208,7 @@ public class Bucket extends BucketInfo {
      * @param kmsKeyName the KMS key resource id
      */
     public static BlobTargetOption kmsKeyName(String kmsKeyName) {
-      return new BlobTargetOption(StorageRpc.Option.KMS_KEY_NAME, kmsKeyName);
+      return new BlobTargetOption(UnifiedOpts.kmsKeyName(kmsKeyName));
     }
 
     /**
@@ -276,116 +216,38 @@ public class Bucket extends BucketInfo {
      * with 'requester_pays' flag.
      */
     public static BlobTargetOption userProject(String userProject) {
-      return new BlobTargetOption(StorageRpc.Option.USER_PROJECT, userProject);
-    }
-
-    static Tuple<BlobInfo, Storage.BlobTargetOption[]> toTargetOptions(
-        BlobInfo info, BlobTargetOption... options) {
-      Set<StorageRpc.Option> optionSet =
-          Sets.immutableEnumSet(Lists.transform(Arrays.asList(options), TO_ENUM));
-      checkArgument(
-          !(optionSet.contains(StorageRpc.Option.IF_METAGENERATION_NOT_MATCH)
-              && optionSet.contains(StorageRpc.Option.IF_METAGENERATION_MATCH)),
-          "metagenerationMatch and metagenerationNotMatch options can not be both provided");
-      checkArgument(
-          !(optionSet.contains(StorageRpc.Option.IF_GENERATION_NOT_MATCH)
-              && optionSet.contains(StorageRpc.Option.IF_GENERATION_MATCH)),
-          "Only one option of generationMatch, doesNotExist or generationNotMatch can be provided");
-      Storage.BlobTargetOption[] convertedOptions = new Storage.BlobTargetOption[options.length];
-      BlobInfo targetInfo = info;
-      int index = 0;
-      for (BlobTargetOption option : options) {
-        Tuple<BlobInfo, Storage.BlobTargetOption> target = option.toTargetOption(targetInfo);
-        targetInfo = target.x();
-        convertedOptions[index++] = target.y();
+      return new BlobTargetOption(UnifiedOpts.userProject(userProject));
+    }
+
+    static Storage.BlobTargetOption[] toTargetOptions(
+        BlobInfo blobInfo, BlobTargetOption... options) {
+      Storage.BlobTargetOption[] targetOptions = new Storage.BlobTargetOption[options.length];
+      for (int i = 0; i < options.length; i++) {
+        ObjectTargetOpt opt = options[i].getOpt();
+        if (opt instanceof ObjectOptExtractor) {
+          ObjectOptExtractor<ObjectTargetOpt> ex = (ObjectOptExtractor<ObjectTargetOpt>) opt;
+          ObjectTargetOpt objectTargetOpt = ex.extractFromBlobInfo(blobInfo);
+          targetOptions[i] = new Storage.BlobTargetOption(objectTargetOpt);
+        } else {
+          targetOptions[i] = new Storage.BlobTargetOption(options[i].getOpt());
+        }
       }
-      return Tuple.of(targetInfo, convertedOptions);
+      return targetOptions;
     }
   }
 
   /** Class for specifying blob write options when {@code Bucket} methods are used. */
-  public static class BlobWriteOption implements Serializable {
-
-    private static final Function<BlobWriteOption, Storage.BlobWriteOption.Option> TO_ENUM =
-        new Function<BlobWriteOption, Storage.BlobWriteOption.Option>() {
-          @Override
-          public Storage.BlobWriteOption.Option apply(BlobWriteOption blobWriteOption) {
-            return blobWriteOption.option;
-          }
-        };
-    private static final long serialVersionUID = 4722190734541993114L;
-
-    private final Storage.BlobWriteOption.Option option;
-    private final Object value;
-
-    private Tuple<BlobInfo, Storage.BlobWriteOption> toWriteOption(BlobInfo blobInfo) {
-      BlobId blobId = blobInfo.getBlobId();
-      switch (option) {
-        case PREDEFINED_ACL:
-          return Tuple.of(
-              blobInfo, Storage.BlobWriteOption.predefinedAcl((Storage.PredefinedAcl) value));
-        case IF_GENERATION_MATCH:
-          blobId = BlobId.of(blobId.getBucket(), blobId.getName(), (Long) value);
-          return Tuple.of(
-              blobInfo.toBuilder().setBlobId(blobId).build(),
-              Storage.BlobWriteOption.generationMatch());
-        case IF_GENERATION_NOT_MATCH:
-          blobId = BlobId.of(blobId.getBucket(), blobId.getName(), (Long) value);
-          return Tuple.of(
-              blobInfo.toBuilder().setBlobId(blobId).build(),
-              Storage.BlobWriteOption.generationNotMatch());
-        case IF_METAGENERATION_MATCH:
-          return Tuple.of(
-              blobInfo.toBuilder().setMetageneration((Long) value).build(),
-              Storage.BlobWriteOption.metagenerationMatch());
-        case IF_METAGENERATION_NOT_MATCH:
-          return Tuple.of(
-              blobInfo.toBuilder().setMetageneration((Long) value).build(),
-              Storage.BlobWriteOption.metagenerationNotMatch());
-        case IF_MD5_MATCH:
-          return Tuple.of(
-              blobInfo.toBuilder().setMd5((String) value).build(),
-              Storage.BlobWriteOption.md5Match());
-        case IF_CRC32C_MATCH:
-          return Tuple.of(
-              blobInfo.toBuilder().setCrc32c((String) value).build(),
-              Storage.BlobWriteOption.crc32cMatch());
-        case CUSTOMER_SUPPLIED_KEY:
-          return Tuple.of(blobInfo, Storage.BlobWriteOption.encryptionKey((String) value));
-        case KMS_KEY_NAME:
-          return Tuple.of(blobInfo, Storage.BlobWriteOption.kmsKeyName((String) value));
-        case USER_PROJECT:
-          return Tuple.of(blobInfo, Storage.BlobWriteOption.userProject((String) value));
-        default:
-          throw new AssertionError("Unexpected enum value");
-      }
-    }
+  public static class BlobWriteOption extends OptionShim<ObjectTargetOpt> implements Serializable {
 
-    private BlobWriteOption(Storage.BlobWriteOption.Option option, Object value) {
-      this.option = option;
-      this.value = value;
-    }
+    private static final long serialVersionUID = -1103041630932223724L;
 
-    @Override
-    public int hashCode() {
-      return Objects.hash(option, value);
-    }
-
-    @Override
-    public boolean equals(Object obj) {
-      if (obj == null) {
-        return false;
-      }
-      if (!(obj instanceof BlobWriteOption)) {
-        return false;
-      }
-      final BlobWriteOption other = (BlobWriteOption) obj;
-      return this.option == other.option && Objects.equals(this.value, other.value);
+    private BlobWriteOption(ObjectTargetOpt opt) {
+      super(opt);
     }
 
     /** Returns an option for specifying blob's predefined ACL configuration. */
     public static BlobWriteOption predefinedAcl(Storage.PredefinedAcl acl) {
-      return new BlobWriteOption(Storage.BlobWriteOption.Option.PREDEFINED_ACL, acl);
+      return new BlobWriteOption(UnifiedOpts.predefinedAcl(acl));
     }
 
     /**
@@ -394,7 +256,7 @@ public class Bucket extends BucketInfo {
      * #generationNotMatch(long)}.
      */
     public static BlobWriteOption doesNotExist() {
-      return new BlobWriteOption(Storage.BlobWriteOption.Option.IF_GENERATION_MATCH, 0L);
+      return new BlobWriteOption(UnifiedOpts.doesNotExist());
     }
 
     /**
@@ -403,7 +265,7 @@ public class Bucket extends BucketInfo {
      * together with {@link #generationNotMatch(long)} or {@link #doesNotExist()}.
      */
     public static BlobWriteOption generationMatch(long generation) {
-      return new BlobWriteOption(Storage.BlobWriteOption.Option.IF_GENERATION_MATCH, generation);
+      return new BlobWriteOption(UnifiedOpts.generationMatch(generation));
     }
 
     /**
@@ -412,8 +274,7 @@ public class Bucket extends BucketInfo {
      * with {@link #generationMatch(long)} or {@link #doesNotExist()}.
      */
     public static BlobWriteOption generationNotMatch(long generation) {
-      return new BlobWriteOption(
-          Storage.BlobWriteOption.Option.IF_GENERATION_NOT_MATCH, generation);
+      return new BlobWriteOption(UnifiedOpts.generationNotMatch(generation));
     }
 
     /**
@@ -422,8 +283,7 @@ public class Bucket extends BucketInfo {
      * together with {@link #metagenerationNotMatch(long)}.
      */
     public static BlobWriteOption metagenerationMatch(long metageneration) {
-      return new BlobWriteOption(
-          Storage.BlobWriteOption.Option.IF_METAGENERATION_MATCH, metageneration);
+      return new BlobWriteOption(UnifiedOpts.metagenerationMatch(metageneration));
     }
 
     /**
@@ -432,8 +292,7 @@ public class Bucket extends BucketInfo {
      * with {@link #metagenerationMatch(long)}.
      */
     public static BlobWriteOption metagenerationNotMatch(long metageneration) {
-      return new BlobWriteOption(
-          Storage.BlobWriteOption.Option.IF_METAGENERATION_NOT_MATCH, metageneration);
+      return new BlobWriteOption(UnifiedOpts.metagenerationNotMatch(metageneration));
     }
 
     /**
@@ -441,7 +300,7 @@ public class Bucket extends BucketInfo {
      * fail if blobs' data MD5 hash does not match the provided value.
      */
     public static BlobWriteOption md5Match(String md5) {
-      return new BlobWriteOption(Storage.BlobWriteOption.Option.IF_MD5_MATCH, md5);
+      return new BlobWriteOption(UnifiedOpts.md5Match(md5));
     }
 
     /**
@@ -449,7 +308,7 @@ public class Bucket extends BucketInfo {
      * will fail if blobs' data CRC32C checksum does not match the provided value.
      */
     public static BlobWriteOption crc32cMatch(String crc32c) {
-      return new BlobWriteOption(Storage.BlobWriteOption.Option.IF_CRC32C_MATCH, crc32c);
+      return new BlobWriteOption(UnifiedOpts.crc32cMatch(crc32c));
     }
 
     /**
@@ -457,8 +316,7 @@ public class Bucket extends BucketInfo {
      * blob.
      */
     public static BlobWriteOption encryptionKey(Key key) {
-      String base64Key = BaseEncoding.base64().encode(key.getEncoded());
-      return new BlobWriteOption(Storage.BlobWriteOption.Option.CUSTOMER_SUPPLIED_KEY, base64Key);
+      return new BlobWriteOption(UnifiedOpts.encryptionKey(key));
     }
 
     /**
@@ -468,7 +326,7 @@ public class Bucket extends BucketInfo {
      * @param key the AES256 encoded in base64
      */
     public static BlobWriteOption encryptionKey(String key) {
-      return new BlobWriteOption(Storage.BlobWriteOption.Option.CUSTOMER_SUPPLIED_KEY, key);
+      return new BlobWriteOption(UnifiedOpts.encryptionKey(key));
     }
 
     /**
@@ -476,30 +334,22 @@ public class Bucket extends BucketInfo {
      * with 'requester_pays' flag.
      */
     public static BlobWriteOption userProject(String userProject) {
-      return new BlobWriteOption(Storage.BlobWriteOption.Option.USER_PROJECT, userProject);
-    }
-
-    static Tuple<BlobInfo, Storage.BlobWriteOption[]> toWriteOptions(
-        BlobInfo info, BlobWriteOption... options) {
-      Set<Storage.BlobWriteOption.Option> optionSet =
-          Sets.immutableEnumSet(Lists.transform(Arrays.asList(options), TO_ENUM));
-      checkArgument(
-          !(optionSet.contains(Storage.BlobWriteOption.Option.IF_METAGENERATION_NOT_MATCH)
-              && optionSet.contains(Storage.BlobWriteOption.Option.IF_METAGENERATION_MATCH)),
-          "metagenerationMatch and metagenerationNotMatch options can not be both provided");
-      checkArgument(
-          !(optionSet.contains(Storage.BlobWriteOption.Option.IF_GENERATION_NOT_MATCH)
-              && optionSet.contains(Storage.BlobWriteOption.Option.IF_GENERATION_MATCH)),
-          "Only one option of generationMatch, doesNotExist or generationNotMatch can be provided");
+      return new BlobWriteOption(UnifiedOpts.userProject(userProject));
+    }
+
+    static Storage.BlobWriteOption[] toWriteOptions(BlobInfo blobInfo, BlobWriteOption... options) {
       Storage.BlobWriteOption[] convertedOptions = new Storage.BlobWriteOption[options.length];
-      BlobInfo writeInfo = info;
-      int index = 0;
-      for (BlobWriteOption option : options) {
-        Tuple<BlobInfo, Storage.BlobWriteOption> write = option.toWriteOption(writeInfo);
-        writeInfo = write.x();
-        convertedOptions[index++] = write.y();
+      for (int i = 0; i < options.length; i++) {
+        ObjectTargetOpt opt = options[i].getOpt();
+        if (opt instanceof ObjectOptExtractor) {
+          ObjectOptExtractor<ObjectTargetOpt> ex = (ObjectOptExtractor<ObjectTargetOpt>) opt;
+          ObjectTargetOpt objectTargetOpt = ex.extractFromBlobInfo(blobInfo);
+          convertedOptions[i] = new Storage.BlobWriteOption(objectTargetOpt);
+        } else {
+          convertedOptions[i] = new Storage.BlobWriteOption(options[i].getOpt());
+        }
       }
-      return Tuple.of(writeInfo, convertedOptions);
+      return convertedOptions;
     }
   }
 
@@ -776,7 +626,10 @@ public class Bucket extends BucketInfo {
   @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   public boolean exists(BucketSourceOption... options) {
     int length = options.length;
-    Storage.BucketGetOption[] getOptions = Arrays.copyOf(toGetOptions(this, options), length + 1);
+    // Don't use static imports of BlobSourceOption, it causes import resolution issues
+    // with the new UnifiedOpts shim interfaces
+    Storage.BucketGetOption[] getOptions =
+        Arrays.copyOf(BucketSourceOption.toGetOptions(this, options), length + 1);
     getOptions[length] = Storage.BucketGetOption.fields();
     return storage.get(getName(), getOptions) != null;
   }
@@ -800,7 +653,9 @@ public class Bucket extends BucketInfo {
    */
   @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   public Bucket reload(BucketSourceOption... options) {
-    return storage.get(getName(), toGetOptions(this, options));
+    // Don't use static imports of BlobSourceOption, it causes import resolution issues
+    // with the new UnifiedOpts shim interfaces
+    return storage.get(getName(), BucketSourceOption.toGetOptions(this, options));
   }
 
   /**
@@ -846,7 +701,9 @@ public class Bucket extends BucketInfo {
    */
   @TransportCompatibility({Transport.HTTP})
   public boolean delete(BucketSourceOption... options) {
-    return storage.delete(getName(), toSourceOptions(this, options));
+    // Don't use static imports of BlobSourceOption, it causes import resolution issues
+    // with the new UnifiedOpts shim interfaces
+    return storage.delete(getName(), BucketSourceOption.toSourceOptions(this, options));
   }
 
   /**
@@ -981,9 +838,8 @@ public class Bucket extends BucketInfo {
   public Blob create(String blob, byte[] content, String contentType, BlobTargetOption... options) {
     BlobInfo blobInfo =
         BlobInfo.newBuilder(BlobId.of(getName(), blob)).setContentType(contentType).build();
-    Tuple<BlobInfo, Storage.BlobTargetOption[]> target =
-        BlobTargetOption.toTargetOptions(blobInfo, options);
-    return storage.create(target.x(), content, target.y());
+    Storage.BlobTargetOption[] targetOptions = BlobTargetOption.toTargetOptions(blobInfo, options);
+    return storage.create(blobInfo, content, targetOptions);
   }
 
   /**
@@ -1011,9 +867,8 @@ public class Bucket extends BucketInfo {
       String blob, InputStream content, String contentType, BlobWriteOption... options) {
     BlobInfo blobInfo =
         BlobInfo.newBuilder(BlobId.of(getName(), blob)).setContentType(contentType).build();
-    Tuple<BlobInfo, Storage.BlobWriteOption[]> write =
-        BlobWriteOption.toWriteOptions(blobInfo, options);
-    return storage.create(write.x(), content, write.y());
+    Storage.BlobWriteOption[] writeOptions = BlobWriteOption.toWriteOptions(blobInfo, options);
+    return storage.create(blobInfo, content, writeOptions);
   }
 
   /**
@@ -1038,9 +893,8 @@ public class Bucket extends BucketInfo {
   @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   public Blob create(String blob, byte[] content, BlobTargetOption... options) {
     BlobInfo blobInfo = BlobInfo.newBuilder(BlobId.of(getName(), blob)).build();
-    Tuple<BlobInfo, Storage.BlobTargetOption[]> target =
-        BlobTargetOption.toTargetOptions(blobInfo, options);
-    return storage.create(target.x(), content, target.y());
+    Storage.BlobTargetOption[] targetOptions = BlobTargetOption.toTargetOptions(blobInfo, options);
+    return storage.create(blobInfo, content, targetOptions);
   }
 
   /**
@@ -1065,9 +919,8 @@ public class Bucket extends BucketInfo {
   @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   public Blob create(String blob, InputStream content, BlobWriteOption... options) {
     BlobInfo blobInfo = BlobInfo.newBuilder(BlobId.of(getName(), blob)).build();
-    Tuple<BlobInfo, Storage.BlobWriteOption[]> write =
-        BlobWriteOption.toWriteOptions(blobInfo, options);
-    return storage.create(write.x(), content, write.y());
+    Storage.BlobWriteOption[] write = BlobWriteOption.toWriteOptions(blobInfo, options);
+    return storage.create(blobInfo, content, write);
   }
 
   /**
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
index f8287ca1..9d840d5c 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
@@ -41,11 +41,11 @@ import com.google.cloud.storage.PostPolicyV4.PostConditionsV4;
 import com.google.cloud.storage.PostPolicyV4.PostFieldsV4;
 import com.google.cloud.storage.UnbufferedReadableByteChannelSession.UnbufferedReadableByteChannel;
 import com.google.cloud.storage.UnbufferedWritableByteChannelSession.UnbufferedWritableByteChannel;
+import com.google.cloud.storage.UnifiedOpts.Opts;
 import com.google.cloud.storage.spi.v1.StorageRpc;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
-import com.google.common.collect.Iterables;
 import com.google.common.hash.HashCode;
 import com.google.common.hash.Hashing;
 import com.google.common.io.ByteStreams;
@@ -139,7 +139,8 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
 
   @Override
   public Bucket create(BucketInfo bucketInfo, BucketTargetOption... options) {
-    final Map<StorageRpc.Option, ?> optionsMap = StorageImpl.optionMap(options);
+    ImmutableMap<StorageRpc.Option, ?> optionsMap =
+        Opts.unwrap(options).resolveFrom(bucketInfo).getRpcOptions();
     GrpcCallContext grpcCallContext = GrpcRequestMetadataSupport.create(optionsMap);
     com.google.storage.v2.Bucket bucket = codecs.bucketInfo().encode(bucketInfo);
     CreateBucketRequest.Builder builder = CreateBucketRequest.newBuilder().setBucket(bucket);
@@ -175,7 +176,8 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
       BlobInfo blobInfo, byte[] content, int offset, int length, BlobTargetOption... options) {
     requireNonNull(blobInfo, "blobInfo must be non null");
     requireNonNull(content, "content must be non null");
-    Map<StorageRpc.Option, ?> optionsMap = StorageImpl.optionMap(options);
+    ImmutableMap<StorageRpc.Option, ?> optionsMap =
+        Opts.unwrap(options).resolveFrom(blobInfo).getRpcOptions();
     GrpcCallContext grpcCallContext = GrpcRequestMetadataSupport.create(optionsMap);
     WriteObjectRequest req = getWriteObjectRequestBuilder(blobInfo, optionsMap).build();
     try {
@@ -223,9 +225,8 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
       throw new StorageException(0, path + " is a directory");
     }
 
-    BlobTargetOption[] translate = translate(options);
-    // TODO: Why does optionMap not accept BlobWriteOption?
-    Map<StorageRpc.Option, ?> optionsMap = StorageImpl.optionMap(translate);
+    ImmutableMap<StorageRpc.Option, ?> optionsMap =
+        Opts.unwrap(options).resolveFrom(blobInfo).getRpcOptions();
     GrpcCallContext grpcCallContext = GrpcRequestMetadataSupport.create(optionsMap);
     WriteObjectRequest req = getWriteObjectRequestBuilder(blobInfo, optionsMap).build();
 
@@ -281,9 +282,8 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
       throws IOException {
     requireNonNull(blobInfo, "blobInfo must be non null");
 
-    BlobTargetOption[] translate = translate(options);
-    // TODO: Why does optionMap not accept BlobWriteOption?
-    Map<StorageRpc.Option, ?> optionsMap = StorageImpl.optionMap(translate);
+    ImmutableMap<StorageRpc.Option, ?> optionsMap =
+        Opts.unwrap(options).resolveFrom(blobInfo).getRpcOptions();
     GrpcCallContext grpcCallContext = GrpcRequestMetadataSupport.create(optionsMap);
     WriteObjectRequest req = getWriteObjectRequestBuilder(blobInfo, optionsMap).build();
 
@@ -321,7 +321,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
 
   @Override
   public Bucket get(String bucket, BucketGetOption... options) {
-    final Map<StorageRpc.Option, ?> optionsMap = StorageImpl.optionMap(options);
+    ImmutableMap<StorageRpc.Option, ?> optionsMap = Opts.unwrap(options).getRpcOptions();
     GrpcCallContext grpcCallContext = GrpcRequestMetadataSupport.create(optionsMap);
     GetBucketRequest.Builder builder =
         GetBucketRequest.newBuilder().setName(bucketNameCodec.encode(bucket));
@@ -351,7 +351,8 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
 
   @Override
   public Blob get(BlobId blob, BlobGetOption... options) {
-    final Map<StorageRpc.Option, ?> optionsMap = StorageImpl.optionMap(options);
+    ImmutableMap<StorageRpc.Option, ?> optionsMap =
+        Opts.unwrap(options).resolveFrom(blob).getRpcOptions();
     GrpcCallContext grpcCallContext = GrpcRequestMetadataSupport.create(optionsMap);
     GetObjectRequest.Builder builder =
         GetObjectRequest.newBuilder().setBucket(blob.getBucket()).setObject(blob.getName());
@@ -377,7 +378,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
   public Page<Bucket> list(BucketListOption... options) {
     UnaryCallable<ListBucketsRequest, ListBucketsPagedResponse> listBucketsCallable =
         grpcStorageStub.listBucketsPagedCallable();
-    final Map<StorageRpc.Option, ?> optionsMap = StorageImpl.optionMap(options);
+    ImmutableMap<StorageRpc.Option, ?> optionsMap = Opts.unwrap(options).getRpcOptions();
     GrpcCallContext grpcCallContext = GrpcRequestMetadataSupport.create(optionsMap);
     String projectId =
         firstNonNull(ZOpt.PROJECT_ID.get(optionsMap), this.getOptions().getProjectId());
@@ -398,7 +399,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
   public Page<Blob> list(String bucket, BlobListOption... options) {
     UnaryCallable<ListObjectsRequest, ListObjectsPagedResponse> listObjectsCallable =
         grpcStorageStub.listObjectsPagedCallable();
-    final Map<StorageRpc.Option, ?> optionsMap = StorageImpl.optionMap(options);
+    ImmutableMap<StorageRpc.Option, ?> optionsMap = Opts.unwrap(options).getRpcOptions();
     GrpcCallContext grpcCallContext = GrpcRequestMetadataSupport.create(optionsMap);
     ListObjectsRequest.Builder builder =
         ListObjectsRequest.newBuilder().setParent(bucketNameCodec.encode(bucket));
@@ -419,7 +420,8 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
 
   @Override
   public Bucket update(BucketInfo bucketInfo, BucketTargetOption... options) {
-    final Map<StorageRpc.Option, ?> optionsMap = StorageImpl.optionMap(options);
+    ImmutableMap<StorageRpc.Option, ?> optionsMap =
+        Opts.unwrap(options).resolveFrom(bucketInfo).getRpcOptions();
     GrpcCallContext grpcCallContext = GrpcRequestMetadataSupport.create(optionsMap);
     com.google.storage.v2.Bucket bucket = codecs.bucketInfo().encode(bucketInfo);
     UpdateBucketRequest.Builder builder = UpdateBucketRequest.newBuilder().setBucket(bucket);
@@ -440,7 +442,8 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
 
   @Override
   public Blob update(BlobInfo blobInfo, BlobTargetOption... options) {
-    final Map<StorageRpc.Option, ?> optionsMap = StorageImpl.optionMap(options);
+    ImmutableMap<StorageRpc.Option, ?> optionsMap =
+        Opts.unwrap(options).resolveFrom(blobInfo).getRpcOptions();
     GrpcCallContext grpcCallContext = GrpcRequestMetadataSupport.create(optionsMap);
     Object object = codecs.blobInfo().encode(blobInfo);
     UpdateObjectRequest.Builder builder = UpdateObjectRequest.newBuilder().setObject(object);
@@ -467,7 +470,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
 
   @Override
   public boolean delete(String bucket, BucketSourceOption... options) {
-    final Map<StorageRpc.Option, ?> optionsMap = StorageImpl.optionMap(options);
+    ImmutableMap<StorageRpc.Option, ?> optionsMap = Opts.unwrap(options).getRpcOptions();
     GrpcCallContext grpcCallContext = GrpcRequestMetadataSupport.create(optionsMap);
     DeleteBucketRequest.Builder builder =
         DeleteBucketRequest.newBuilder().setName(bucketNameCodec.encode(bucket));
@@ -496,7 +499,8 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
 
   @Override
   public boolean delete(BlobId blob, BlobSourceOption... options) {
-    final Map<StorageRpc.Option, ?> optionsMap = StorageImpl.optionMap(options);
+    ImmutableMap<StorageRpc.Option, ?> optionsMap =
+        Opts.unwrap(options).resolveFrom(blob).getRpcOptions();
     GrpcCallContext grpcCallContext = GrpcRequestMetadataSupport.create(optionsMap);
     DeleteObjectRequest.Builder builder =
         DeleteObjectRequest.newBuilder().setBucket(blob.getBucket()).setObject(blob.getName());
@@ -529,7 +533,9 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
   @Override
   public Blob compose(ComposeRequest composeRequest) {
     final Map<StorageRpc.Option, ?> optionsMap =
-        StorageImpl.optionMap(Iterables.toArray(composeRequest.getTargetOptions(), Option.class));
+        Opts.unwrap(composeRequest.getTargetOptions())
+            .resolveFrom(composeRequest.getTarget())
+            .getRpcOptions();
     GrpcCallContext grpcCallContext = GrpcRequestMetadataSupport.create(optionsMap);
     ComposeObjectRequest.Builder builder = ComposeObjectRequest.newBuilder();
     composeRequest.getSourceBlobs().stream()
@@ -571,7 +577,8 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
   @Override
   public byte[] readAllBytes(BlobId blob, BlobSourceOption... options) {
 
-    final Map<StorageRpc.Option, ?> optionsMap = StorageImpl.optionMap(options);
+    ImmutableMap<StorageRpc.Option, ?> optionsMap =
+        Opts.unwrap(options).resolveFrom(blob).getRpcOptions();
 
     ReadObjectRequest readObjectRequest = getReadObjectRequest(blob, optionsMap);
     UnbufferedReadableByteChannelSession<Object> session =
@@ -604,7 +611,8 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
 
   @Override
   public GrpcBlobReadChannel reader(BlobId blob, BlobSourceOption... options) {
-    final Map<StorageRpc.Option, ?> optionsMap = StorageImpl.optionMap(options);
+    ImmutableMap<StorageRpc.Option, ?> optionsMap =
+        Opts.unwrap(options).resolveFrom(blob).getRpcOptions();
     ReadObjectRequest request = getReadObjectRequest(blob, optionsMap);
     return new GrpcBlobReadChannel(grpcStorageStub.readObjectCallable(), request);
   }
@@ -612,7 +620,8 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
   @Override
   public void downloadTo(BlobId blob, Path path, BlobSourceOption... options) {
 
-    final Map<StorageRpc.Option, ?> optionsMap = StorageImpl.optionMap(options);
+    ImmutableMap<StorageRpc.Option, ?> optionsMap =
+        Opts.unwrap(options).resolveFrom(blob).getRpcOptions();
 
     // TODO: handle StorageRpc.Option.RETURN_RAW_INPUT_STREAM impacts
 
@@ -636,7 +645,8 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
   @Override
   public void downloadTo(BlobId blob, OutputStream outputStream, BlobSourceOption... options) {
 
-    final Map<StorageRpc.Option, ?> optionsMap = StorageImpl.optionMap(options);
+    ImmutableMap<StorageRpc.Option, ?> optionsMap =
+        Opts.unwrap(options).resolveFrom(blob).getRpcOptions();
 
     // TODO: handle StorageRpc.Option.RETURN_RAW_INPUT_STREAM impacts
 
@@ -659,9 +669,8 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
 
   @Override
   public GrpcBlobWriteChannel writer(BlobInfo blobInfo, BlobWriteOption... options) {
-    BlobTargetOption[] translate = translate(options);
-    // TODO: Why does optionMap not accept BlobWriteOption?
-    Map<StorageRpc.Option, ?> optionsMap = StorageImpl.optionMap(translate);
+    ImmutableMap<StorageRpc.Option, ?> optionsMap =
+        Opts.unwrap(options).resolveFrom(blobInfo).getRpcOptions();
     GrpcCallContext grpcCallContext = GrpcRequestMetadataSupport.create(optionsMap);
     WriteObjectRequest req = getWriteObjectRequestBuilder(blobInfo, optionsMap).build();
     return new GrpcBlobWriteChannel(
@@ -891,7 +900,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
   public Page<HmacKeyMetadata> listHmacKeys(ListHmacKeysOption... options) {
     UnaryCallable<ListHmacKeysRequest, ListHmacKeysPagedResponse> listHmacKeysCallable =
         grpcStorageStub.listHmacKeysPagedCallable();
-    final Map<StorageRpc.Option, ?> optionsMap = StorageImpl.optionMap(options);
+    ImmutableMap<StorageRpc.Option, ?> optionsMap = Opts.unwrap(options).getRpcOptions();
     GrpcCallContext grpcCallContext = GrpcRequestMetadataSupport.create(optionsMap);
     String projectId =
         firstNonNull(ZOpt.PROJECT_ID.get(optionsMap), this.getOptions().getProjectId());
@@ -914,7 +923,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
 
   @Override
   public void deleteHmacKey(HmacKeyMetadata hmacKeyMetadata, DeleteHmacKeyOption... options) {
-    final Map<StorageRpc.Option, ?> optionsMap = StorageImpl.optionMap(options);
+    ImmutableMap<StorageRpc.Option, ?> optionsMap = Opts.unwrap(options).getRpcOptions();
     GrpcCallContext grpcCallContext = GrpcRequestMetadataSupport.create(optionsMap);
     DeleteHmacKeyRequest req =
         DeleteHmacKeyRequest.newBuilder()
@@ -989,14 +998,6 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     return (GrpcStorageOptions) super.getOptions();
   }
 
-  private BlobWriteOption[] translate(BlobTargetOption[] options) {
-    return todo();
-  }
-
-  private BlobTargetOption[] translate(BlobWriteOption[] options) {
-    return todo();
-  }
-
   private Blob getBlob(ApiFuture<WriteObjectResponse> result) {
     // TODO: investigate if we need to unnest any exception
     WriteObjectResponse response = ApiExceptions.callAndTranslateApiException(result);
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/Option.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Option.java
index 6bb872d1..214c4e92 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/Option.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/Option.java
@@ -16,53 +16,17 @@
 
 package com.google.cloud.storage;
 
-import static com.google.common.base.Preconditions.checkNotNull;
-
-import com.google.cloud.storage.spi.v1.StorageRpc;
-import com.google.common.base.MoreObjects;
+import com.google.cloud.storage.UnifiedOpts.Opt;
 import java.io.Serializable;
-import java.util.Objects;
 
 /** Base class for Storage operation option. */
-public abstract class Option implements Serializable {
-
-  private static final long serialVersionUID = -73199088766477208L;
-
-  private final StorageRpc.Option rpcOption;
-  private final Object value;
-
-  Option(StorageRpc.Option rpcOption, Object value) {
-    this.rpcOption = checkNotNull(rpcOption);
-    this.value = value;
-  }
-
-  StorageRpc.Option getRpcOption() {
-    return rpcOption;
-  }
-
-  Object getValue() {
-    return value;
-  }
+@Deprecated
+public abstract class Option<O extends Opt> extends UnifiedOpts.OptionShim<O>
+    implements Serializable {
 
-  @Override
-  public boolean equals(Object obj) {
-    if (!(obj instanceof Option)) {
-      return false;
-    }
-    Option other = (Option) obj;
-    return Objects.equals(rpcOption, other.rpcOption) && Objects.equals(value, other.value);
-  }
-
-  @Override
-  public int hashCode() {
-    return Objects.hash(rpcOption, value);
-  }
+  private static final long serialVersionUID = -3517676609070123326L;
 
-  @Override
-  public String toString() {
-    return MoreObjects.toStringHelper(this)
-        .add("name", rpcOption.value())
-        .add("value", value)
-        .toString();
+  Option(O opt) {
+    super(opt);
   }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java
index bf5a6f03..e8a7978f 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java
@@ -28,18 +28,24 @@ import com.google.cloud.FieldSelector.Helper;
 import com.google.cloud.Policy;
 import com.google.cloud.ReadChannel;
 import com.google.cloud.Service;
-import com.google.cloud.Tuple;
 import com.google.cloud.WriteChannel;
 import com.google.cloud.storage.Acl.Entity;
 import com.google.cloud.storage.HmacKey.HmacKeyMetadata;
 import com.google.cloud.storage.PostPolicyV4.PostConditionsV4;
 import com.google.cloud.storage.PostPolicyV4.PostFieldsV4;
 import com.google.cloud.storage.TransportCompatibility.Transport;
-import com.google.cloud.storage.spi.v1.StorageRpc;
+import com.google.cloud.storage.UnifiedOpts.BucketListOpt;
+import com.google.cloud.storage.UnifiedOpts.BucketSourceOpt;
+import com.google.cloud.storage.UnifiedOpts.BucketTargetOpt;
+import com.google.cloud.storage.UnifiedOpts.HmacKeyListOpt;
+import com.google.cloud.storage.UnifiedOpts.HmacKeySourceOpt;
+import com.google.cloud.storage.UnifiedOpts.HmacKeyTargetOpt;
+import com.google.cloud.storage.UnifiedOpts.ObjectListOpt;
+import com.google.cloud.storage.UnifiedOpts.ObjectSourceOpt;
+import com.google.cloud.storage.UnifiedOpts.ObjectTargetOpt;
+import com.google.cloud.storage.UnifiedOpts.OptionShim;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Iterables;
-import com.google.common.collect.Lists;
-import com.google.common.io.BaseEncoding;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
@@ -53,7 +59,6 @@ import java.util.LinkedHashSet;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
-import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
@@ -189,27 +194,22 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
   }
 
   /** Class for specifying bucket target options. */
-  class BucketTargetOption extends Option {
+  class BucketTargetOption extends Option<BucketTargetOpt> {
 
-    private static final long serialVersionUID = -5880204616982900975L;
+    private static final long serialVersionUID = 22594853046651193L;
 
-    private BucketTargetOption(StorageRpc.Option rpcOption, Object value) {
-      super(rpcOption, value);
-    }
-
-    private BucketTargetOption(StorageRpc.Option rpcOption) {
-      this(rpcOption, null);
+    private BucketTargetOption(BucketTargetOpt opt) {
+      super(opt);
     }
 
     /** Returns an option for specifying bucket's predefined ACL configuration. */
     public static BucketTargetOption predefinedAcl(PredefinedAcl acl) {
-      return new BucketTargetOption(StorageRpc.Option.PREDEFINED_ACL, acl.getEntry());
+      return new BucketTargetOption(UnifiedOpts.predefinedAcl(acl));
     }
 
     /** Returns an option for specifying bucket's default ACL configuration for blobs. */
     public static BucketTargetOption predefinedDefaultObjectAcl(PredefinedAcl acl) {
-      return new BucketTargetOption(
-          StorageRpc.Option.PREDEFINED_DEFAULT_OBJECT_ACL, acl.getEntry());
+      return new BucketTargetOption(UnifiedOpts.predefinedDefaultObjectAcl(acl));
     }
 
     /**
@@ -217,7 +217,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * fail if metageneration does not match.
      */
     public static BucketTargetOption metagenerationMatch() {
-      return new BucketTargetOption(StorageRpc.Option.IF_METAGENERATION_MATCH);
+      return new BucketTargetOption(UnifiedOpts.metagenerationMatchExtractor());
     }
 
     /**
@@ -225,7 +225,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * will fail if metageneration matches.
      */
     public static BucketTargetOption metagenerationNotMatch() {
-      return new BucketTargetOption(StorageRpc.Option.IF_METAGENERATION_NOT_MATCH);
+      return new BucketTargetOption(UnifiedOpts.metagenerationNotMatchExtractor());
     }
 
     /**
@@ -233,7 +233,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * `requester_pays` flag enabled to assign operation costs.
      */
     public static BucketTargetOption userProject(String userProject) {
-      return new BucketTargetOption(StorageRpc.Option.USER_PROJECT, userProject);
+      return new BucketTargetOption(UnifiedOpts.userProject(userProject));
     }
 
     /**
@@ -245,17 +245,17 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      *     patch</a>
      */
     public static BucketTargetOption projection(String projection) {
-      return new BucketTargetOption(StorageRpc.Option.PROJECTION, projection);
+      return new BucketTargetOption(UnifiedOpts.projection(projection));
     }
   }
 
   /** Class for specifying bucket source options. */
-  class BucketSourceOption extends Option {
+  class BucketSourceOption extends Option<BucketSourceOpt> {
 
-    private static final long serialVersionUID = 5185657617120212117L;
+    private static final long serialVersionUID = -162739527257621625L;
 
-    private BucketSourceOption(StorageRpc.Option rpcOption, Object value) {
-      super(rpcOption, value);
+    BucketSourceOption(BucketSourceOpt opt) {
+      super(opt);
     }
 
     /**
@@ -263,7 +263,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * fail if bucket's metageneration does not match the provided value.
      */
     public static BucketSourceOption metagenerationMatch(long metageneration) {
-      return new BucketSourceOption(StorageRpc.Option.IF_METAGENERATION_MATCH, metageneration);
+      return new BucketSourceOption(UnifiedOpts.metagenerationMatch(metageneration));
     }
 
     /**
@@ -271,7 +271,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * will fail if bucket's metageneration matches the provided value.
      */
     public static BucketSourceOption metagenerationNotMatch(long metageneration) {
-      return new BucketSourceOption(StorageRpc.Option.IF_METAGENERATION_NOT_MATCH, metageneration);
+      return new BucketSourceOption(UnifiedOpts.metagenerationNotMatch(metageneration));
     }
 
     /**
@@ -279,18 +279,19 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * with 'requester_pays' flag.
      */
     public static BucketSourceOption userProject(String userProject) {
-      return new BucketSourceOption(StorageRpc.Option.USER_PROJECT, userProject);
+      return new BucketSourceOption(UnifiedOpts.userProject(userProject));
     }
 
     public static BucketSourceOption requestedPolicyVersion(long version) {
-      return new BucketSourceOption(StorageRpc.Option.REQUESTED_POLICY_VERSION, version);
+      return new BucketSourceOption(UnifiedOpts.requestedPolicyVersion(version));
     }
   }
 
   /** Class for specifying listHmacKeys options */
-  class ListHmacKeysOption extends Option {
-    private ListHmacKeysOption(StorageRpc.Option rpcOption, Object value) {
-      super(rpcOption, value);
+  class ListHmacKeysOption extends Option<HmacKeyListOpt> {
+
+    private ListHmacKeysOption(HmacKeyListOpt opt) {
+      super(opt);
     }
 
     /**
@@ -298,18 +299,17 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * keys for all accounts will be listed.
      */
     public static ListHmacKeysOption serviceAccount(ServiceAccount serviceAccount) {
-      return new ListHmacKeysOption(
-          StorageRpc.Option.SERVICE_ACCOUNT_EMAIL, serviceAccount.getEmail());
+      return new ListHmacKeysOption(UnifiedOpts.serviceAccount(serviceAccount));
     }
 
     /** Returns an option for the maximum amount of HMAC keys returned per page. */
     public static ListHmacKeysOption maxResults(long pageSize) {
-      return new ListHmacKeysOption(StorageRpc.Option.MAX_RESULTS, pageSize);
+      return new ListHmacKeysOption(UnifiedOpts.pageSize(pageSize));
     }
 
     /** Returns an option to specify the page token from which to start listing HMAC keys. */
     public static ListHmacKeysOption pageToken(String pageToken) {
-      return new ListHmacKeysOption(StorageRpc.Option.PAGE_TOKEN, pageToken);
+      return new ListHmacKeysOption(UnifiedOpts.pageToken(pageToken));
     }
 
     /**
@@ -317,7 +317,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * by default.
      */
     public static ListHmacKeysOption showDeletedKeys(boolean showDeletedKeys) {
-      return new ListHmacKeysOption(StorageRpc.Option.SHOW_DELETED_KEYS, showDeletedKeys);
+      return new ListHmacKeysOption(UnifiedOpts.showDeletedKeys(showDeletedKeys));
     }
 
     /**
@@ -325,7 +325,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * Requester Pays buckets.
      */
     public static ListHmacKeysOption userProject(String userProject) {
-      return new ListHmacKeysOption(StorageRpc.Option.USER_PROJECT, userProject);
+      return new ListHmacKeysOption(UnifiedOpts.userProject(userProject));
     }
 
     /**
@@ -333,14 +333,15 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * Application Default Credentials.
      */
     public static ListHmacKeysOption projectId(String projectId) {
-      return new ListHmacKeysOption(StorageRpc.Option.PROJECT_ID, projectId);
+      return new ListHmacKeysOption(UnifiedOpts.projectId(projectId));
     }
   }
 
   /** Class for specifying createHmacKey options */
-  class CreateHmacKeyOption extends Option {
-    private CreateHmacKeyOption(StorageRpc.Option rpcOption, Object value) {
-      super(rpcOption, value);
+  class CreateHmacKeyOption extends Option<HmacKeyTargetOpt> {
+
+    private CreateHmacKeyOption(HmacKeyTargetOpt opt) {
+      super(opt);
     }
 
     /**
@@ -348,7 +349,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * Requester Pays buckets.
      */
     public static CreateHmacKeyOption userProject(String userProject) {
-      return new CreateHmacKeyOption(StorageRpc.Option.USER_PROJECT, userProject);
+      return new CreateHmacKeyOption(UnifiedOpts.userProject(userProject));
     }
 
     /**
@@ -356,14 +357,14 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * Application Default Credentials.
      */
     public static CreateHmacKeyOption projectId(String projectId) {
-      return new CreateHmacKeyOption(StorageRpc.Option.PROJECT_ID, projectId);
+      return new CreateHmacKeyOption(UnifiedOpts.projectId(projectId));
     }
   }
 
   /** Class for specifying getHmacKey options */
-  class GetHmacKeyOption extends Option {
-    private GetHmacKeyOption(StorageRpc.Option rpcOption, Object value) {
-      super(rpcOption, value);
+  class GetHmacKeyOption extends Option<HmacKeySourceOpt> {
+    private GetHmacKeyOption(HmacKeySourceOpt opt) {
+      super(opt);
     }
 
     /**
@@ -371,7 +372,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * Requester Pays buckets.
      */
     public static GetHmacKeyOption userProject(String userProject) {
-      return new GetHmacKeyOption(StorageRpc.Option.USER_PROJECT, userProject);
+      return new GetHmacKeyOption(UnifiedOpts.userProject(userProject));
     }
 
     /**
@@ -379,14 +380,14 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * Application Default Credentials.
      */
     public static GetHmacKeyOption projectId(String projectId) {
-      return new GetHmacKeyOption(StorageRpc.Option.PROJECT_ID, projectId);
+      return new GetHmacKeyOption(UnifiedOpts.projectId(projectId));
     }
   }
 
   /** Class for specifying deleteHmacKey options */
-  class DeleteHmacKeyOption extends Option {
-    private DeleteHmacKeyOption(StorageRpc.Option rpcOption, Object value) {
-      super(rpcOption, value);
+  class DeleteHmacKeyOption extends Option<HmacKeyTargetOpt> {
+    private DeleteHmacKeyOption(HmacKeyTargetOpt opt) {
+      super(opt);
     }
 
     /**
@@ -394,14 +395,14 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * Requester Pays buckets.
      */
     public static DeleteHmacKeyOption userProject(String userProject) {
-      return new DeleteHmacKeyOption(StorageRpc.Option.USER_PROJECT, userProject);
+      return new DeleteHmacKeyOption(UnifiedOpts.userProject(userProject));
     }
   }
 
   /** Class for specifying updateHmacKey options */
-  class UpdateHmacKeyOption extends Option {
-    private UpdateHmacKeyOption(StorageRpc.Option rpcOption, Object value) {
-      super(rpcOption, value);
+  class UpdateHmacKeyOption extends Option<HmacKeyTargetOpt> {
+    private UpdateHmacKeyOption(HmacKeyTargetOpt opt) {
+      super(opt);
     }
 
     /**
@@ -409,21 +410,17 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * Requester Pays buckets.
      */
     public static UpdateHmacKeyOption userProject(String userProject) {
-      return new UpdateHmacKeyOption(StorageRpc.Option.USER_PROJECT, userProject);
+      return new UpdateHmacKeyOption(UnifiedOpts.userProject(userProject));
     }
   }
 
   /** Class for specifying bucket get options. */
-  class BucketGetOption extends Option {
+  class BucketGetOption extends Option<BucketSourceOpt> {
 
-    private static final long serialVersionUID = 1901844869484087395L;
+    private static final long serialVersionUID = 6162812462707653332L;
 
-    private BucketGetOption(StorageRpc.Option rpcOption, long metageneration) {
-      super(rpcOption, metageneration);
-    }
-
-    private BucketGetOption(StorageRpc.Option rpcOption, String value) {
-      super(rpcOption, value);
+    BucketGetOption(BucketSourceOpt opt) {
+      super(opt);
     }
 
     /**
@@ -431,7 +428,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * fail if bucket's metageneration does not match the provided value.
      */
     public static BucketGetOption metagenerationMatch(long metageneration) {
-      return new BucketGetOption(StorageRpc.Option.IF_METAGENERATION_MATCH, metageneration);
+      return new BucketGetOption(UnifiedOpts.metagenerationMatch(metageneration));
     }
 
     /**
@@ -439,7 +436,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * will fail if bucket's metageneration matches the provided value.
      */
     public static BucketGetOption metagenerationNotMatch(long metageneration) {
-      return new BucketGetOption(StorageRpc.Option.IF_METAGENERATION_NOT_MATCH, metageneration);
+      return new BucketGetOption(UnifiedOpts.metagenerationNotMatch(metageneration));
     }
 
     /**
@@ -447,7 +444,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * with 'requester_pays' flag.
      */
     public static BucketGetOption userProject(String userProject) {
-      return new BucketGetOption(StorageRpc.Option.USER_PROJECT, userProject);
+      return new BucketGetOption(UnifiedOpts.userProject(userProject));
     }
 
     /**
@@ -457,34 +454,30 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * specified.
      */
     public static BucketGetOption fields(BucketField... fields) {
-      return new BucketGetOption(
-          StorageRpc.Option.FIELDS, Helper.selector(BucketField.REQUIRED_FIELDS, fields));
+      String selector = Helper.selector(BucketField.REQUIRED_FIELDS, fields);
+      return new BucketGetOption(UnifiedOpts.fields(selector));
     }
   }
 
   /** Class for specifying blob target options. */
-  class BlobTargetOption extends Option {
-
-    private static final long serialVersionUID = 214616862061934846L;
+  class BlobTargetOption extends Option<ObjectTargetOpt> {
 
-    private BlobTargetOption(StorageRpc.Option rpcOption, Object value) {
-      super(rpcOption, value);
-    }
+    private static final long serialVersionUID = -389155232891981906L;
 
-    private BlobTargetOption(StorageRpc.Option rpcOption) {
-      this(rpcOption, null);
+    BlobTargetOption(ObjectTargetOpt opt) {
+      super(opt);
     }
 
     /** Returns an option for specifying blob's predefined ACL configuration. */
     public static BlobTargetOption predefinedAcl(PredefinedAcl acl) {
-      return new BlobTargetOption(StorageRpc.Option.PREDEFINED_ACL, acl.getEntry());
+      return new BlobTargetOption(UnifiedOpts.predefinedAcl(acl));
     }
 
     /**
      * Returns an option that causes an operation to succeed only if the target blob does not exist.
      */
     public static BlobTargetOption doesNotExist() {
-      return new BlobTargetOption(StorageRpc.Option.IF_GENERATION_MATCH, 0L);
+      return new BlobTargetOption(UnifiedOpts.doesNotExist());
     }
 
     /**
@@ -492,7 +485,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * fail if generation does not match.
      */
     public static BlobTargetOption generationMatch() {
-      return new BlobTargetOption(StorageRpc.Option.IF_GENERATION_MATCH);
+      return new BlobTargetOption(UnifiedOpts.generationMatchExtractor());
     }
 
     /**
@@ -500,7 +493,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * will fail if generation matches.
      */
     public static BlobTargetOption generationNotMatch() {
-      return new BlobTargetOption(StorageRpc.Option.IF_GENERATION_NOT_MATCH);
+      return new BlobTargetOption(UnifiedOpts.generationNotMatchExtractor());
     }
 
     /**
@@ -508,7 +501,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * fail if metageneration does not match.
      */
     public static BlobTargetOption metagenerationMatch() {
-      return new BlobTargetOption(StorageRpc.Option.IF_METAGENERATION_MATCH);
+      return new BlobTargetOption(UnifiedOpts.metagenerationMatchExtractor());
     }
 
     /**
@@ -516,7 +509,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * fail if metageneration matches.
      */
     public static BlobTargetOption metagenerationNotMatch() {
-      return new BlobTargetOption(StorageRpc.Option.IF_METAGENERATION_NOT_MATCH);
+      return new BlobTargetOption(UnifiedOpts.metagenerationNotMatchExtractor());
     }
 
     /**
@@ -524,7 +517,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * will create a blob with disableGzipContent; at present, this is only for upload.
      */
     public static BlobTargetOption disableGzipContent() {
-      return new BlobTargetOption(StorageRpc.Option.IF_DISABLE_GZIP_CONTENT, true);
+      return new BlobTargetOption(UnifiedOpts.disableGzipContent());
     }
 
     /**
@@ -533,7 +526,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * does not appear in a RPC call.
      */
     public static BlobTargetOption detectContentType() {
-      return new BlobTargetOption(StorageRpc.Option.DETECT_CONTENT_TYPE, true);
+      return new BlobTargetOption(UnifiedOpts.detectContentType());
     }
 
     /**
@@ -541,8 +534,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * blob.
      */
     public static BlobTargetOption encryptionKey(Key key) {
-      String base64Key = BaseEncoding.base64().encode(key.getEncoded());
-      return new BlobTargetOption(StorageRpc.Option.CUSTOMER_SUPPLIED_KEY, base64Key);
+      return new BlobTargetOption(UnifiedOpts.encryptionKey(key));
     }
 
     /**
@@ -550,7 +542,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * with 'requester_pays' flag.
      */
     public static BlobTargetOption userProject(String userProject) {
-      return new BlobTargetOption(StorageRpc.Option.USER_PROJECT, userProject);
+      return new BlobTargetOption(UnifiedOpts.userProject(userProject));
     }
 
     /**
@@ -560,102 +552,34 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * @param key the AES256 encoded in base64
      */
     public static BlobTargetOption encryptionKey(String key) {
-      return new BlobTargetOption(StorageRpc.Option.CUSTOMER_SUPPLIED_KEY, key);
+      return new BlobTargetOption(UnifiedOpts.encryptionKey(key));
     }
 
     /** Returns an option to set a customer-managed key for server-side encryption of the blob. */
     public static BlobTargetOption kmsKeyName(String kmsKeyName) {
-      return new BlobTargetOption(StorageRpc.Option.KMS_KEY_NAME, kmsKeyName);
-    }
-
-    static Tuple<BlobInfo, BlobTargetOption[]> convert(BlobInfo info, BlobWriteOption... options) {
-      BlobInfo.Builder infoBuilder = info.toBuilder().setCrc32c(null).setMd5(null);
-      List<BlobTargetOption> targetOptions = Lists.newArrayListWithCapacity(options.length);
-      for (BlobWriteOption option : options) {
-        switch (option.option) {
-          case IF_CRC32C_MATCH:
-            infoBuilder.setCrc32c(info.getCrc32c());
-            break;
-          case IF_MD5_MATCH:
-            infoBuilder.setMd5(info.getMd5());
-            break;
-          default:
-            targetOptions.add(option.toTargetOption());
-            break;
-        }
-      }
-      return Tuple.of(
-          infoBuilder.build(), targetOptions.toArray(new BlobTargetOption[targetOptions.size()]));
+      return new BlobTargetOption(UnifiedOpts.kmsKeyName(kmsKeyName));
     }
   }
 
   /** Class for specifying blob write options. */
-  class BlobWriteOption implements Serializable {
+  class BlobWriteOption extends OptionShim<ObjectTargetOpt> implements Serializable {
 
-    private static final long serialVersionUID = -3880421670966224580L;
+    private static final long serialVersionUID = 1834774766750256808L;
 
-    private final Option option;
-    private final Object value;
-
-    enum Option {
-      PREDEFINED_ACL,
-      IF_GENERATION_MATCH,
-      IF_GENERATION_NOT_MATCH,
-      IF_METAGENERATION_MATCH,
-      IF_METAGENERATION_NOT_MATCH,
-      IF_MD5_MATCH,
-      IF_CRC32C_MATCH,
-      CUSTOMER_SUPPLIED_KEY,
-      KMS_KEY_NAME,
-      USER_PROJECT,
-      DETECT_CONTENT_TYPE,
-      IF_DISABLE_GZIP_CONTENT;
-
-      StorageRpc.Option toRpcOption() {
-        return StorageRpc.Option.valueOf(this.name());
-      }
-    }
-
-    BlobTargetOption toTargetOption() {
-      return new BlobTargetOption(this.option.toRpcOption(), this.value);
-    }
-
-    private BlobWriteOption(Option option, Object value) {
-      this.option = option;
-      this.value = value;
-    }
-
-    private BlobWriteOption(Option option) {
-      this(option, null);
-    }
-
-    @Override
-    public int hashCode() {
-      return Objects.hash(option, value);
-    }
-
-    @Override
-    public boolean equals(Object obj) {
-      if (obj == null) {
-        return false;
-      }
-      if (!(obj instanceof BlobWriteOption)) {
-        return false;
-      }
-      final BlobWriteOption other = (BlobWriteOption) obj;
-      return this.option == other.option && Objects.equals(this.value, other.value);
+    BlobWriteOption(ObjectTargetOpt opt) {
+      super(opt);
     }
 
     /** Returns an option for specifying blob's predefined ACL configuration. */
     public static BlobWriteOption predefinedAcl(PredefinedAcl acl) {
-      return new BlobWriteOption(Option.PREDEFINED_ACL, acl.getEntry());
+      return new BlobWriteOption(UnifiedOpts.predefinedAcl(acl));
     }
 
     /**
      * Returns an option that causes an operation to succeed only if the target blob does not exist.
      */
     public static BlobWriteOption doesNotExist() {
-      return new BlobWriteOption(Option.IF_GENERATION_MATCH, 0L);
+      return new BlobWriteOption(UnifiedOpts.doesNotExist());
     }
 
     /**
@@ -663,7 +587,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * fail if generation does not match.
      */
     public static BlobWriteOption generationMatch() {
-      return new BlobWriteOption(Option.IF_GENERATION_MATCH);
+      return new BlobWriteOption(UnifiedOpts.generationMatchExtractor());
     }
 
     /**
@@ -671,7 +595,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * will fail if generation matches.
      */
     public static BlobWriteOption generationNotMatch() {
-      return new BlobWriteOption(Option.IF_GENERATION_NOT_MATCH);
+      return new BlobWriteOption(UnifiedOpts.generationNotMatchExtractor());
     }
 
     /**
@@ -679,7 +603,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * fail if metageneration does not match.
      */
     public static BlobWriteOption metagenerationMatch() {
-      return new BlobWriteOption(Option.IF_METAGENERATION_MATCH);
+      return new BlobWriteOption(UnifiedOpts.metagenerationMatchExtractor());
     }
 
     /**
@@ -687,7 +611,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * fail if metageneration matches.
      */
     public static BlobWriteOption metagenerationNotMatch() {
-      return new BlobWriteOption(Option.IF_METAGENERATION_NOT_MATCH);
+      return new BlobWriteOption(UnifiedOpts.metagenerationNotMatchExtractor());
     }
 
     /**
@@ -695,7 +619,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * fail if blobs' data MD5 hash does not match.
      */
     public static BlobWriteOption md5Match() {
-      return new BlobWriteOption(Option.IF_MD5_MATCH, true);
+      return new BlobWriteOption(UnifiedOpts.md5MatchExtractor());
     }
 
     /**
@@ -703,7 +627,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * will fail if blobs' data CRC32C checksum does not match.
      */
     public static BlobWriteOption crc32cMatch() {
-      return new BlobWriteOption(Option.IF_CRC32C_MATCH, true);
+      return new BlobWriteOption(UnifiedOpts.crc32cMatchExtractor());
     }
 
     /**
@@ -711,8 +635,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * blob.
      */
     public static BlobWriteOption encryptionKey(Key key) {
-      String base64Key = BaseEncoding.base64().encode(key.getEncoded());
-      return new BlobWriteOption(Option.CUSTOMER_SUPPLIED_KEY, base64Key);
+      return new BlobWriteOption(UnifiedOpts.encryptionKey(key));
     }
 
     /**
@@ -722,7 +645,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * @param key the AES256 encoded in base64
      */
     public static BlobWriteOption encryptionKey(String key) {
-      return new BlobWriteOption(Option.CUSTOMER_SUPPLIED_KEY, key);
+      return new BlobWriteOption(UnifiedOpts.encryptionKey(key));
     }
 
     /**
@@ -731,7 +654,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * @param kmsKeyName the KMS key resource id
      */
     public static BlobWriteOption kmsKeyName(String kmsKeyName) {
-      return new BlobWriteOption(Option.KMS_KEY_NAME, kmsKeyName);
+      return new BlobWriteOption(UnifiedOpts.kmsKeyName(kmsKeyName));
     }
 
     /**
@@ -739,7 +662,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * with 'requester_pays' flag.
      */
     public static BlobWriteOption userProject(String userProject) {
-      return new BlobWriteOption(Option.USER_PROJECT, userProject);
+      return new BlobWriteOption(UnifiedOpts.userProject(userProject));
     }
 
     /**
@@ -747,7 +670,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * the bucket.
      */
     public static BlobWriteOption disableGzipContent() {
-      return new BlobWriteOption(Option.IF_DISABLE_GZIP_CONTENT, true);
+      return new BlobWriteOption(UnifiedOpts.disableGzipContent());
     }
 
     /**
@@ -756,17 +679,17 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * does not appear in a RPC call.
      */
     public static BlobWriteOption detectContentType() {
-      return new BlobWriteOption(Option.DETECT_CONTENT_TYPE, true);
+      return new BlobWriteOption(UnifiedOpts.detectContentType());
     }
   }
 
   /** Class for specifying blob source options. */
-  class BlobSourceOption extends Option {
+  class BlobSourceOption extends Option<ObjectSourceOpt> {
 
-    private static final long serialVersionUID = -3712768261070182991L;
+    private static final long serialVersionUID = 2131185332093994401L;
 
-    private BlobSourceOption(StorageRpc.Option rpcOption, Object value) {
-      super(rpcOption, value);
+    BlobSourceOption(ObjectSourceOpt opt) {
+      super(opt);
     }
 
     /**
@@ -777,7 +700,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * BlobId} is provided an exception is thrown.
      */
     public static BlobSourceOption generationMatch() {
-      return new BlobSourceOption(StorageRpc.Option.IF_GENERATION_MATCH, null);
+      return new BlobSourceOption(UnifiedOpts.generationMatchExtractor());
     }
 
     /**
@@ -785,7 +708,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * fail if blob's generation does not match the provided value.
      */
     public static BlobSourceOption generationMatch(long generation) {
-      return new BlobSourceOption(StorageRpc.Option.IF_GENERATION_MATCH, generation);
+      return new BlobSourceOption(UnifiedOpts.generationMatch(generation));
     }
 
     /**
@@ -800,7 +723,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      */
     @Deprecated
     public static BlobSourceOption generationNotMatch() {
-      return new BlobSourceOption(StorageRpc.Option.IF_GENERATION_NOT_MATCH, null);
+      return new BlobSourceOption(UnifiedOpts.generationNotMatchExtractor());
     }
 
     /**
@@ -808,7 +731,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * will fail if blob's generation matches the provided value.
      */
     public static BlobSourceOption generationNotMatch(long generation) {
-      return new BlobSourceOption(StorageRpc.Option.IF_GENERATION_NOT_MATCH, generation);
+      return new BlobSourceOption(UnifiedOpts.generationNotMatch(generation));
     }
 
     /**
@@ -816,7 +739,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * fail if blob's metageneration does not match the provided value.
      */
     public static BlobSourceOption metagenerationMatch(long metageneration) {
-      return new BlobSourceOption(StorageRpc.Option.IF_METAGENERATION_MATCH, metageneration);
+      return new BlobSourceOption(UnifiedOpts.metagenerationMatch(metageneration));
     }
 
     /**
@@ -824,7 +747,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * fail if blob's metageneration matches the provided value.
      */
     public static BlobSourceOption metagenerationNotMatch(long metageneration) {
-      return new BlobSourceOption(StorageRpc.Option.IF_METAGENERATION_NOT_MATCH, metageneration);
+      return new BlobSourceOption(UnifiedOpts.metagenerationNotMatch(metageneration));
     }
 
     /**
@@ -832,8 +755,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * blob.
      */
     public static BlobSourceOption decryptionKey(Key key) {
-      String base64Key = BaseEncoding.base64().encode(key.getEncoded());
-      return new BlobSourceOption(StorageRpc.Option.CUSTOMER_SUPPLIED_KEY, base64Key);
+      return new BlobSourceOption(UnifiedOpts.decryptionKey(key));
     }
 
     /**
@@ -843,7 +765,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * @param key the AES256 encoded in base64
      */
     public static BlobSourceOption decryptionKey(String key) {
-      return new BlobSourceOption(StorageRpc.Option.CUSTOMER_SUPPLIED_KEY, key);
+      return new BlobSourceOption(UnifiedOpts.decryptionKey(key));
     }
 
     /**
@@ -851,7 +773,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * with 'requester_pays' flag.
      */
     public static BlobSourceOption userProject(String userProject) {
-      return new BlobSourceOption(StorageRpc.Option.USER_PROJECT, userProject);
+      return new BlobSourceOption(UnifiedOpts.userProject(userProject));
     }
 
     /**
@@ -860,26 +782,17 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * true for ReadChannel.read().
      */
     public static BlobSourceOption shouldReturnRawInputStream(boolean shouldReturnRawInputStream) {
-      return new BlobSourceOption(
-          StorageRpc.Option.RETURN_RAW_INPUT_STREAM, shouldReturnRawInputStream);
+      return new BlobSourceOption(UnifiedOpts.returnRawInputStream(shouldReturnRawInputStream));
     }
   }
 
   /** Class for specifying blob get options. */
-  class BlobGetOption extends Option {
+  class BlobGetOption extends Option<ObjectSourceOpt> {
 
-    private static final long serialVersionUID = 803817709703661480L;
-
-    private BlobGetOption(StorageRpc.Option rpcOption, Long value) {
-      super(rpcOption, value);
-    }
-
-    private BlobGetOption(StorageRpc.Option rpcOption, String value) {
-      super(rpcOption, value);
-    }
+    private static final long serialVersionUID = 4464929616050835795L;
 
-    private BlobGetOption(StorageRpc.Option rpcOption, boolean value) {
-      super(rpcOption, value);
+    BlobGetOption(ObjectSourceOpt opt) {
+      super(opt);
     }
 
     /**
@@ -890,7 +803,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * BlobId} is provided an exception is thrown.
      */
     public static BlobGetOption generationMatch() {
-      return new BlobGetOption(StorageRpc.Option.IF_GENERATION_MATCH, (Long) null);
+      return new BlobGetOption(UnifiedOpts.generationMatchExtractor());
     }
 
     /**
@@ -898,7 +811,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * fail if blob's generation does not match the provided value.
      */
     public static BlobGetOption generationMatch(long generation) {
-      return new BlobGetOption(StorageRpc.Option.IF_GENERATION_MATCH, generation);
+      return new BlobGetOption(UnifiedOpts.generationMatch(generation));
     }
 
     /**
@@ -913,7 +826,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      */
     @Deprecated
     public static BlobGetOption generationNotMatch() {
-      return new BlobGetOption(StorageRpc.Option.IF_GENERATION_NOT_MATCH, (Long) null);
+      return new BlobGetOption(UnifiedOpts.generationNotMatchExtractor());
     }
 
     /**
@@ -921,7 +834,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * will fail if blob's generation matches the provided value.
      */
     public static BlobGetOption generationNotMatch(long generation) {
-      return new BlobGetOption(StorageRpc.Option.IF_GENERATION_NOT_MATCH, generation);
+      return new BlobGetOption(UnifiedOpts.generationNotMatch(generation));
     }
 
     /**
@@ -929,7 +842,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * fail if blob's metageneration does not match the provided value.
      */
     public static BlobGetOption metagenerationMatch(long metageneration) {
-      return new BlobGetOption(StorageRpc.Option.IF_METAGENERATION_MATCH, metageneration);
+      return new BlobGetOption(UnifiedOpts.metagenerationMatch(metageneration));
     }
 
     /**
@@ -937,7 +850,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * fail if blob's metageneration matches the provided value.
      */
     public static BlobGetOption metagenerationNotMatch(long metageneration) {
-      return new BlobGetOption(StorageRpc.Option.IF_METAGENERATION_NOT_MATCH, metageneration);
+      return new BlobGetOption(UnifiedOpts.metagenerationNotMatch(metageneration));
     }
 
     /**
@@ -947,8 +860,8 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * specified.
      */
     public static BlobGetOption fields(BlobField... fields) {
-      return new BlobGetOption(
-          StorageRpc.Option.FIELDS, Helper.selector(BlobField.REQUIRED_FIELDS, fields));
+      String selector = Helper.selector(BlobField.REQUIRED_FIELDS, fields);
+      return new BlobGetOption(UnifiedOpts.fields(selector));
     }
 
     /**
@@ -956,7 +869,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * with 'requester_pays' flag.
      */
     public static BlobGetOption userProject(String userProject) {
-      return new BlobGetOption(StorageRpc.Option.USER_PROJECT, userProject);
+      return new BlobGetOption(UnifiedOpts.userProject(userProject));
     }
 
     /**
@@ -964,8 +877,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * blob.
      */
     public static BlobGetOption decryptionKey(Key key) {
-      String base64Key = BaseEncoding.base64().encode(key.getEncoded());
-      return new BlobGetOption(StorageRpc.Option.CUSTOMER_SUPPLIED_KEY, base64Key);
+      return new BlobGetOption(UnifiedOpts.decryptionKey(key));
     }
 
     /**
@@ -975,7 +887,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * @param key the AES256 encoded in base64
      */
     public static BlobGetOption decryptionKey(String key) {
-      return new BlobGetOption(StorageRpc.Option.CUSTOMER_SUPPLIED_KEY, key);
+      return new BlobGetOption(UnifiedOpts.decryptionKey(key));
     }
 
     /**
@@ -984,28 +896,27 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * true for ReadChannel.read().
      */
     public static BlobGetOption shouldReturnRawInputStream(boolean shouldReturnRawInputStream) {
-      return new BlobGetOption(
-          StorageRpc.Option.RETURN_RAW_INPUT_STREAM, shouldReturnRawInputStream);
+      return new BlobGetOption(UnifiedOpts.returnRawInputStream(shouldReturnRawInputStream));
     }
   }
 
   /** Class for specifying bucket list options. */
-  class BucketListOption extends Option {
+  class BucketListOption extends Option<BucketListOpt> {
 
-    private static final long serialVersionUID = 8754017079673290353L;
+    private static final long serialVersionUID = 5717590216719591953L;
 
-    private BucketListOption(StorageRpc.Option option, Object value) {
-      super(option, value);
+    private BucketListOption(BucketListOpt opt) {
+      super(opt);
     }
 
     /** Returns an option to specify the maximum number of buckets returned per page. */
     public static BucketListOption pageSize(long pageSize) {
-      return new BucketListOption(StorageRpc.Option.MAX_RESULTS, pageSize);
+      return new BucketListOption(UnifiedOpts.pageSize(pageSize));
     }
 
     /** Returns an option to specify the page token from which to start listing buckets. */
     public static BucketListOption pageToken(String pageToken) {
-      return new BucketListOption(StorageRpc.Option.PAGE_TOKEN, pageToken);
+      return new BucketListOption(UnifiedOpts.pageToken(pageToken));
     }
 
     /**
@@ -1013,7 +924,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * prefix.
      */
     public static BucketListOption prefix(String prefix) {
-      return new BucketListOption(StorageRpc.Option.PREFIX, prefix);
+      return new BucketListOption(UnifiedOpts.prefix(prefix));
     }
 
     /**
@@ -1021,7 +932,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * with 'requester_pays' flag.
      */
     public static BucketListOption userProject(String userProject) {
-      return new BucketListOption(StorageRpc.Option.USER_PROJECT, userProject);
+      return new BucketListOption(UnifiedOpts.userProject(userProject));
     }
 
     /**
@@ -1031,30 +942,29 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * specified.
      */
     public static BucketListOption fields(BucketField... fields) {
-      return new BucketListOption(
-          StorageRpc.Option.FIELDS,
-          Helper.listSelector("items", BucketField.REQUIRED_FIELDS, fields));
+      String selector = Helper.listSelector("items", BucketField.REQUIRED_FIELDS, fields);
+      return new BucketListOption(UnifiedOpts.fields(selector));
     }
   }
 
   /** Class for specifying blob list options. */
-  class BlobListOption extends Option {
+  class BlobListOption extends Option<ObjectListOpt> {
 
     private static final String[] TOP_LEVEL_FIELDS = {"prefixes"};
-    private static final long serialVersionUID = 9083383524788661294L;
+    private static final long serialVersionUID = 5677832374576757707L;
 
-    private BlobListOption(StorageRpc.Option option, Object value) {
-      super(option, value);
+    private BlobListOption(ObjectListOpt opt) {
+      super(opt);
     }
 
     /** Returns an option to specify the maximum number of blobs returned per page. */
     public static BlobListOption pageSize(long pageSize) {
-      return new BlobListOption(StorageRpc.Option.MAX_RESULTS, pageSize);
+      return new BlobListOption(UnifiedOpts.pageSize(pageSize));
     }
 
     /** Returns an option to specify the page token from which to start listing blobs. */
     public static BlobListOption pageToken(String pageToken) {
-      return new BlobListOption(StorageRpc.Option.PAGE_TOKEN, pageToken);
+      return new BlobListOption(UnifiedOpts.pageToken(pageToken));
     }
 
     /**
@@ -1062,7 +972,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * prefix.
      */
     public static BlobListOption prefix(String prefix) {
-      return new BlobListOption(StorageRpc.Option.PREFIX, prefix);
+      return new BlobListOption(UnifiedOpts.prefix(prefix));
     }
 
     /**
@@ -1076,7 +986,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * Duplicate directory blobs are omitted.
      */
     public static BlobListOption currentDirectory() {
-      return new BlobListOption(StorageRpc.Option.DELIMITER, true);
+      return new BlobListOption(UnifiedOpts.currentDirectory());
     }
 
     /**
@@ -1086,7 +996,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      *     as well.
      */
     public static BlobListOption delimiter(String delimiter) {
-      return new BlobListOption(StorageRpc.Option.DELIMITER, delimiter);
+      return new BlobListOption(UnifiedOpts.delimiter(delimiter));
     }
 
     /**
@@ -1097,7 +1007,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * @param startOffset startOffset to filter the results
      */
     public static BlobListOption startOffset(String startOffset) {
-      return new BlobListOption(StorageRpc.Option.START_OFF_SET, startOffset);
+      return new BlobListOption(UnifiedOpts.startOffset(startOffset));
     }
 
     /**
@@ -1108,7 +1018,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * @param endOffset endOffset to filter the results
      */
     public static BlobListOption endOffset(String endOffset) {
-      return new BlobListOption(StorageRpc.Option.END_OFF_SET, endOffset);
+      return new BlobListOption(UnifiedOpts.endOffset(endOffset));
     }
 
     /**
@@ -1118,7 +1028,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * @param userProject projectId of the billing user project.
      */
     public static BlobListOption userProject(String userProject) {
-      return new BlobListOption(StorageRpc.Option.USER_PROJECT, userProject);
+      return new BlobListOption(UnifiedOpts.userProject(userProject));
     }
 
     /**
@@ -1127,7 +1037,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * @see <a href="https://cloud.google.com/storage/docs/object-versioning">Object Versioning</a>
      */
     public static BlobListOption versions(boolean versions) {
-      return new BlobListOption(StorageRpc.Option.VERSIONS, versions);
+      return new BlobListOption(UnifiedOpts.versionsFilter(versions));
     }
 
     /**
@@ -1137,9 +1047,9 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
      * specified.
      */
     public static BlobListOption fields(BlobField... fields) {
-      return new BlobListOption(
-          StorageRpc.Option.FIELDS,
-          Helper.listSelector(TOP_LEVEL_FIELDS, "items", BlobField.REQUIRED_FIELDS, fields));
+      String selector =
+          Helper.listSelector(TOP_LEVEL_FIELDS, "items", BlobField.REQUIRED_FIELDS, fields);
+      return new BlobListOption(UnifiedOpts.fields(selector));
     }
   }
 
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageBatch.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageBatch.java
index 5cc19f97..2b89d442 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageBatch.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageBatch.java
@@ -23,6 +23,7 @@ import com.google.api.services.storage.model.StorageObject;
 import com.google.cloud.storage.Storage.BlobGetOption;
 import com.google.cloud.storage.Storage.BlobSourceOption;
 import com.google.cloud.storage.Storage.BlobTargetOption;
+import com.google.cloud.storage.UnifiedOpts.Opts;
 import com.google.cloud.storage.spi.v1.RpcBatch;
 import com.google.cloud.storage.spi.v1.StorageRpc;
 import com.google.common.annotations.VisibleForTesting;
@@ -99,8 +100,8 @@ public class StorageBatch {
   public StorageBatchResult<Boolean> delete(BlobId blob, BlobSourceOption... options) {
     StorageBatchResult<Boolean> result = new StorageBatchResult<>();
     RpcBatch.Callback<Void> callback = createDeleteCallback(result);
-    Map<StorageRpc.Option, ?> optionMap = StorageImpl.optionMap(blob, options);
-    batch.addDelete(Conversions.apiary().blobId().encode(blob), callback, optionMap);
+    Map<StorageRpc.Option, ?> optionsMap = Opts.unwrap(options).resolveFrom(blob).getRpcOptions();
+    batch.addDelete(Conversions.apiary().blobId().encode(blob), callback, optionsMap);
     return result;
   }
 
@@ -113,7 +114,8 @@ public class StorageBatch {
   public StorageBatchResult<Blob> update(BlobInfo blobInfo, BlobTargetOption... options) {
     StorageBatchResult<Blob> result = new StorageBatchResult<>();
     RpcBatch.Callback<StorageObject> callback = createUpdateCallback(this.options, result);
-    Map<StorageRpc.Option, ?> optionMap = StorageImpl.optionMap(blobInfo, options);
+    Map<StorageRpc.Option, ?> optionMap =
+        Opts.unwrap(options).resolveFrom(blobInfo).getRpcOptions();
     batch.addPatch(Conversions.apiary().blobInfo().encode(blobInfo), callback, optionMap);
     return result;
   }
@@ -139,8 +141,8 @@ public class StorageBatch {
   public StorageBatchResult<Blob> get(BlobId blob, BlobGetOption... options) {
     StorageBatchResult<Blob> result = new StorageBatchResult<>();
     RpcBatch.Callback<StorageObject> callback = createGetCallback(this.options, result);
-    Map<StorageRpc.Option, ?> optionMap = StorageImpl.optionMap(blob, options);
-    batch.addGet(Conversions.apiary().blobId().encode(blob), callback, optionMap);
+    Map<StorageRpc.Option, ?> optionsMap = Opts.unwrap(options).resolveFrom(blob).getRpcOptions();
+    batch.addGet(Conversions.apiary().blobId().encode(blob), callback, optionsMap);
     return result;
   }
 
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java
index 54d5f925..757578fc 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java
@@ -19,15 +19,6 @@ package com.google.cloud.storage;
 import static com.google.cloud.RetryHelper.runWithRetries;
 import static com.google.cloud.storage.PolicyHelper.convertToApiPolicy;
 import static com.google.cloud.storage.SignedUrlEncodingHelper.Rfc3986UriEncode;
-import static com.google.cloud.storage.spi.v1.StorageRpc.Option.DELIMITER;
-import static com.google.cloud.storage.spi.v1.StorageRpc.Option.IF_GENERATION_MATCH;
-import static com.google.cloud.storage.spi.v1.StorageRpc.Option.IF_GENERATION_NOT_MATCH;
-import static com.google.cloud.storage.spi.v1.StorageRpc.Option.IF_METAGENERATION_MATCH;
-import static com.google.cloud.storage.spi.v1.StorageRpc.Option.IF_METAGENERATION_NOT_MATCH;
-import static com.google.cloud.storage.spi.v1.StorageRpc.Option.IF_SOURCE_GENERATION_MATCH;
-import static com.google.cloud.storage.spi.v1.StorageRpc.Option.IF_SOURCE_GENERATION_NOT_MATCH;
-import static com.google.cloud.storage.spi.v1.StorageRpc.Option.IF_SOURCE_METAGENERATION_MATCH;
-import static com.google.cloud.storage.spi.v1.StorageRpc.Option.IF_SOURCE_METAGENERATION_NOT_MATCH;
 import static com.google.common.base.MoreObjects.firstNonNull;
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkState;
@@ -47,7 +38,6 @@ import com.google.cloud.PageImpl.NextPageFetcher;
 import com.google.cloud.Policy;
 import com.google.cloud.ReadChannel;
 import com.google.cloud.RetryHelper.RetryHelperException;
-import com.google.cloud.Tuple;
 import com.google.cloud.WriteChannel;
 import com.google.cloud.storage.Acl.Entity;
 import com.google.cloud.storage.HmacKey.HmacKeyMetadata;
@@ -55,6 +45,9 @@ import com.google.cloud.storage.PostPolicyV4.ConditionV4Type;
 import com.google.cloud.storage.PostPolicyV4.PostConditionsV4;
 import com.google.cloud.storage.PostPolicyV4.PostFieldsV4;
 import com.google.cloud.storage.PostPolicyV4.PostPolicyV4Document;
+import com.google.cloud.storage.UnifiedOpts.ObjectSourceOpt;
+import com.google.cloud.storage.UnifiedOpts.ObjectTargetOpt;
+import com.google.cloud.storage.UnifiedOpts.Opts;
 import com.google.cloud.storage.spi.v1.StorageRpc;
 import com.google.cloud.storage.spi.v1.StorageRpc.RewriteRequest;
 import com.google.common.base.CharMatcher;
@@ -126,7 +119,8 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
   public Bucket create(BucketInfo bucketInfo, BucketTargetOption... options) {
     final com.google.api.services.storage.model.Bucket bucketPb =
         codecs.bucketInfo().encode(bucketInfo);
-    final Map<StorageRpc.Option, ?> optionsMap = optionMap(bucketInfo, options);
+    final Map<StorageRpc.Option, ?> optionsMap =
+        Opts.unwrap(options).resolveFrom(bucketInfo).getRpcOptions();
     ResultRetryAlgorithm<?> algorithm =
         retryAlgorithmManager.getForBucketsCreate(bucketPb, optionsMap);
     return run(
@@ -182,9 +176,11 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
   @Override
   @Deprecated
   public Blob create(BlobInfo blobInfo, InputStream content, BlobWriteOption... options) {
-    Tuple<BlobInfo, BlobTargetOption[]> targetOptions = BlobTargetOption.convert(blobInfo, options);
-    StorageObject blobPb = codecs.blobInfo().encode(targetOptions.x());
-    Map<StorageRpc.Option, ?> optionsMap = optionMap(targetOptions.x(), targetOptions.y());
+    Opts<ObjectTargetOpt> opts = Opts.unwrap(options).resolveFrom(blobInfo);
+    Map<StorageRpc.Option, ?> optionsMap = opts.getRpcOptions();
+    BlobInfo.Builder builder = blobInfo.toBuilder().setMd5(null).setCrc32c(null);
+    BlobInfo updated = opts.blobInfoMapper().apply(builder).build();
+    StorageObject blobPb = codecs.blobInfo().encode(updated);
     InputStream inputStreamParam =
         firstNonNull(content, new ByteArrayInputStream(EMPTY_BYTE_ARRAY));
     // retries are not safe when the input is an InputStream, so we can't retry.
@@ -202,8 +198,11 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
       final int length,
       BlobTargetOption... options) {
     Preconditions.checkNotNull(content);
-    final StorageObject blobPb = codecs.blobInfo().encode(info);
-    final Map<StorageRpc.Option, ?> optionsMap = optionMap(info, options);
+    Opts<ObjectTargetOpt> opts = Opts.unwrap(options).resolveFrom(info);
+    final Map<StorageRpc.Option, ?> optionsMap = opts.getRpcOptions();
+
+    BlobInfo updated = opts.blobInfoMapper().apply(info.toBuilder()).build();
+    final StorageObject blobPb = codecs.blobInfo().encode(updated);
     ResultRetryAlgorithm<?> algorithm =
         retryAlgorithmManager.getForObjectsCreate(blobPb, optionsMap);
     return run(
@@ -276,7 +275,7 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
   public Bucket get(String bucket, BucketGetOption... options) {
     final com.google.api.services.storage.model.Bucket bucketPb =
         codecs.bucketInfo().encode(BucketInfo.of(bucket));
-    final Map<StorageRpc.Option, ?> optionsMap = optionMap(options);
+    ImmutableMap<StorageRpc.Option, ?> optionsMap = Opts.unwrap(options).getRpcOptions();
     ResultRetryAlgorithm<?> algorithm =
         retryAlgorithmManager.getForBucketsGet(bucketPb, optionsMap);
     return run(
@@ -293,7 +292,8 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
   @Override
   public Blob get(BlobId blob, BlobGetOption... options) {
     final StorageObject storedObject = codecs.blobId().encode(blob);
-    final Map<StorageRpc.Option, ?> optionsMap = optionMap(blob, options);
+    ImmutableMap<StorageRpc.Option, ?> optionsMap =
+        Opts.unwrap(options).resolveFrom(blob).getRpcOptions();
     ResultRetryAlgorithm<?> algorithm =
         retryAlgorithmManager.getForObjectsGet(storedObject, optionsMap);
     return run(
@@ -377,12 +377,14 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
 
   @Override
   public Page<Bucket> list(BucketListOption... options) {
-    return listBuckets(getOptions(), optionMap(options));
+    ImmutableMap<StorageRpc.Option, ?> optionsMap = Opts.unwrap(options).getRpcOptions();
+    return listBuckets(getOptions(), optionsMap);
   }
 
   @Override
   public Page<Blob> list(final String bucket, BlobListOption... options) {
-    return listBlobs(bucket, getOptions(), optionMap(options));
+    ImmutableMap<StorageRpc.Option, ?> optionsMap = Opts.unwrap(options).getRpcOptions();
+    return listBlobs(bucket, getOptions(), optionsMap);
   }
 
   private static Page<Bucket> listBuckets(
@@ -440,7 +442,8 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
   public Bucket update(BucketInfo bucketInfo, BucketTargetOption... options) {
     final com.google.api.services.storage.model.Bucket bucketPb =
         codecs.bucketInfo().encode(bucketInfo);
-    final Map<StorageRpc.Option, ?> optionsMap = optionMap(bucketInfo, options);
+    final Map<StorageRpc.Option, ?> optionsMap =
+        Opts.unwrap(options).resolveFrom(bucketInfo).getRpcOptions();
     ResultRetryAlgorithm<?> algorithm =
         retryAlgorithmManager.getForBucketsUpdate(bucketPb, optionsMap);
     return run(
@@ -451,13 +454,14 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
 
   @Override
   public Blob update(BlobInfo blobInfo, BlobTargetOption... options) {
-    final StorageObject storageObject = codecs.blobInfo().encode(blobInfo);
-    final Map<StorageRpc.Option, ?> optionsMap = optionMap(blobInfo, options);
-    ResultRetryAlgorithm<?> algorithm =
-        retryAlgorithmManager.getForObjectsUpdate(storageObject, optionsMap);
+    Opts<ObjectTargetOpt> opts = Opts.unwrap(options).resolveFrom(blobInfo);
+    Map<StorageRpc.Option, ?> optionsMap = opts.getRpcOptions();
+    BlobInfo updated = opts.blobInfoMapper().apply(blobInfo.toBuilder()).build();
+    StorageObject pb = codecs.blobInfo().encode(updated);
+    ResultRetryAlgorithm<?> algorithm = retryAlgorithmManager.getForObjectsUpdate(pb, optionsMap);
     return run(
         algorithm,
-        () -> storageRpc.patch(storageObject, optionsMap),
+        () -> storageRpc.patch(pb, optionsMap),
         (x) -> {
           BlobInfo info = Conversions.apiary().blobInfo().decode(x);
           return info.asBlob(this);
@@ -473,7 +477,7 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
   public boolean delete(String bucket, BucketSourceOption... options) {
     final com.google.api.services.storage.model.Bucket bucketPb =
         codecs.bucketInfo().encode(BucketInfo.of(bucket));
-    final Map<StorageRpc.Option, ?> optionsMap = optionMap(options);
+    ImmutableMap<StorageRpc.Option, ?> optionsMap = Opts.unwrap(options).getRpcOptions();
     ResultRetryAlgorithm<?> algorithm =
         retryAlgorithmManager.getForBucketsDelete(bucketPb, optionsMap);
     return run(algorithm, () -> storageRpc.delete(bucketPb, optionsMap), Function.identity());
@@ -487,7 +491,8 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
   @Override
   public boolean delete(BlobId blob, BlobSourceOption... options) {
     final StorageObject storageObject = codecs.blobId().encode(blob);
-    final Map<StorageRpc.Option, ?> optionsMap = optionMap(blob, options);
+    ImmutableMap<StorageRpc.Option, ?> optionsMap =
+        Opts.unwrap(options).resolveFrom(blob).getRpcOptions();
     ResultRetryAlgorithm<?> algorithm =
         retryAlgorithmManager.getForObjectsDelete(storageObject, optionsMap);
     return run(algorithm, () -> storageRpc.delete(storageObject, optionsMap), Function.identity());
@@ -502,6 +507,7 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
   public Blob compose(final ComposeRequest composeRequest) {
     final List<StorageObject> sources =
         Lists.newArrayListWithCapacity(composeRequest.getSourceBlobs().size());
+    BlobInfo target = composeRequest.getTarget();
     for (ComposeRequest.SourceBlob sourceBlob : composeRequest.getSourceBlobs()) {
       sources.add(
           codecs
@@ -509,22 +515,18 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
               .encode(
                   BlobInfo.newBuilder(
                           BlobId.of(
-                              composeRequest.getTarget().getBucket(),
-                              sourceBlob.getName(),
-                              sourceBlob.getGeneration()))
+                              target.getBucket(), sourceBlob.getName(), sourceBlob.getGeneration()))
                       .build()));
     }
-    final StorageObject target = codecs.blobInfo().encode(composeRequest.getTarget());
-    final Map<StorageRpc.Option, ?> targetOptions =
-        optionMap(
-            composeRequest.getTarget().getGeneration(),
-            composeRequest.getTarget().getMetageneration(),
-            composeRequest.getTargetOptions());
+    Opts<ObjectTargetOpt> targetOpts =
+        Opts.unwrap(composeRequest.getTargetOptions()).resolveFrom(target);
+    StorageObject targetPb = codecs.blobInfo().encode(composeRequest.getTarget());
+    Map<StorageRpc.Option, ?> targetOptions = targetOpts.getRpcOptions();
     ResultRetryAlgorithm<?> algorithm =
-        retryAlgorithmManager.getForObjectsCompose(sources, target, targetOptions);
+        retryAlgorithmManager.getForObjectsCompose(sources, targetPb, targetOptions);
     return run(
         algorithm,
-        () -> storageRpc.compose(sources, target, targetOptions),
+        () -> storageRpc.compose(sources, targetPb, targetOptions),
         (x) -> {
           BlobInfo info = Conversions.apiary().blobInfo().decode(x);
           return info.asBlob(this);
@@ -533,22 +535,23 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
 
   @Override
   public CopyWriter copy(final CopyRequest copyRequest) {
-    final StorageObject source = codecs.blobId().encode(copyRequest.getSource());
-    final Map<StorageRpc.Option, ?> sourceOptions =
-        optionMap(
-            copyRequest.getSource().getGeneration(), null, copyRequest.getSourceOptions(), true);
-    final StorageObject targetObject = codecs.blobInfo().encode(copyRequest.getTarget());
-    final Map<StorageRpc.Option, ?> targetOptions =
-        optionMap(
-            copyRequest.getTarget().getGeneration(),
-            copyRequest.getTarget().getMetageneration(),
-            copyRequest.getTargetOptions());
+    BlobId source = copyRequest.getSource();
+    BlobInfo target = copyRequest.getTarget();
+    Opts<ObjectSourceOpt> sourceOpts =
+        Opts.unwrap(copyRequest.getSourceOptions()).resolveFrom(source).projectAsSource();
+    Opts<ObjectTargetOpt> targetOpts =
+        Opts.unwrap(copyRequest.getTargetOptions()).resolveFrom(target);
+
+    StorageObject sourcePb = codecs.blobId().encode(source);
+    StorageObject targetPb = codecs.blobInfo().encode(target);
+    ImmutableMap<StorageRpc.Option, ?> sourceOptions = sourceOpts.getRpcOptions();
+    ImmutableMap<StorageRpc.Option, ?> targetOptions = targetOpts.getRpcOptions();
     RewriteRequest rewriteRequest =
         new RewriteRequest(
-            source,
+            sourcePb,
             sourceOptions,
             copyRequest.overrideInfo(),
-            targetObject,
+            targetPb,
             targetOptions,
             copyRequest.getMegabytesCopiedPerChunk());
     ResultRetryAlgorithm<?> algorithm = retryAlgorithmManager.getForObjectsRewrite(rewriteRequest);
@@ -566,7 +569,9 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
   @Override
   public byte[] readAllBytes(BlobId blob, BlobSourceOption... options) {
     final StorageObject storageObject = codecs.blobId().encode(blob);
-    final Map<StorageRpc.Option, ?> optionsMap = optionMap(blob, options);
+    Opts<ObjectSourceOpt> unwrap = Opts.unwrap(options);
+    Opts<ObjectSourceOpt> resolve = unwrap.resolveFrom(blob);
+    ImmutableMap<StorageRpc.Option, ?> optionsMap = resolve.getRpcOptions();
     ResultRetryAlgorithm<?> algorithm =
         retryAlgorithmManager.getForObjectsGet(storageObject, optionsMap);
     return run(algorithm, () -> storageRpc.load(storageObject, optionsMap), Function.identity());
@@ -579,13 +584,14 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
 
   @Override
   public ReadChannel reader(String bucket, String blob, BlobSourceOption... options) {
-    Map<StorageRpc.Option, ?> optionsMap = optionMap(options);
+    ImmutableMap<StorageRpc.Option, ?> optionsMap = Opts.unwrap(options).getRpcOptions();
     return new BlobReadChannel(getOptions(), BlobId.of(bucket, blob), optionsMap);
   }
 
   @Override
   public ReadChannel reader(BlobId blob, BlobSourceOption... options) {
-    Map<StorageRpc.Option, ?> optionsMap = optionMap(blob, options);
+    ImmutableMap<StorageRpc.Option, ?> optionsMap =
+        Opts.unwrap(options).resolveFrom(blob).getRpcOptions();
     return new BlobReadChannel(getOptions(), blob, optionsMap);
   }
 
@@ -602,22 +608,35 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
   public void downloadTo(BlobId blob, OutputStream outputStream, BlobSourceOption... options) {
     final CountingOutputStream countingOutputStream = new CountingOutputStream(outputStream);
     final StorageObject pb = codecs.blobId().encode(blob);
-    final Map<StorageRpc.Option, ?> requestOptions = optionMap(blob, options);
-    ResultRetryAlgorithm<?> algorithm = retryAlgorithmManager.getForObjectsGet(pb, requestOptions);
+    ImmutableMap<StorageRpc.Option, ?> optionsMap =
+        Opts.unwrap(options).resolveFrom(blob).getRpcOptions();
+    ResultRetryAlgorithm<?> algorithm = retryAlgorithmManager.getForObjectsGet(pb, optionsMap);
     run(
         algorithm,
         callable(
             () -> {
               storageRpc.read(
-                  pb, requestOptions, countingOutputStream.getCount(), countingOutputStream);
+                  pb, optionsMap, countingOutputStream.getCount(), countingOutputStream);
             }),
         Function.identity());
   }
 
   @Override
   public BlobWriteChannel writer(BlobInfo blobInfo, BlobWriteOption... options) {
-    Tuple<BlobInfo, BlobTargetOption[]> targetOptions = BlobTargetOption.convert(blobInfo, options);
-    return writer(targetOptions.x(), targetOptions.y());
+    Opts<ObjectTargetOpt> opts = Opts.unwrap(options).resolveFrom(blobInfo);
+    final Map<StorageRpc.Option, ?> optionsMap = opts.getRpcOptions();
+    BlobInfo.Builder builder = blobInfo.toBuilder().setMd5(null).setCrc32c(null);
+    BlobInfo updated = opts.blobInfoMapper().apply(builder).build();
+    return BlobWriteChannel.newBuilder()
+        .setStorageOptions(getOptions())
+        .setUploadIdSupplier(
+            ResumableMedia.startUploadForBlobInfo(
+                getOptions(),
+                updated,
+                optionsMap,
+                retryAlgorithmManager.getForResumableUploadSessionCreate(optionsMap)))
+        .setAlgorithmForWrite(retryAlgorithmManager.getForResumableUploadSessionWrite(optionsMap))
+        .build();
   }
 
   @Override
@@ -632,21 +651,8 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
         .setUploadIdSupplier(
             ResumableMedia.startUploadForSignedUrl(
                 getOptions(), signedURL, forResumableUploadSessionCreate))
-        .setAlgorithmForWrite(retryAlgorithmManager.getForResumableUploadSessionWrite(optionMap()))
-        .build();
-  }
-
-  private BlobWriteChannel writer(BlobInfo blobInfo, BlobTargetOption... options) {
-    final Map<StorageRpc.Option, ?> optionsMap = optionMap(blobInfo, options);
-    return BlobWriteChannel.newBuilder()
-        .setStorageOptions(getOptions())
-        .setUploadIdSupplier(
-            ResumableMedia.startUploadForBlobInfo(
-                getOptions(),
-                blobInfo,
-                optionsMap,
-                retryAlgorithmManager.getForResumableUploadSessionCreate(optionsMap)))
-        .setAlgorithmForWrite(retryAlgorithmManager.getForResumableUploadSessionWrite(optionsMap))
+        .setAlgorithmForWrite(
+            retryAlgorithmManager.getForResumableUploadSessionWrite(Collections.emptyMap()))
         .build();
   }
 
@@ -1126,7 +1132,7 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
   @Override
   public Acl getAcl(final String bucket, final Entity entity, BucketSourceOption... options) {
     String pb = codecs.entity().encode(entity);
-    final Map<StorageRpc.Option, ?> optionsMap = optionMap(options);
+    ImmutableMap<StorageRpc.Option, ?> optionsMap = Opts.unwrap(options).getRpcOptions();
     ResultRetryAlgorithm<?> algorithm = retryAlgorithmManager.getForBucketAclGet(pb, optionsMap);
     return run(
         algorithm, () -> storageRpc.getAcl(bucket, pb, optionsMap), codecs.bucketAcl()::decode);
@@ -1141,7 +1147,7 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
   public boolean deleteAcl(
       final String bucket, final Entity entity, BucketSourceOption... options) {
     final String pb = codecs.entity().encode(entity);
-    final Map<StorageRpc.Option, ?> optionsMap = optionMap(options);
+    ImmutableMap<StorageRpc.Option, ?> optionsMap = Opts.unwrap(options).getRpcOptions();
     ResultRetryAlgorithm<?> algorithm = retryAlgorithmManager.getForBucketAclDelete(pb, optionsMap);
     return run(algorithm, () -> storageRpc.deleteAcl(bucket, pb, optionsMap), Function.identity());
   }
@@ -1154,7 +1160,7 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
   @Override
   public Acl createAcl(String bucket, Acl acl, BucketSourceOption... options) {
     final BucketAccessControl aclPb = codecs.bucketAcl().encode(acl).setBucket(bucket);
-    final Map<StorageRpc.Option, ?> optionsMap = optionMap(options);
+    ImmutableMap<StorageRpc.Option, ?> optionsMap = Opts.unwrap(options).getRpcOptions();
     ResultRetryAlgorithm<?> algorithm =
         retryAlgorithmManager.getForBucketAclCreate(aclPb, optionsMap);
     return run(
@@ -1169,7 +1175,7 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
   @Override
   public Acl updateAcl(String bucket, Acl acl, BucketSourceOption... options) {
     final BucketAccessControl aclPb = codecs.bucketAcl().encode(acl).setBucket(bucket);
-    final Map<StorageRpc.Option, ?> optionsMap = optionMap(options);
+    ImmutableMap<StorageRpc.Option, ?> optionsMap = Opts.unwrap(options).getRpcOptions();
     ResultRetryAlgorithm<?> algorithm =
         retryAlgorithmManager.getForBucketAclUpdate(aclPb, optionsMap);
     return run(algorithm, () -> storageRpc.patchAcl(aclPb, optionsMap), codecs.bucketAcl()::decode);
@@ -1182,7 +1188,7 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
 
   @Override
   public List<Acl> listAcls(final String bucket, BucketSourceOption... options) {
-    final Map<StorageRpc.Option, ?> optionsMap = optionMap(options);
+    ImmutableMap<StorageRpc.Option, ?> optionsMap = Opts.unwrap(options).getRpcOptions();
     ResultRetryAlgorithm<?> algorithm =
         retryAlgorithmManager.getForBucketAclList(bucket, optionsMap);
     return run(
@@ -1316,24 +1322,25 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
   public HmacKey createHmacKey(
       final ServiceAccount serviceAccount, final CreateHmacKeyOption... options) {
     String pb = serviceAccount.getEmail();
-    Map<StorageRpc.Option, ?> optionsMap = optionMap(options);
+    ImmutableMap<StorageRpc.Option, ?> optionsMap = Opts.unwrap(options).getRpcOptions();
     ResultRetryAlgorithm<?> algorithm = retryAlgorithmManager.getForHmacKeyCreate(pb, optionsMap);
     return run(algorithm, () -> storageRpc.createHmacKey(pb, optionsMap), codecs.hmacKey()::decode);
   }
 
   @Override
   public Page<HmacKeyMetadata> listHmacKeys(ListHmacKeysOption... options) {
-    return listHmacKeys(getOptions(), retryAlgorithmManager, optionMap(options));
+    ImmutableMap<StorageRpc.Option, ?> optionsMap = Opts.unwrap(options).getRpcOptions();
+    return listHmacKeys(getOptions(), retryAlgorithmManager, optionsMap);
   }
 
   @Override
   public HmacKeyMetadata getHmacKey(final String accessId, final GetHmacKeyOption... options) {
-    Map<StorageRpc.Option, ?> optionsMap = optionMap(options);
+    ImmutableMap<StorageRpc.Option, ?> optionsMap = Opts.unwrap(options).getRpcOptions();
     ResultRetryAlgorithm<?> algorithm =
         retryAlgorithmManager.getForHmacKeyGet(accessId, optionsMap);
     return run(
         algorithm,
-        () -> storageRpc.getHmacKey(accessId, optionMap(options)),
+        () -> storageRpc.getHmacKey(accessId, optionsMap),
         codecs.hmacKeyMetadata()::decode);
   }
 
@@ -1341,7 +1348,7 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
       final HmacKeyMetadata hmacKeyMetadata, final UpdateHmacKeyOption... options) {
     com.google.api.services.storage.model.HmacKeyMetadata pb =
         codecs.hmacKeyMetadata().encode(hmacKeyMetadata);
-    Map<StorageRpc.Option, ?> optionsMap = optionMap(options);
+    ImmutableMap<StorageRpc.Option, ?> optionsMap = Opts.unwrap(options).getRpcOptions();
     ResultRetryAlgorithm<?> algorithm = retryAlgorithmManager.getForHmacKeyUpdate(pb, optionsMap);
     return run(
         algorithm,
@@ -1367,7 +1374,7 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
   public void deleteHmacKey(final HmacKeyMetadata metadata, final DeleteHmacKeyOption... options) {
     com.google.api.services.storage.model.HmacKeyMetadata pb =
         codecs.hmacKeyMetadata().encode(metadata);
-    Map<StorageRpc.Option, ?> optionsMap = optionMap(options);
+    ImmutableMap<StorageRpc.Option, ?> optionsMap = Opts.unwrap(options).getRpcOptions();
     ResultRetryAlgorithm<?> algorithm = retryAlgorithmManager.getForHmacKeyDelete(pb, optionsMap);
     run(
         algorithm,
@@ -1403,7 +1410,7 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
 
   @Override
   public Policy getIamPolicy(final String bucket, BucketSourceOption... options) {
-    final Map<StorageRpc.Option, ?> optionsMap = optionMap(options);
+    ImmutableMap<StorageRpc.Option, ?> optionsMap = Opts.unwrap(options).getRpcOptions();
     ResultRetryAlgorithm<?> algorithm =
         retryAlgorithmManager.getForBucketsGetIamPolicy(bucket, optionsMap);
     return run(
@@ -1416,7 +1423,7 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
   public Policy setIamPolicy(
       final String bucket, final Policy policy, BucketSourceOption... options) {
     com.google.api.services.storage.model.Policy pb = convertToApiPolicy(policy);
-    final Map<StorageRpc.Option, ?> optionsMap = optionMap(options);
+    ImmutableMap<StorageRpc.Option, ?> optionsMap = Opts.unwrap(options).getRpcOptions();
     ResultRetryAlgorithm<?> algorithm =
         retryAlgorithmManager.getForBucketsSetIamPolicy(bucket, pb, optionsMap);
     return run(
@@ -1428,7 +1435,7 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
   @Override
   public List<Boolean> testIamPermissions(
       final String bucket, final List<String> permissions, BucketSourceOption... options) {
-    final Map<StorageRpc.Option, ?> optionsMap = optionMap(options);
+    ImmutableMap<StorageRpc.Option, ?> optionsMap = Opts.unwrap(options).getRpcOptions();
     ResultRetryAlgorithm<?> algorithm =
         retryAlgorithmManager.getForBucketsTestIamPermissions(bucket, permissions, optionsMap);
     return run(
@@ -1449,7 +1456,8 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
   public Bucket lockRetentionPolicy(BucketInfo bucketInfo, BucketTargetOption... options) {
     final com.google.api.services.storage.model.Bucket bucketPb =
         codecs.bucketInfo().encode(bucketInfo);
-    final Map<StorageRpc.Option, ?> optionsMap = optionMap(bucketInfo, options);
+    final Map<StorageRpc.Option, ?> optionsMap =
+        Opts.unwrap(options).resolveFrom(bucketInfo).getRpcOptions();
     ResultRetryAlgorithm<?> algorithm =
         retryAlgorithmManager.getForBucketsLockRetentionPolicy(bucketPb, optionsMap);
     return run(
@@ -1570,82 +1578,4 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
   public HttpStorageOptions getOptions() {
     return (HttpStorageOptions) super.getOptions();
   }
-
-  private static <T> void addToOptionMap(
-      StorageRpc.Option option, T defaultValue, Map<StorageRpc.Option, Object> map) {
-    addToOptionMap(option, option, defaultValue, map);
-  }
-
-  private static <T> void addToOptionMap(
-      StorageRpc.Option getOption,
-      StorageRpc.Option putOption,
-      T defaultValue,
-      Map<StorageRpc.Option, Object> map) {
-    if (map.containsKey(getOption)) {
-      @SuppressWarnings("unchecked")
-      T value = (T) map.remove(getOption);
-      checkArgument(
-          value != null || defaultValue != null,
-          "Option " + getOption.value() + " is missing a value");
-      value = firstNonNull(value, defaultValue);
-      map.put(putOption, value);
-    }
-  }
-
-  private static Map<StorageRpc.Option, ?> optionMap(
-      Long generation, Long metaGeneration, Iterable<? extends Option> options) {
-    return optionMap(generation, metaGeneration, options, false);
-  }
-
-  private static Map<StorageRpc.Option, ?> optionMap(
-      Long generation,
-      Long metaGeneration,
-      Iterable<? extends Option> options,
-      boolean useAsSource) {
-    Map<StorageRpc.Option, Object> temp = Maps.newEnumMap(StorageRpc.Option.class);
-    for (Option option : options) {
-      Object prev = temp.put(option.getRpcOption(), option.getValue());
-      checkArgument(prev == null, "Duplicate option %s", option);
-    }
-    if (Boolean.TRUE.equals(temp.get(DELIMITER))) {
-      temp.remove(DELIMITER);
-      temp.put(DELIMITER, PATH_DELIMITER);
-    } else if (null != temp.get(DELIMITER)) {
-      temp.put(DELIMITER, temp.get(DELIMITER));
-    }
-    if (useAsSource) {
-      addToOptionMap(IF_GENERATION_MATCH, IF_SOURCE_GENERATION_MATCH, generation, temp);
-      addToOptionMap(IF_GENERATION_NOT_MATCH, IF_SOURCE_GENERATION_NOT_MATCH, generation, temp);
-      addToOptionMap(IF_METAGENERATION_MATCH, IF_SOURCE_METAGENERATION_MATCH, metaGeneration, temp);
-      addToOptionMap(
-          IF_METAGENERATION_NOT_MATCH, IF_SOURCE_METAGENERATION_NOT_MATCH, metaGeneration, temp);
-    } else {
-      addToOptionMap(IF_GENERATION_MATCH, generation, temp);
-      addToOptionMap(IF_GENERATION_NOT_MATCH, generation, temp);
-      addToOptionMap(IF_METAGENERATION_MATCH, metaGeneration, temp);
-      addToOptionMap(IF_METAGENERATION_NOT_MATCH, metaGeneration, temp);
-    }
-    return ImmutableMap.copyOf(temp);
-  }
-
-  static Map<StorageRpc.Option, ?> optionMap(Option... options) {
-    return optionMap(null, null, Arrays.asList(options));
-  }
-
-  private static Map<StorageRpc.Option, ?> optionMap(
-      Long generation, Long metaGeneration, Option... options) {
-    return optionMap(generation, metaGeneration, Arrays.asList(options));
-  }
-
-  private static Map<StorageRpc.Option, ?> optionMap(BucketInfo bucketInfo, Option... options) {
-    return optionMap(null, bucketInfo.getMetageneration(), options);
-  }
-
-  static Map<StorageRpc.Option, ?> optionMap(BlobInfo blobInfo, Option... options) {
-    return optionMap(blobInfo.getGeneration(), blobInfo.getMetageneration(), options);
-  }
-
-  static Map<StorageRpc.Option, ?> optionMap(BlobId blobId, Option... options) {
-    return optionMap(blobId.getGeneration(), null, options);
-  }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java
new file mode 100644
index 00000000..8b94b6c8
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java
@@ -0,0 +1,1438 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage;
+
+import static com.google.cloud.storage.Utils.todo;
+import static com.google.common.base.Preconditions.checkArgument;
+
+import com.google.api.gax.grpc.GrpcCallContext;
+import com.google.cloud.storage.spi.v1.StorageRpc;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.io.BaseEncoding;
+import com.google.storage.v2.GetObjectRequest;
+import java.io.Serializable;
+import java.net.FileNameMap;
+import java.net.URLConnection;
+import java.security.Key;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Locale;
+import java.util.Objects;
+import java.util.function.Function;
+import java.util.function.Predicate;
+import java.util.stream.Stream;
+import javax.crypto.spec.SecretKeySpec;
+import org.checkerframework.checker.nullness.qual.NonNull;
+
+/**
+ * The set of all "Options" we currently support for per-call parameters.
+ *
+ * <p>Most often, each of the respective types correspond to one of the parameters from <a
+ * target="_blank" rel="noopener noreferrer"
+ * href="https://cloud.google.com/storage/docs/json_api/v1/parameters">HTTP headers and common query
+ * string parameters for JSON</a>. In the case of gRPC, sometimes the parameters are in the specific
+ * request message or in grpc metadata.
+ */
+final class UnifiedOpts {
+
+  /** Base interface type for each of the new options we're supporting. */
+  interface Opt extends Serializable {}
+
+  /**
+   * A specialization of {@link java.util.function.UnaryOperator} which maintains its lower type for
+   * {@link #identity()} and {@link #andThen(Mapper)}.
+   */
+  @FunctionalInterface
+  interface Mapper<T> {
+    T apply(T t);
+
+    default Mapper<T> andThen(Mapper<T> then) {
+      return t -> then.apply(apply(t));
+    }
+
+    static <T> Mapper<T> identity() {
+      return t -> t;
+    }
+  }
+
+  /** Base, marker interface of those Opts which represent a get/read/origin type relationship. */
+  private interface SourceOpt extends Opt {}
+
+  /**
+   * Base, marker interface of those Opts which represent a set/write/destination type relationship.
+   */
+  private interface TargetOpt extends Opt {}
+
+  /** Base, marker interface of those Opts which apply to listing operations. */
+  private interface ListOpt extends Opt {}
+
+  /** Marker interface of those Opts which are applicable to Bucket operations. */
+  private interface ApplicableBucket {}
+
+  /** Marker interface of those Opts which are applicable to Object/Blob operations. */
+  private interface ApplicableObject {}
+
+  /** Marker interface of those Opts which are applicable to HmacKey operations. */
+  private interface ApplicableHmacKey {}
+
+  /** Base interface for those Opts which may expose their values via gRPC Metadata */
+  private interface GrpcMetadataMapper {
+    default Mapper<GrpcCallContext> getGrpcMetadataMapper() {
+      return Mapper.identity();
+    }
+  }
+
+  /** Base interface for those Opts which are applicable to Object List operations */
+  interface ObjectListOpt extends GrpcMetadataMapper, ListOpt, ApplicableObject {}
+
+  /**
+   * Base interface for those Opts which are applicable to Object Source (get/read/origin
+   * relationship) operations
+   */
+  interface ObjectSourceOpt extends GrpcMetadataMapper, SourceOpt, ApplicableObject {
+    Mapper<GetObjectRequest.Builder> getObject();
+  }
+
+  /**
+   * Base interface for those Opts which are applicable to Object Target (set/write/destination
+   * relationship) operations
+   */
+  interface ObjectTargetOpt extends GrpcMetadataMapper, TargetOpt, ApplicableObject {
+    default Mapper<BlobInfo.Builder> blobInfo() {
+      return Mapper.identity();
+    }
+  }
+
+  /** Base interface for those Opts which are applicable to Bucket List operations */
+  interface BucketListOpt extends GrpcMetadataMapper, ListOpt, ApplicableBucket {}
+
+  /**
+   * Base interface for those Opts which are applicable to Bucket Source (get/read/origin
+   * relationship) operations
+   */
+  interface BucketSourceOpt extends GrpcMetadataMapper, SourceOpt, ApplicableBucket {}
+
+  /**
+   * Base interface for those Opts which are applicable to Bucket Target (set/write/destination
+   * relationship) operations
+   */
+  interface BucketTargetOpt extends GrpcMetadataMapper, TargetOpt, ApplicableBucket {}
+
+  /** Base interface for those Opts which are applicable to HmacKey List operations */
+  interface HmacKeyListOpt extends GrpcMetadataMapper, ListOpt, ApplicableHmacKey {}
+
+  /**
+   * Base interface for those Opts which are applicable to HmacKey Source (get/read/origin
+   * relationship) operations
+   */
+  interface HmacKeySourceOpt extends GrpcMetadataMapper, SourceOpt, ApplicableHmacKey {}
+
+  /**
+   * Base interface for those Opts which are applicable to HmacKey Target (set/write/destination
+   * relationship) operations
+   */
+  interface HmacKeyTargetOpt extends GrpcMetadataMapper, TargetOpt, ApplicableHmacKey {}
+
+  /**
+   * Some Options have a corresponding "SOURCE" version, this interface provide a construct for
+   * accessing an projecting those Opts which can be turned into a "SOURCE" version.
+   */
+  interface ProjectAsSource<O extends Opt> {
+    O asSource();
+  }
+
+  /**
+   * This class extends off {@link ObjectSourceOpt} and {@link ObjectTargetOpt} in order to satisfy
+   * some the shimming constraints of the subclasses of {@link OptionShim}.
+   *
+   * <p>All the methods from these parent interfaces will NEVER be called, and are stubbed simply to
+   * satisfy the need for them to be declared. They are stubbed to use identity methods so that if
+   * they somehow do ever leak through and are called they won't cause issue for customers.
+   *
+   * <p>If/when we're able to remove all the {@link Option} classes, this interface should be
+   * refactored to remove the inheritance, instead providing an explicit pre-processing phase to opt
+   * resolution.
+   */
+  interface ObjectOptExtractor<O extends Opt> extends Opt, ObjectSourceOpt, ObjectTargetOpt {
+    O extractFromBlobInfo(BlobInfo info);
+
+    O extractFromBlobId(BlobId id);
+
+    @Override
+    default Mapper<GrpcCallContext> getGrpcMetadataMapper() {
+      return Mapper.identity();
+    }
+
+    @Override
+    default Mapper<GetObjectRequest.Builder> getObject() {
+      return Mapper.identity();
+    }
+  }
+
+  /**
+   * This class extends off {@link ObjectSourceOpt} and {@link ObjectTargetOpt} in order to satisfy
+   * some the shimming constraints of the subclasses of {@link OptionShim}.
+   *
+   * <p>All the methods from these parent interfaces will NEVER be called, and are stubbed simply to
+   * satisfy the need for them to be declared. They are stubbed to use identity methods so that if
+   * they somehow do ever leak through and are called they won't cause issue for customers.
+   *
+   * <p>If/when we're able to remove all the {@link Option} classes, this interface should be
+   * refactored to remove the inheritance, instead providing an explicit pre-processing phase to opt
+   * resolution.
+   */
+  interface BucketOptExtractor<O extends Opt> extends Opt, BucketSourceOpt, BucketTargetOpt {
+    O extractFromBucketInfo(BucketInfo info);
+
+    @Override
+    default Mapper<GrpcCallContext> getGrpcMetadataMapper() {
+      return Mapper.identity();
+    }
+  }
+
+  /* --
+  Factory methods for each of the supported Opts, along with some of their requisite
+  compatibility overloads
+  -- */
+
+  static Crc32cMatch crc32cMatch(String i) {
+    return new Crc32cMatch(i);
+  }
+
+  static Delimiter currentDirectory() {
+    return new Delimiter("/");
+  }
+
+  static DecryptionKey decryptionKey(String s) {
+    return new DecryptionKey(new SecretKeySpec(BaseEncoding.base64().decode(s), "AES256"));
+  }
+
+  static DecryptionKey decryptionKey(Key k) {
+    return new DecryptionKey(k);
+  }
+
+  static Delimiter delimiter(String s) {
+    return new Delimiter(s);
+  }
+
+  static DetectContentType detectContentType() {
+    return DetectContentType.INSTANCE;
+  }
+
+  static DisableGzipContent disableGzipContent() {
+    return new DisableGzipContent(true);
+  }
+
+  static GenerationMatch doesNotExist() {
+    return new GenerationMatch(0);
+  }
+
+  static EncryptionKey encryptionKey(String s) {
+    return new EncryptionKey(new SecretKeySpec(BaseEncoding.base64().decode(s), "AES256"));
+  }
+
+  static EncryptionKey encryptionKey(Key k) {
+    return new EncryptionKey(k);
+  }
+
+  static EndOffset endOffset(String s) {
+    return new EndOffset(s);
+  }
+
+  @Deprecated
+  static Fields fields(String s) {
+    // TODO: do we care if the string provided is empty?
+    return new Fields(s);
+  }
+
+  static GenerationMatch generationMatch(long l) {
+    return new GenerationMatch(l);
+  }
+
+  static GenerationNotMatch generationNotMatch(long l) {
+    return new GenerationNotMatch(l);
+  }
+
+  static KmsKeyName kmsKeyName(String s) {
+    return new KmsKeyName(s);
+  }
+
+  static Md5Match md5Match(String s) {
+    return new Md5Match(s);
+  }
+
+  static MetagenerationMatch metagenerationMatch(long l) {
+    return new MetagenerationMatch(l);
+  }
+
+  static MetagenerationNotMatch metagenerationNotMatch(long l) {
+    return new MetagenerationNotMatch(l);
+  }
+
+  static PageSize pageSize(long l) {
+    return new PageSize(l);
+  }
+
+  static PageToken pageToken(String s) {
+    return new PageToken(s);
+  }
+
+  static PredefinedAcl predefinedAcl(Storage.PredefinedAcl p) {
+    return new PredefinedAcl(p.getEntry());
+  }
+
+  static PredefinedDefaultObjectAcl predefinedDefaultObjectAcl(Storage.PredefinedAcl p) {
+    return new PredefinedDefaultObjectAcl(p.getEntry());
+  }
+
+  static Prefix prefix(String s) {
+    return new Prefix(s);
+  }
+
+  static ProjectId projectId(String s) {
+    return new ProjectId(s);
+  }
+
+  static Projection projection(String s) {
+    return new Projection(s);
+  }
+
+  static RequestedPolicyVersion requestedPolicyVersion(long l) {
+    return new RequestedPolicyVersion(l);
+  }
+
+  static ReturnRawInputStream returnRawInputStream(boolean b) {
+    return new ReturnRawInputStream(b);
+  }
+
+  static ServiceAccount serviceAccount(com.google.cloud.storage.ServiceAccount s) {
+    return new ServiceAccount(s.getEmail());
+  }
+
+  static ShowDeletedKeys showDeletedKeys(boolean b) {
+    return new ShowDeletedKeys(b);
+  }
+
+  static StartOffset startOffset(String s) {
+    return new StartOffset(s);
+  }
+
+  static UserProject userProject(String s) {
+    return new UserProject(s);
+  }
+
+  static VersionsFilter versionsFilter(boolean b) {
+    return new VersionsFilter(b);
+  }
+
+  @Deprecated
+  static GenerationMatchExtractor generationMatchExtractor() {
+    return GenerationMatchExtractor.INSTANCE;
+  }
+
+  @Deprecated
+  static GenerationNotMatchExtractor generationNotMatchExtractor() {
+    return GenerationNotMatchExtractor.INSTANCE;
+  }
+
+  @Deprecated
+  static MetagenerationMatchExtractor metagenerationMatchExtractor() {
+    return MetagenerationMatchExtractor.INSTANCE;
+  }
+
+  @Deprecated
+  static MetagenerationNotMatchExtractor metagenerationNotMatchExtractor() {
+    return MetagenerationNotMatchExtractor.INSTANCE;
+  }
+
+  @Deprecated
+  static Crc32cMatchExtractor crc32cMatchExtractor() {
+    return Crc32cMatchExtractor.INSTANCE;
+  }
+
+  @Deprecated
+  static Md5MatchExtractor md5MatchExtractor() {
+    return Md5MatchExtractor.INSTANCE;
+  }
+
+  @Deprecated
+  static final class Crc32cMatch implements ObjectTargetOpt {
+    private static final long serialVersionUID = -8680237667319155418L;
+    private final String val;
+
+    private Crc32cMatch(String val) {
+      this.val = val;
+    }
+
+    @Override
+    public Mapper<BlobInfo.Builder> blobInfo() {
+      return b -> b.setCrc32c(val);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) {
+        return true;
+      }
+      if (!(o instanceof Crc32cMatch)) {
+        return false;
+      }
+      Crc32cMatch that = (Crc32cMatch) o;
+      return Objects.equals(val, that.val);
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(val);
+    }
+
+    @Override
+    public String toString() {
+      return "Crc32cMatch{val='" + val + "'}";
+    }
+  }
+
+  /** @see EncryptionKey */
+  static final class DecryptionKey extends RpcOptVal<Key> implements ObjectSourceOpt {
+    private static final long serialVersionUID = -4583830666730826055L;
+
+    private DecryptionKey(Key val) {
+      super(StorageRpc.Option.CUSTOMER_SUPPLIED_KEY, val);
+    }
+
+    @Override
+    public Mapper<ImmutableMap.Builder<StorageRpc.Option, Object>> mapper() {
+      return b ->
+          b.put(
+              StorageRpc.Option.CUSTOMER_SUPPLIED_KEY,
+              BaseEncoding.base64().encode(val.getEncoded()));
+    }
+
+    @Override
+    public Mapper<GetObjectRequest.Builder> getObject() {
+      return todo();
+    }
+  }
+
+  static final class Delimiter extends RpcOptVal<String> implements ObjectListOpt {
+    private static final long serialVersionUID = 8639409337839854122L;
+
+    private Delimiter(String val) {
+      super(StorageRpc.Option.DELIMITER, val);
+    }
+  }
+
+  static final class DisableGzipContent extends RpcOptVal<@NonNull Boolean>
+      implements ObjectTargetOpt {
+    private static final long serialVersionUID = -8673296387131912651L;
+
+    private DisableGzipContent(boolean val) {
+      super(StorageRpc.Option.IF_DISABLE_GZIP_CONTENT, val);
+    }
+  }
+
+  /** @see DecryptionKey */
+  static final class EncryptionKey extends RpcOptVal<Key> implements ObjectTargetOpt {
+    private static final long serialVersionUID = 7563566358784847875L;
+
+    private EncryptionKey(Key val) {
+      super(StorageRpc.Option.CUSTOMER_SUPPLIED_KEY, val);
+    }
+
+    @Override
+    public Mapper<ImmutableMap.Builder<StorageRpc.Option, Object>> mapper() {
+      return b ->
+          b.put(
+              StorageRpc.Option.CUSTOMER_SUPPLIED_KEY,
+              BaseEncoding.base64().encode(val.getEncoded()));
+    }
+  }
+
+  /** @see StartOffset */
+  static final class EndOffset extends RpcOptVal<String> implements ObjectListOpt {
+    private static final long serialVersionUID = -4571919566602569625L;
+
+    private EndOffset(String val) {
+      super(StorageRpc.Option.END_OFF_SET, val);
+    }
+  }
+
+  static final class Fields extends RpcOptVal<String>
+      implements ObjectSourceOpt, ObjectListOpt, BucketSourceOpt, BucketListOpt {
+    private static final long serialVersionUID = -6861404961336468409L;
+
+    private Fields(String val) {
+      super(StorageRpc.Option.FIELDS, val);
+    }
+
+    @Override
+    public Mapper<GetObjectRequest.Builder> getObject() {
+      return Mapper.identity();
+    }
+  }
+
+  /**
+   * @see GenerationNotMatch
+   * @see SourceGenerationMatch
+   */
+  static final class GenerationMatch extends RpcOptVal<@NonNull Long>
+      implements ObjectSourceOpt, ObjectTargetOpt, ProjectAsSource<SourceGenerationMatch> {
+    private static final long serialVersionUID = -2356341166190897807L;
+
+    private GenerationMatch(long val) {
+      super(StorageRpc.Option.IF_GENERATION_MATCH, val);
+    }
+
+    @Override
+    public Mapper<GetObjectRequest.Builder> getObject() {
+      return b -> b.setIfGenerationMatch(val);
+    }
+
+    @Override
+    public SourceGenerationMatch asSource() {
+      return new SourceGenerationMatch(val);
+    }
+  }
+
+  /**
+   * @see GenerationMatch
+   * @see SourceGenerationNotMatch
+   */
+  static final class GenerationNotMatch extends RpcOptVal<@NonNull Long>
+      implements ObjectSourceOpt, ObjectTargetOpt, ProjectAsSource<SourceGenerationNotMatch> {
+    private static final long serialVersionUID = -6055322302594035351L;
+
+    private GenerationNotMatch(long val) {
+      super(StorageRpc.Option.IF_GENERATION_NOT_MATCH, val);
+    }
+
+    @Override
+    public Mapper<GetObjectRequest.Builder> getObject() {
+      return b -> b.setIfGenerationNotMatch(val);
+    }
+
+    @Override
+    public SourceGenerationNotMatch asSource() {
+      return new SourceGenerationNotMatch(val);
+    }
+  }
+
+  static final class KmsKeyName extends RpcOptVal<String> implements ObjectTargetOpt {
+    private static final long serialVersionUID = -3337302773119117013L;
+
+    private KmsKeyName(String val) {
+      super(StorageRpc.Option.KMS_KEY_NAME, val);
+    }
+  }
+
+  @Deprecated
+  static final class Md5Match implements ObjectTargetOpt {
+    private static final long serialVersionUID = -4497633005169883788L;
+    private final String val;
+
+    private Md5Match(String val) {
+      this.val = val;
+    }
+
+    @Override
+    public Mapper<BlobInfo.Builder> blobInfo() {
+      return b -> b.setMd5(val);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) {
+        return true;
+      }
+      if (!(o instanceof Md5Match)) {
+        return false;
+      }
+      Md5Match md5Match = (Md5Match) o;
+      return Objects.equals(val, md5Match.val);
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(val);
+    }
+
+    @Override
+    public String toString() {
+      return "Md5Match{val='" + val + "'}";
+    }
+  }
+
+  /**
+   * @see MetagenerationNotMatch
+   * @see SourceMetagenerationMatch
+   */
+  static final class MetagenerationMatch extends RpcOptVal<@NonNull Long>
+      implements BucketSourceOpt,
+          BucketTargetOpt,
+          ObjectSourceOpt,
+          ObjectTargetOpt,
+          ProjectAsSource<SourceMetagenerationMatch> {
+    private static final long serialVersionUID = 5508074897592817147L;
+
+    private MetagenerationMatch(long val) {
+      super(StorageRpc.Option.IF_METAGENERATION_MATCH, val);
+    }
+
+    @Override
+    public Mapper<GetObjectRequest.Builder> getObject() {
+      return b -> b.setIfMetagenerationMatch(val);
+    }
+
+    @Override
+    public SourceMetagenerationMatch asSource() {
+      return new SourceMetagenerationMatch(val);
+    }
+  }
+
+  /**
+   * @see MetagenerationMatch
+   * @see SourceMetagenerationNotMatch
+   */
+  static final class MetagenerationNotMatch extends RpcOptVal<@NonNull Long>
+      implements BucketSourceOpt,
+          BucketTargetOpt,
+          ObjectSourceOpt,
+          ObjectTargetOpt,
+          ProjectAsSource<SourceMetagenerationNotMatch> {
+    private static final long serialVersionUID = 6869928996186950306L;
+
+    private MetagenerationNotMatch(long val) {
+      super(StorageRpc.Option.IF_METAGENERATION_NOT_MATCH, val);
+    }
+
+    @Override
+    public Mapper<GetObjectRequest.Builder> getObject() {
+      return b -> b.setIfMetagenerationNotMatch(val);
+    }
+
+    @Override
+    public SourceMetagenerationNotMatch asSource() {
+      return new SourceMetagenerationNotMatch(val);
+    }
+  }
+
+  static final class PageSize extends RpcOptVal<@NonNull Long>
+      implements BucketListOpt, ObjectListOpt, HmacKeyListOpt {
+    private static final long serialVersionUID = -3510673708181397881L;
+
+    private PageSize(long val) {
+      super(StorageRpc.Option.MAX_RESULTS, val);
+    }
+  }
+
+  static final class PageToken extends RpcOptVal<String>
+      implements BucketListOpt, ObjectListOpt, HmacKeyListOpt {
+    private static final long serialVersionUID = -542427084922230782L;
+
+    private PageToken(String val) {
+      super(StorageRpc.Option.PAGE_TOKEN, val);
+    }
+  }
+
+  static final class PredefinedAcl extends RpcOptVal<String>
+      implements BucketTargetOpt, ObjectTargetOpt {
+    private static final long serialVersionUID = 4189588503372535057L;
+
+    private PredefinedAcl(String val) {
+      super(StorageRpc.Option.PREDEFINED_ACL, val);
+    }
+  }
+
+  static final class PredefinedDefaultObjectAcl extends RpcOptVal<String>
+      implements BucketTargetOpt {
+    private static final long serialVersionUID = 6598022065653572605L;
+
+    private PredefinedDefaultObjectAcl(String val) {
+      super(StorageRpc.Option.PREDEFINED_DEFAULT_OBJECT_ACL, val);
+    }
+  }
+
+  static final class Prefix extends RpcOptVal<String> implements BucketListOpt, ObjectListOpt {
+    private static final long serialVersionUID = 155278267048093608L;
+
+    private Prefix(String val) {
+      super(StorageRpc.Option.PREFIX, val);
+    }
+  }
+
+  /**
+   * This is a required property of hmac related operations. Preferably, we'd be able to push the
+   * defaulting to the creation of a new instance of one of the model objects
+   */
+  @Deprecated
+  static final class ProjectId extends RpcOptVal<String>
+      implements HmacKeySourceOpt, HmacKeyTargetOpt, HmacKeyListOpt {
+    private static final long serialVersionUID = 1471462503030451598L;
+
+    private ProjectId(String val) {
+      super(StorageRpc.Option.PROJECT_ID, val);
+    }
+  }
+
+  static final class Projection extends RpcOptVal<String> implements BucketTargetOpt {
+    private static final long serialVersionUID = -1260415089938322394L;
+
+    private Projection(String val) {
+      super(StorageRpc.Option.PROJECTION, val);
+    }
+  }
+
+  /**
+   * @see GenerationMatch
+   * @see SourceGenerationNotMatch
+   */
+  static final class SourceGenerationMatch extends RpcOptVal<@NonNull Long>
+      implements ObjectSourceOpt, ObjectTargetOpt {
+    private static final long serialVersionUID = 5530465094492461956L;
+
+    private SourceGenerationMatch(@NonNull Long val) {
+      super(StorageRpc.Option.IF_SOURCE_GENERATION_MATCH, val);
+    }
+
+    @Override
+    public Mapper<GetObjectRequest.Builder> getObject() {
+      return Mapper.identity();
+    }
+  }
+
+  /**
+   * @see GenerationNotMatch
+   * @see SourceGenerationMatch
+   */
+  static final class SourceGenerationNotMatch extends RpcOptVal<@NonNull Long>
+      implements ObjectSourceOpt, ObjectTargetOpt {
+    private static final long serialVersionUID = 313414895558156715L;
+
+    private SourceGenerationNotMatch(@NonNull Long val) {
+      super(StorageRpc.Option.IF_SOURCE_GENERATION_NOT_MATCH, val);
+    }
+
+    @Override
+    public Mapper<GetObjectRequest.Builder> getObject() {
+      return Mapper.identity();
+    }
+  }
+
+  /**
+   * @see MetagenerationMatch
+   * @see SourceMetagenerationNotMatch
+   */
+  static final class SourceMetagenerationMatch extends RpcOptVal<@NonNull Long>
+      implements BucketSourceOpt, BucketTargetOpt, ObjectSourceOpt, ObjectTargetOpt {
+    private static final long serialVersionUID = -3643340315457580094L;
+
+    private SourceMetagenerationMatch(@NonNull Long val) {
+      super(StorageRpc.Option.IF_SOURCE_METAGENERATION_MATCH, val);
+    }
+
+    @Override
+    public Mapper<GetObjectRequest.Builder> getObject() {
+      return Mapper.identity();
+    }
+  }
+
+  /**
+   * @see MetagenerationNotMatch
+   * @see SourceMetagenerationMatch
+   */
+  static final class SourceMetagenerationNotMatch extends RpcOptVal<@NonNull Long>
+      implements BucketSourceOpt, BucketTargetOpt, ObjectSourceOpt, ObjectTargetOpt {
+    private static final long serialVersionUID = -6682202521743160969L;
+
+    private SourceMetagenerationNotMatch(@NonNull Long val) {
+      super(StorageRpc.Option.IF_SOURCE_METAGENERATION_NOT_MATCH, val);
+    }
+
+    @Override
+    public Mapper<GetObjectRequest.Builder> getObject() {
+      return Mapper.identity();
+    }
+  }
+
+  static final class RequestedPolicyVersion extends RpcOptVal<@NonNull Long>
+      implements BucketSourceOpt {
+    private static final long serialVersionUID = 7044856817626952830L;
+
+    private RequestedPolicyVersion(Long val) {
+      super(StorageRpc.Option.REQUESTED_POLICY_VERSION, val);
+    }
+  }
+
+  static final class ReturnRawInputStream extends RpcOptVal<@NonNull Boolean>
+      implements ObjectSourceOpt {
+    private static final long serialVersionUID = 505293506385742781L;
+
+    private ReturnRawInputStream(boolean val) {
+      super(StorageRpc.Option.RETURN_RAW_INPUT_STREAM, val);
+    }
+
+    @Override
+    public Mapper<GetObjectRequest.Builder> getObject() {
+      return todo();
+    }
+  }
+
+  static final class ServiceAccount extends RpcOptVal<String> implements HmacKeyListOpt {
+    private static final long serialVersionUID = 1630581690347694016L;
+
+    private ServiceAccount(String val) {
+      super(StorageRpc.Option.SERVICE_ACCOUNT_EMAIL, val);
+    }
+  }
+
+  static final class SetContentType implements ObjectTargetOpt {
+    private static final long serialVersionUID = -5358445952573187492L;
+    private final String val;
+
+    private SetContentType(String val) {
+      this.val = val;
+    }
+
+    @Override
+    public Mapper<BlobInfo.Builder> blobInfo() {
+      return b -> b.setContentType(val);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) {
+        return true;
+      }
+      if (!(o instanceof SetContentType)) {
+        return false;
+      }
+      SetContentType that = (SetContentType) o;
+      return Objects.equals(val, that.val);
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(val);
+    }
+
+    @Override
+    public String toString() {
+      return "SetContentType{val='" + val + "'}";
+    }
+  }
+
+  static final class ShowDeletedKeys extends RpcOptVal<@NonNull Boolean> implements HmacKeyListOpt {
+    private static final long serialVersionUID = 6650364639734728488L;
+
+    private ShowDeletedKeys(boolean val) {
+      super(StorageRpc.Option.SHOW_DELETED_KEYS, val);
+    }
+  }
+
+  /** @see EndOffset */
+  static final class StartOffset extends RpcOptVal<String> implements ObjectListOpt {
+    private static final long serialVersionUID = 7763387382950935370L;
+
+    private StartOffset(String val) {
+      super(StorageRpc.Option.START_OFF_SET, val);
+    }
+  }
+
+  static final class UserProject extends RpcOptVal<String>
+      implements BucketSourceOpt,
+          BucketTargetOpt,
+          BucketListOpt,
+          ObjectSourceOpt,
+          ObjectTargetOpt,
+          ObjectListOpt,
+          HmacKeySourceOpt,
+          HmacKeyTargetOpt,
+          HmacKeyListOpt {
+    private static final long serialVersionUID = -3580124936740285695L;
+
+    private UserProject(String val) {
+      super(StorageRpc.Option.USER_PROJECT, val);
+    }
+
+    @Override
+    public Mapper<GrpcCallContext> getGrpcMetadataMapper() {
+      return ctx ->
+          ctx.withExtraHeaders(ImmutableMap.of("X-Goog-User-Project", ImmutableList.of(val)));
+    }
+
+    @Override
+    public Mapper<GetObjectRequest.Builder> getObject() {
+      return Mapper.identity();
+    }
+  }
+
+  static final class VersionsFilter extends RpcOptVal<@NonNull Boolean> implements ObjectListOpt {
+    private VersionsFilter(boolean val) {
+      super(StorageRpc.Option.VERSIONS, val);
+    }
+  }
+
+  /**
+   * Attempt to extract a crc32c value from an Object. If no crc32c value is extracted the produced
+   * Opt will be an effective no-op.
+   *
+   * @see Crc32cMatch
+   * @deprecated Use {@link BlobInfo.Builder#setCrc32c(String)}
+   */
+  @Deprecated
+  static final class Crc32cMatchExtractor implements ObjectOptExtractor<ObjectTargetOpt> {
+    private static final Crc32cMatchExtractor INSTANCE = new Crc32cMatchExtractor();
+    private static final long serialVersionUID = 2222053443431466916L;
+
+    @Deprecated
+    private Crc32cMatchExtractor() {}
+
+    @Override
+    public ObjectTargetOpt extractFromBlobInfo(BlobInfo info) {
+      String crc32c = info.getCrc32c();
+      if (crc32c != null) {
+        return crc32cMatch(crc32c);
+      } else {
+        return NoOpObjectTargetOpt.INSTANCE;
+      }
+    }
+
+    @Override
+    public ObjectTargetOpt extractFromBlobId(BlobId id) {
+      return NoOpObjectTargetOpt.INSTANCE;
+    }
+
+    /** prevent java serialization from using a new instance */
+    private Object readResolve() {
+      return INSTANCE;
+    }
+  }
+
+  /**
+   * Attempt to determine the content type of an Object based on it's {@link BlobInfo#getName()}. If
+   * no name value is extracted, or the value is not a known extension the content type will be
+   * {@code application/octet-stream}
+   *
+   * @see SetContentType
+   * @see URLConnection#getFileNameMap()
+   * @see FileNameMap
+   * @deprecated Use {@link BlobInfo.Builder#setContentType(String)}
+   */
+  @Deprecated
+  static final class DetectContentType implements ObjectOptExtractor<ObjectTargetOpt> {
+    private static final DetectContentType INSTANCE = new DetectContentType();
+    private static final FileNameMap FILE_NAME_MAP = URLConnection.getFileNameMap();
+    private static final long serialVersionUID = 1L;
+
+    @Deprecated
+    private DetectContentType() {}
+
+    @Override
+    public ObjectTargetOpt extractFromBlobInfo(BlobInfo info) {
+      String contentType = info.getContentType();
+      if (contentType != null && !contentType.isEmpty()) {
+        return NoOpObjectTargetOpt.INSTANCE;
+      }
+
+      return detectForName(info.getName());
+    }
+
+    @Override
+    public ObjectTargetOpt extractFromBlobId(BlobId id) {
+      return detectForName(id.getName());
+    }
+
+    private ObjectTargetOpt detectForName(String name) {
+      if (name != null) {
+        String nameLower = name.toLowerCase(Locale.ENGLISH);
+        String contentTypeFor = FILE_NAME_MAP.getContentTypeFor(nameLower);
+        if (contentTypeFor != null) {
+          return new SetContentType(contentTypeFor);
+        }
+      }
+      return new SetContentType("application/octet-stream");
+    }
+    /** prevent java serialization from using a new instance */
+    private Object readResolve() {
+      return INSTANCE;
+    }
+  }
+
+  /**
+   * Attempt to extract a generation value from an Object. If no generation value is extracted an
+   * {@link IllegalArgumentException} will be thrown.
+   *
+   * @see GenerationMatch
+   * @deprecated Use {@link #generationMatch(long)}
+   */
+  @Deprecated
+  static final class GenerationMatchExtractor implements ObjectOptExtractor<GenerationMatch> {
+    private static final GenerationMatchExtractor INSTANCE = new GenerationMatchExtractor();
+    private static final long serialVersionUID = -4016709200925410921L;
+
+    @Deprecated
+    private GenerationMatchExtractor() {}
+
+    @Override
+    public GenerationMatch extractFromBlobInfo(BlobInfo info) {
+      Long generation = info.getGeneration();
+      checkArgument(generation != null, "Option ifGenerationMatch is missing a value");
+      return generationMatch(generation);
+    }
+
+    @Override
+    public GenerationMatch extractFromBlobId(BlobId id) {
+      Long generation = id.getGeneration();
+      checkArgument(generation != null, "Option ifGenerationMatch is missing a value");
+      return generationMatch(generation);
+    }
+    /** prevent java serialization from using a new instance */
+    private Object readResolve() {
+      return INSTANCE;
+    }
+  }
+
+  /**
+   * Attempt to extract a generation value from an Object. If no generation value is extracted an
+   * {@link IllegalArgumentException} will be thrown.
+   *
+   * @see GenerationNotMatch
+   * @deprecated Use {@link #generationNotMatch(long)}
+   */
+  @Deprecated
+  static final class GenerationNotMatchExtractor implements ObjectOptExtractor<GenerationNotMatch> {
+    private static final GenerationNotMatchExtractor INSTANCE = new GenerationNotMatchExtractor();
+    private static final long serialVersionUID = 2419121370772040679L;
+
+    @Deprecated
+    private GenerationNotMatchExtractor() {}
+
+    @Override
+    public GenerationNotMatch extractFromBlobInfo(BlobInfo info) {
+      Long generation = info.getGeneration();
+      checkArgument(generation != null, "Option ifGenerationNotMatch is missing a value");
+      return generationNotMatch(generation);
+    }
+
+    @Override
+    public GenerationNotMatch extractFromBlobId(BlobId id) {
+      Long generation = id.getGeneration();
+      checkArgument(generation != null, "Option ifGenerationNotMatch is missing a value");
+      return generationNotMatch(generation);
+    }
+    /** prevent java serialization from using a new instance */
+    private Object readResolve() {
+      return INSTANCE;
+    }
+  }
+
+  /**
+   * Attempt to extract an md5 value from an Object. If no md5 value is extracted the produced Opt
+   * will be an effective no-op.
+   *
+   * @see Md5Match
+   * @deprecated Use {@link BlobInfo.Builder#setMd5(String)}
+   */
+  @Deprecated
+  static final class Md5MatchExtractor implements ObjectOptExtractor<ObjectTargetOpt> {
+    private static final Md5MatchExtractor INSTANCE = new Md5MatchExtractor();
+    private static final long serialVersionUID = -227445210555345030L;
+
+    @Deprecated
+    private Md5MatchExtractor() {}
+
+    @Override
+    public ObjectTargetOpt extractFromBlobInfo(BlobInfo info) {
+      String md5 = info.getMd5();
+      if (md5 != null) {
+        return md5Match(md5);
+      } else {
+        return NoOpObjectTargetOpt.INSTANCE;
+      }
+    }
+
+    @Override
+    public ObjectTargetOpt extractFromBlobId(BlobId id) {
+      return NoOpObjectTargetOpt.INSTANCE;
+    }
+    /** prevent java serialization from using a new instance */
+    private Object readResolve() {
+      return INSTANCE;
+    }
+  }
+
+  /**
+   * Attempt to extract a metageneration value from a Bucket or Object. If no metageneration value
+   * is extracted an {@link IllegalArgumentException} will be thrown.
+   *
+   * @see MetagenerationMatch
+   * @deprecated Use {@link #metagenerationMatch(long)}
+   */
+  @Deprecated
+  static final class MetagenerationMatchExtractor
+      implements ObjectOptExtractor<ObjectTargetOpt>, BucketOptExtractor<MetagenerationMatch> {
+    private static final MetagenerationMatchExtractor INSTANCE = new MetagenerationMatchExtractor();
+    private static final long serialVersionUID = -9012665484224118046L;
+
+    @Deprecated
+    private MetagenerationMatchExtractor() {}
+
+    @Override
+    public MetagenerationMatch extractFromBlobInfo(BlobInfo info) {
+      Long metageneration = info.getMetageneration();
+      checkArgument(metageneration != null, "Option ifMetagenerationMatch is missing a value");
+      return metagenerationMatch(metageneration);
+    }
+
+    @Override
+    public ObjectTargetOpt extractFromBlobId(BlobId id) {
+      return NoOpObjectTargetOpt.INSTANCE;
+    }
+
+    @Override
+    public MetagenerationMatch extractFromBucketInfo(BucketInfo info) {
+      Long metageneration = info.getMetageneration();
+      checkArgument(metageneration != null, "Option ifMetagenerationMatch is missing a value");
+      return metagenerationMatch(metageneration);
+    }
+
+    // Both parent interfaces define this method, we need to declare a dis-ambiguous one
+    @Override
+    public Mapper<GrpcCallContext> getGrpcMetadataMapper() {
+      return Mapper.identity();
+    }
+    /** prevent java serialization from using a new instance */
+    private Object readResolve() {
+      return INSTANCE;
+    }
+  }
+
+  /**
+   * Attempt to extract a metageneration value from a Bucket or Object. If no metageneration value
+   * is extracted an {@link IllegalArgumentException} will be thrown.
+   *
+   * @see MetagenerationNotMatch
+   * @deprecated Use {@link #metagenerationNotMatch(long)}
+   */
+  @Deprecated
+  static final class MetagenerationNotMatchExtractor
+      implements ObjectOptExtractor<ObjectTargetOpt>, BucketOptExtractor<MetagenerationNotMatch> {
+    private static final MetagenerationNotMatchExtractor INSTANCE =
+        new MetagenerationNotMatchExtractor();
+    private static final long serialVersionUID = -732561730735045523L;
+
+    @Deprecated
+    private MetagenerationNotMatchExtractor() {}
+
+    @Override
+    public MetagenerationNotMatch extractFromBlobInfo(BlobInfo info) {
+      Long metageneration = info.getMetageneration();
+      checkArgument(metageneration != null, "Option ifMetagenerationNotMatch is missing a value");
+      return metagenerationNotMatch(metageneration);
+    }
+
+    @Override
+    public ObjectTargetOpt extractFromBlobId(BlobId id) {
+      return NoOpObjectTargetOpt.INSTANCE;
+    }
+
+    @Override
+    public MetagenerationNotMatch extractFromBucketInfo(BucketInfo info) {
+      Long metageneration = info.getMetageneration();
+      checkArgument(metageneration != null, "Option ifMetagenerationNotMatch is missing a value");
+      return metagenerationNotMatch(metageneration);
+    }
+    // Both parent interfaces define this method, we need to declare a dis-ambiguous one
+    @Override
+    public Mapper<GrpcCallContext> getGrpcMetadataMapper() {
+      return Mapper.identity();
+    }
+
+    /** prevent java serialization from using a new instance */
+    private Object readResolve() {
+      return INSTANCE;
+    }
+  }
+
+  /**
+   * Internal only implementation of {@link ObjectTargetOpt} which is a No-op.
+   *
+   * <p>The instance of this class can be returned when a no-op is necessary.
+   */
+  @VisibleForTesting
+  static final class NoOpObjectTargetOpt implements ObjectTargetOpt {
+    @VisibleForTesting static final NoOpObjectTargetOpt INSTANCE = new NoOpObjectTargetOpt();
+    private static final long serialVersionUID = -3702724179751638748L;
+
+    private NoOpObjectTargetOpt() {}
+
+    @Override
+    public Mapper<GrpcCallContext> getGrpcMetadataMapper() {
+      return Mapper.identity();
+    }
+
+    @Override
+    public Mapper<BlobInfo.Builder> blobInfo() {
+      return Mapper.identity();
+    }
+
+    /** prevent java serialization from using a new instance */
+    private Object readResolve() {
+      return INSTANCE;
+    }
+  }
+
+  /**
+   * A shim class used by {@link Option} to allow a common parent which isn't part of the public
+   * api.
+   *
+   * <p>{@link Option} itself and all it's subclasses are now obsolete, and should be removed when
+   * we're able to remove them from the public api.
+   */
+  @Deprecated
+  abstract static class OptionShim<O extends Opt> implements Serializable {
+    private static final long serialVersionUID = -1026813326366179926L;
+    private final O opt;
+
+    OptionShim(O opt) {
+      this.opt = opt;
+    }
+
+    O getOpt() {
+      return opt;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) {
+        return true;
+      }
+      if (!(o instanceof OptionShim)) {
+        return false;
+      }
+      OptionShim<?> that = (OptionShim<?>) o;
+      return Objects.equals(opt, that.opt);
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(opt);
+    }
+
+    @Override
+    public String toString() {
+      return this.getClass().getSimpleName() + "{opt=" + opt + '}';
+    }
+  }
+
+  /**
+   * Base class for those {@link Opt}s which correspond to one or more {@link StorageRpc.Option}
+   * keys.
+   *
+   * @param <T>
+   */
+  private abstract static class RpcOptVal<T> implements Opt {
+    private static final long serialVersionUID = -86698141922923191L;
+    protected final StorageRpc.Option key;
+    protected final T val;
+
+    private RpcOptVal(StorageRpc.Option key, T val) {
+      this.key = key;
+      this.val = val;
+    }
+
+    public Mapper<ImmutableMap.Builder<StorageRpc.Option, Object>> mapper() {
+      return b -> b.put(key, val);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) {
+        return true;
+      }
+      if (!(o instanceof RpcOptVal)) {
+        return false;
+      }
+      RpcOptVal<?> rpcOptVal = (RpcOptVal<?>) o;
+      return Objects.equals(key, rpcOptVal.key) && Objects.equals(val, rpcOptVal.val);
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(key, val);
+    }
+
+    @Override
+    public String toString() {
+      return this.getClass().getSimpleName() + "{key=" + key + ", val=" + val + '}';
+    }
+  }
+
+  /**
+   * Internal "collection" class to represent a set of {@link Opt}s, and to provide useful
+   * transformations to individual mappers or to resolve any extractors providing a new instance
+   * without extractors.
+   */
+  @SuppressWarnings("unchecked")
+  static final class Opts<T extends Opt> {
+
+    private final ImmutableList<T> opts;
+
+    private Opts(ImmutableList<T> opts) {
+      this.opts = opts;
+    }
+
+    /**
+     * Resolve any extractors relative to the provided {@link BlobInfo} and return a new instance.
+     */
+    Opts<T> resolveFrom(BlobInfo info) {
+      ImmutableList<T> resolvedOpts =
+          opts.stream()
+              .map(
+                  o -> {
+                    if (o instanceof ObjectOptExtractor) {
+                      ObjectOptExtractor<T> ex = (ObjectOptExtractor<T>) o;
+                      return ex.extractFromBlobInfo(info);
+                    } else {
+                      return o;
+                    }
+                  })
+              .collect(ImmutableList.toImmutableList());
+      return new Opts<>(resolvedOpts);
+    }
+
+    /** Resolve any extractors relative to the provided {@link BlobId} and return a new instance. */
+    Opts<T> resolveFrom(BlobId id) {
+      ImmutableList<T> resolvedOpts =
+          opts.stream()
+              .map(
+                  o -> {
+                    if (o instanceof ObjectOptExtractor) {
+                      ObjectOptExtractor<T> ex = (ObjectOptExtractor<T>) o;
+                      return ex.extractFromBlobId(id);
+                    } else {
+                      return o;
+                    }
+                  })
+              .collect(ImmutableList.toImmutableList());
+      return new Opts<>(resolvedOpts);
+    }
+
+    /**
+     * Resolve any extractors relative to the provided {@link BucketInfo} and return a new instance.
+     */
+    Opts<T> resolveFrom(BucketInfo info) {
+      ImmutableList<T> resolvedOpts =
+          opts.stream()
+              .map(
+                  o -> {
+                    if (o instanceof BucketOptExtractor) {
+                      BucketOptExtractor<T> ex = (BucketOptExtractor<T>) o;
+                      return ex.extractFromBucketInfo(info);
+                    } else {
+                      return o;
+                    }
+                  })
+              .collect(ImmutableList.toImmutableList());
+      return new Opts<>(resolvedOpts);
+    }
+
+    Opts<T> projectAsSource() {
+      ImmutableList<T> projectedOpts =
+          opts.stream()
+              .map(
+                  o -> {
+                    if (o instanceof ProjectAsSource) {
+                      ProjectAsSource<T> p = (ProjectAsSource<T>) o;
+                      return p.asSource();
+                    } else {
+                      return o;
+                    }
+                  })
+              .collect(ImmutableList.toImmutableList());
+      return new Opts<>(projectedOpts);
+    }
+
+    /**
+     * Attempt to construct a {@link StorageRpc} compatible map of {@link StorageRpc.Option}.
+     *
+     * <p>Validation ensures an absence of duplicate keys, and mutually exclusive keys.
+     */
+    ImmutableMap<StorageRpc.Option, ?> getRpcOptions() {
+      ImmutableMap.Builder<StorageRpc.Option, Object> builder =
+          rpcOptionMapper().apply(ImmutableMap.builder());
+      return builder.buildOrThrow();
+    }
+
+    Mapper<GrpcCallContext> grpcMetadataMapper() {
+      return fuseMappers(GrpcMetadataMapper.class, GrpcMetadataMapper::getGrpcMetadataMapper);
+    }
+
+    Mapper<GetObjectRequest.Builder> getObjectRequest() {
+      return fuseMappers(ObjectSourceOpt.class, ObjectSourceOpt::getObject);
+    }
+
+    Mapper<BlobInfo.Builder> blobInfoMapper() {
+      return fuseMappers(ObjectTargetOpt.class, ObjectTargetOpt::blobInfo);
+    }
+
+    private Mapper<ImmutableMap.Builder<StorageRpc.Option, Object>> rpcOptionMapper() {
+      return fuseMappers(RpcOptVal.class, RpcOptVal::mapper);
+    }
+
+    private <R, O> Mapper<O> fuseMappers(Class<R> c, Function<R, Mapper<O>> f) {
+      return filterTo(c).map(f).reduce(Mapper.identity(), Mapper::andThen);
+    }
+
+    @SuppressWarnings("unchecked")
+    private <R> Stream<R> filterTo(Class<R> c) {
+      // TODO: figure out if there is need for an "isApplicableTo" predicate
+      return opts.stream().filter(isInstanceOf(c)).map(x -> (R) x);
+    }
+
+    static <T extends Opt> Opts<T> from(T... ts) {
+      return new Opts<>(ImmutableList.copyOf(ts));
+    }
+
+    /**
+     * Given an array of OptionShim, extract the opt from each of them to construct a new instance
+     * of {@link Opts}
+     */
+    static <O extends Opt, T extends OptionShim<O>> Opts<O> unwrap(T[] ts) {
+      ImmutableList<O> collect =
+          Arrays.stream(ts).map(OptionShim::getOpt).collect(ImmutableList.toImmutableList());
+      return new Opts<>(collect);
+    }
+
+    /**
+     * Given a collection of OptionShim, extract the opt from each of them to construct a new
+     * instance of {@link Opts}
+     */
+    static <O extends Opt, T extends OptionShim<O>> Opts<O> unwrap(Collection<T> ts) {
+      ImmutableList<O> collect =
+          ts.stream().map(OptionShim::getOpt).collect(ImmutableList.toImmutableList());
+      return new Opts<>(collect);
+    }
+
+    /** Create a predicate which is able to effectively perform an {@code instanceof} check */
+    private static <T> Predicate<T> isInstanceOf(Class<?> c) {
+      return t -> c.isAssignableFrom(t.getClass());
+    }
+  }
+}


Test patch:
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/BucketTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/BucketTest.java
index 76b09ccd..99e7bb17 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/BucketTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/BucketTest.java
@@ -25,7 +25,6 @@ import static org.easymock.EasyMock.replay;
 import static org.easymock.EasyMock.verify;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
@@ -40,6 +39,8 @@ import com.google.cloud.storage.BucketInfo.DeleteRule;
 import com.google.cloud.storage.BucketInfo.LifecycleRule;
 import com.google.cloud.storage.BucketInfo.LifecycleRule.LifecycleAction;
 import com.google.cloud.storage.BucketInfo.LifecycleRule.LifecycleCondition;
+import com.google.cloud.storage.Storage.BlobTargetOption;
+import com.google.cloud.storage.Storage.BlobWriteOption;
 import com.google.common.base.Function;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
@@ -54,7 +55,6 @@ import java.util.List;
 import java.util.Map;
 import javax.crypto.spec.SecretKeySpec;
 import org.junit.After;
-import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 
@@ -367,11 +367,7 @@ public class BucketTest {
   @Test
   public void testCreateWithOptions() throws Exception {
     initializeExpectedBucket();
-    BlobInfo info =
-        BlobInfo.newBuilder(BlobId.of("b", "n", 42L))
-            .setContentType(CONTENT_TYPE)
-            .setMetageneration(24L)
-            .build();
+    BlobInfo info = BlobInfo.newBuilder(BlobId.of("b", "n")).setContentType(CONTENT_TYPE).build();
     Blob expectedBlob = new Blob(serviceMockReturnsOptions, new BlobInfo.BuilderImpl(info));
     byte[] content = {0xD, 0xE, 0xA, 0xD};
     Storage.PredefinedAcl acl = Storage.PredefinedAcl.ALL_AUTHENTICATED_USERS;
@@ -380,8 +376,8 @@ public class BucketTest {
             storage.create(
                 info,
                 content,
-                Storage.BlobTargetOption.generationMatch(),
-                Storage.BlobTargetOption.metagenerationMatch(),
+                new BlobTargetOption(UnifiedOpts.generationMatch(42L)),
+                new BlobTargetOption(UnifiedOpts.metagenerationMatch(24L)),
                 Storage.BlobTargetOption.predefinedAcl(acl),
                 Storage.BlobTargetOption.encryptionKey(BASE64_KEY),
                 Storage.BlobTargetOption.userProject(USER_PROJECT)))
@@ -437,12 +433,11 @@ public class BucketTest {
   @Test
   public void testCreateNotExists() throws Exception {
     initializeExpectedBucket();
-    BlobInfo info =
-        BlobInfo.newBuilder(BlobId.of("b", "n", 0L)).setContentType(CONTENT_TYPE).build();
+    BlobInfo info = BlobInfo.newBuilder(BlobId.of("b", "n")).setContentType(CONTENT_TYPE).build();
     Blob expectedBlob = new Blob(serviceMockReturnsOptions, new BlobInfo.BuilderImpl(info));
     byte[] content = {0xD, 0xE, 0xA, 0xD};
     expect(storage.getOptions()).andReturn(mockOptions);
-    expect(storage.create(info, content, Storage.BlobTargetOption.generationMatch()))
+    expect(storage.create(info, content, new BlobTargetOption(UnifiedOpts.doesNotExist())))
         .andReturn(expectedBlob);
     replay(storage);
     initializeBucket();
@@ -450,46 +445,6 @@ public class BucketTest {
     assertEquals(expectedBlob, blob);
   }
 
-  @Test
-  public void testCreateWithWrongGenerationOptions() throws Exception {
-    initializeExpectedBucket();
-    expect(storage.getOptions()).andReturn(mockOptions);
-    replay(storage);
-    initializeBucket();
-    byte[] content = {0xD, 0xE, 0xA, 0xD};
-    try {
-      bucket.create(
-          "n",
-          content,
-          CONTENT_TYPE,
-          Bucket.BlobTargetOption.generationMatch(42L),
-          Bucket.BlobTargetOption.generationNotMatch(24L));
-      Assert.fail();
-    } catch (IllegalArgumentException ex) {
-      assertNotNull(ex.getMessage());
-    }
-  }
-
-  @Test
-  public void testCreateWithWrongMetagenerationOptions() throws Exception {
-    initializeExpectedBucket();
-    expect(storage.getOptions()).andReturn(mockOptions);
-    replay(storage);
-    initializeBucket();
-    byte[] content = {0xD, 0xE, 0xA, 0xD};
-    try {
-      bucket.create(
-          "n",
-          content,
-          CONTENT_TYPE,
-          Bucket.BlobTargetOption.metagenerationMatch(42L),
-          Bucket.BlobTargetOption.metagenerationNotMatch(24L));
-      Assert.fail();
-    } catch (IllegalArgumentException ex) {
-      assertNotNull(ex.getMessage());
-    }
-  }
-
   @Test
   @SuppressWarnings({"unchecked", "deprecation"})
   public void testCreateFromStream() throws Exception {
@@ -526,14 +481,8 @@ public class BucketTest {
   @SuppressWarnings({"unchecked", "deprecation"})
   public void testCreateFromStreamWithOptions() throws Exception {
     initializeExpectedBucket();
-    BlobInfo info =
-        BlobInfo.newBuilder(BlobId.of("b", "n", 42L))
-            .setContentType(CONTENT_TYPE)
-            .setMetageneration(24L)
-            .setCrc32c("crc")
-            .setMd5("md5")
-            .build();
-    Blob expectedBlob = new Blob(serviceMockReturnsOptions, new BlobInfo.BuilderImpl(info));
+    BlobInfo info = BlobInfo.newBuilder(BlobId.of("b", "n")).setContentType(CONTENT_TYPE).build();
+    Blob expectedBlob = info.asBlob(serviceMockReturnsOptions);
     byte[] content = {0xD, 0xE, 0xA, 0xD};
     Storage.PredefinedAcl acl = Storage.PredefinedAcl.ALL_AUTHENTICATED_USERS;
     InputStream streamContent = new ByteArrayInputStream(content);
@@ -542,11 +491,11 @@ public class BucketTest {
             storage.create(
                 info,
                 streamContent,
-                Storage.BlobWriteOption.generationMatch(),
-                Storage.BlobWriteOption.metagenerationMatch(),
+                new BlobWriteOption(UnifiedOpts.generationMatch(42L)),
+                new BlobWriteOption(UnifiedOpts.metagenerationMatch(24L)),
                 Storage.BlobWriteOption.predefinedAcl(acl),
-                Storage.BlobWriteOption.crc32cMatch(),
-                Storage.BlobWriteOption.md5Match(),
+                new BlobWriteOption(UnifiedOpts.crc32cMatch("crc")),
+                new BlobWriteOption(UnifiedOpts.md5Match("md5")),
                 Storage.BlobWriteOption.encryptionKey(BASE64_KEY),
                 Storage.BlobWriteOption.userProject(USER_PROJECT)))
         .andReturn(expectedBlob);
@@ -589,13 +538,12 @@ public class BucketTest {
   @SuppressWarnings({"unchecked", "deprecation"})
   public void testCreateFromStreamNotExists() throws Exception {
     initializeExpectedBucket();
-    BlobInfo info =
-        BlobInfo.newBuilder(BlobId.of("b", "n", 0L)).setContentType(CONTENT_TYPE).build();
-    Blob expectedBlob = new Blob(serviceMockReturnsOptions, new BlobInfo.BuilderImpl(info));
+    BlobInfo info = BlobInfo.newBuilder(BlobId.of("b", "n")).setContentType(CONTENT_TYPE).build();
+    Blob expectedBlob = info.asBlob(serviceMockReturnsOptions);
     byte[] content = {0xD, 0xE, 0xA, 0xD};
     InputStream streamContent = new ByteArrayInputStream(content);
     expect(storage.getOptions()).andReturn(mockOptions);
-    expect(storage.create(info, streamContent, Storage.BlobWriteOption.generationMatch()))
+    expect(storage.create(info, streamContent, new BlobWriteOption(UnifiedOpts.doesNotExist())))
         .andReturn(expectedBlob);
     replay(storage);
     initializeBucket();
@@ -604,48 +552,6 @@ public class BucketTest {
     assertEquals(expectedBlob, blob);
   }
 
-  @Test
-  public void testCreateFromStreamWithWrongGenerationOptions() throws Exception {
-    initializeExpectedBucket();
-    expect(storage.getOptions()).andReturn(mockOptions);
-    replay(storage);
-    initializeBucket();
-    byte[] content = {0xD, 0xE, 0xA, 0xD};
-    InputStream streamContent = new ByteArrayInputStream(content);
-    try {
-      bucket.create(
-          "n",
-          streamContent,
-          CONTENT_TYPE,
-          Bucket.BlobWriteOption.generationMatch(42L),
-          Bucket.BlobWriteOption.generationNotMatch(24L));
-      Assert.fail();
-    } catch (IllegalArgumentException ex) {
-      assertNotNull(ex.getMessage());
-    }
-  }
-
-  @Test
-  public void testCreateFromStreamWithWrongMetagenerationOptions() throws Exception {
-    initializeExpectedBucket();
-    expect(storage.getOptions()).andReturn(mockOptions);
-    replay(storage);
-    initializeBucket();
-    byte[] content = {0xD, 0xE, 0xA, 0xD};
-    InputStream streamContent = new ByteArrayInputStream(content);
-    try {
-      bucket.create(
-          "n",
-          streamContent,
-          CONTENT_TYPE,
-          Bucket.BlobWriteOption.metagenerationMatch(42L),
-          Bucket.BlobWriteOption.metagenerationNotMatch(24L));
-      Assert.fail();
-    } catch (IllegalArgumentException ex) {
-      assertNotNull(ex.getMessage());
-    }
-  }
-
   @Test
   public void testGetAcl() throws Exception {
     initializeExpectedBucket();
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ObjectOptExtractorTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ObjectOptExtractorTest.java
new file mode 100644
index 00000000..06ada97b
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ObjectOptExtractorTest.java
@@ -0,0 +1,204 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage;
+
+import static com.google.common.truth.Truth.assertThat;
+import static org.junit.Assert.assertThrows;
+
+import com.google.cloud.storage.UnifiedOpts.Crc32cMatchExtractor;
+import com.google.cloud.storage.UnifiedOpts.GenerationMatch;
+import com.google.cloud.storage.UnifiedOpts.GenerationMatchExtractor;
+import com.google.cloud.storage.UnifiedOpts.GenerationNotMatch;
+import com.google.cloud.storage.UnifiedOpts.GenerationNotMatchExtractor;
+import com.google.cloud.storage.UnifiedOpts.Md5MatchExtractor;
+import com.google.cloud.storage.UnifiedOpts.MetagenerationMatchExtractor;
+import com.google.cloud.storage.UnifiedOpts.MetagenerationNotMatchExtractor;
+import com.google.cloud.storage.UnifiedOpts.NoOpObjectTargetOpt;
+import org.junit.Test;
+import org.junit.function.ThrowingRunnable;
+
+public final class ObjectOptExtractorTest {
+
+  @Test
+  public void generationMatch_success() {
+    GenerationMatchExtractor ex = UnifiedOpts.generationMatchExtractor();
+    BlobId id = BlobId.of("b", "o", 7654L);
+    BlobInfo info = BlobInfo.newBuilder(id).build();
+
+    GenerationMatch expected = UnifiedOpts.generationMatch(7654);
+    assertThat(ex.extractFromBlobId(id)).isEqualTo(expected);
+    assertThat(ex.extractFromBlobInfo(info)).isEqualTo(expected);
+  }
+
+  @Test
+  public void generationMatch_errorOnNull() {
+    GenerationMatchExtractor ex = UnifiedOpts.generationMatchExtractor();
+    BlobId id = BlobId.of("b", "o");
+    BlobInfo info = BlobInfo.newBuilder(id).build();
+
+    assertExceptions(
+        () -> ex.extractFromBlobId(id), () -> ex.extractFromBlobInfo(info), "ifGenerationMatch");
+  }
+
+  @Test
+  public void generationNotMatch_success() {
+    GenerationNotMatchExtractor ex = UnifiedOpts.generationNotMatchExtractor();
+    BlobId id = BlobId.of("b", "o", 7654L);
+    BlobInfo info = BlobInfo.newBuilder(id).build();
+
+    GenerationNotMatch expected = UnifiedOpts.generationNotMatch(7654);
+    assertThat(ex.extractFromBlobId(id)).isEqualTo(expected);
+    assertThat(ex.extractFromBlobInfo(info)).isEqualTo(expected);
+  }
+
+  @Test
+  public void generationNotMatch_errorOnNull() {
+    GenerationNotMatchExtractor ex = UnifiedOpts.generationNotMatchExtractor();
+    BlobId id = BlobId.of("b", "o");
+    BlobInfo info = BlobInfo.newBuilder(id).build();
+
+    assertExceptions(
+        () -> ex.extractFromBlobId(id), () -> ex.extractFromBlobInfo(info), "ifGenerationNotMatch");
+  }
+
+  @Test
+  public void metagenerationMatch_success_blob() {
+    MetagenerationMatchExtractor ex = UnifiedOpts.metagenerationMatchExtractor();
+    BlobId id = BlobId.of("b", "o");
+    BlobInfo info = BlobInfo.newBuilder(id).setMetageneration(3L).build();
+
+    assertThat(ex.extractFromBlobId(id)).isEqualTo(NoOpObjectTargetOpt.INSTANCE);
+    assertThat(ex.extractFromBlobInfo(info)).isEqualTo(UnifiedOpts.metagenerationMatch(3));
+  }
+
+  @Test
+  public void metagenerationMatch_errorOnNull_blob() {
+    MetagenerationMatchExtractor ex = UnifiedOpts.metagenerationMatchExtractor();
+    BlobId id = BlobId.of("b", "o");
+    BlobInfo info = BlobInfo.newBuilder(id).build();
+
+    assertThat(ex.extractFromBlobId(id)).isEqualTo(NoOpObjectTargetOpt.INSTANCE);
+    IllegalArgumentException e1 =
+        assertThrows(IllegalArgumentException.class, () -> ex.extractFromBlobInfo(info));
+    assertThat(e1).hasMessageThat().contains("ifMetagenerationMatch");
+  }
+
+  @Test
+  public void metagenerationNotMatch_success_blob() {
+    MetagenerationNotMatchExtractor ex = UnifiedOpts.metagenerationNotMatchExtractor();
+    BlobId id = BlobId.of("b", "o");
+    BlobInfo info = BlobInfo.newBuilder(id).setMetageneration(3L).build();
+
+    assertThat(ex.extractFromBlobId(id)).isEqualTo(NoOpObjectTargetOpt.INSTANCE);
+    assertThat(ex.extractFromBlobInfo(info)).isEqualTo(UnifiedOpts.metagenerationNotMatch(3));
+  }
+
+  @Test
+  public void metagenerationNotMatch_errorOnNull_blob() {
+    MetagenerationNotMatchExtractor ex = UnifiedOpts.metagenerationNotMatchExtractor();
+    BlobId id = BlobId.of("b", "o");
+    BlobInfo info = BlobInfo.newBuilder(id).build();
+
+    assertThat(ex.extractFromBlobId(id)).isEqualTo(NoOpObjectTargetOpt.INSTANCE);
+    IllegalArgumentException e1 =
+        assertThrows(IllegalArgumentException.class, () -> ex.extractFromBlobInfo(info));
+    assertThat(e1).hasMessageThat().contains("ifMetagenerationNotMatch");
+  }
+
+  @Test
+  public void metagenerationMatch_success_bucket() {
+    MetagenerationMatchExtractor ex = UnifiedOpts.metagenerationMatchExtractor();
+    BucketInfo info = BucketInfo.newBuilder("b").setMetageneration(3L).build();
+
+    assertThat(ex.extractFromBucketInfo(info)).isEqualTo(UnifiedOpts.metagenerationMatch(3));
+  }
+
+  @Test
+  public void metagenerationMatch_errorOnNull_bucket() {
+    MetagenerationMatchExtractor ex = UnifiedOpts.metagenerationMatchExtractor();
+    BucketInfo info = BucketInfo.newBuilder("b").build();
+
+    IllegalArgumentException e1 =
+        assertThrows(IllegalArgumentException.class, () -> ex.extractFromBucketInfo(info));
+    assertThat(e1).hasMessageThat().contains("ifMetagenerationMatch");
+  }
+
+  @Test
+  public void metagenerationNotMatch_success_bucket() {
+    MetagenerationNotMatchExtractor ex = UnifiedOpts.metagenerationNotMatchExtractor();
+    BucketInfo info = BucketInfo.newBuilder("b").setMetageneration(3L).build();
+
+    assertThat(ex.extractFromBucketInfo(info)).isEqualTo(UnifiedOpts.metagenerationNotMatch(3));
+  }
+
+  @Test
+  public void metagenerationNotMatch_errorOnNull_bucket() {
+    MetagenerationNotMatchExtractor ex = UnifiedOpts.metagenerationNotMatchExtractor();
+    BucketInfo info = BucketInfo.newBuilder("b").build();
+
+    IllegalArgumentException e1 =
+        assertThrows(IllegalArgumentException.class, () -> ex.extractFromBucketInfo(info));
+    assertThat(e1).hasMessageThat().contains("ifMetagenerationNotMatch");
+  }
+
+  @Test
+  public void md5Match_success() {
+    Md5MatchExtractor ex = UnifiedOpts.md5MatchExtractor();
+    BlobId id = BlobId.of("b", "o");
+    BlobInfo info = BlobInfo.newBuilder(id).setMd5("md5").build();
+
+    assertThat(ex.extractFromBlobId(id)).isEqualTo(NoOpObjectTargetOpt.INSTANCE);
+    assertThat(ex.extractFromBlobInfo(info)).isEqualTo(UnifiedOpts.md5Match("md5"));
+  }
+
+  @Test
+  public void md5Match_noOpOnNull() {
+    Md5MatchExtractor ex = UnifiedOpts.md5MatchExtractor();
+    BlobId id = BlobId.of("b", "o");
+    BlobInfo info = BlobInfo.newBuilder(id).build();
+
+    assertThat(ex.extractFromBlobId(id)).isEqualTo(NoOpObjectTargetOpt.INSTANCE);
+    assertThat(ex.extractFromBlobInfo(info)).isEqualTo(NoOpObjectTargetOpt.INSTANCE);
+  }
+
+  @Test
+  public void crc32cMatch_success() {
+    Crc32cMatchExtractor ex = UnifiedOpts.crc32cMatchExtractor();
+    BlobId id = BlobId.of("b", "o");
+    BlobInfo info = BlobInfo.newBuilder(id).setCrc32c("crc32c").build();
+
+    assertThat(ex.extractFromBlobId(id)).isEqualTo(NoOpObjectTargetOpt.INSTANCE);
+    assertThat(ex.extractFromBlobInfo(info)).isEqualTo(UnifiedOpts.crc32cMatch("crc32c"));
+  }
+
+  @Test
+  public void crc32cMatch_noOpOnNull() {
+    Crc32cMatchExtractor ex = UnifiedOpts.crc32cMatchExtractor();
+    BlobId id = BlobId.of("b", "o");
+    BlobInfo info = BlobInfo.newBuilder(id).build();
+
+    assertThat(ex.extractFromBlobId(id)).isEqualTo(NoOpObjectTargetOpt.INSTANCE);
+    assertThat(ex.extractFromBlobInfo(info)).isEqualTo(NoOpObjectTargetOpt.INSTANCE);
+  }
+
+  private static void assertExceptions(ThrowingRunnable r1, ThrowingRunnable r2, String contains) {
+    IllegalArgumentException e1 = assertThrows(IllegalArgumentException.class, r1);
+    assertThat(e1).hasMessageThat().contains(contains);
+    IllegalArgumentException e2 = assertThrows(IllegalArgumentException.class, r2);
+    assertThat(e2).hasMessageThat().contains(contains);
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/OptionTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/OptionTest.java
deleted file mode 100644
index 549c7e97..00000000
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/OptionTest.java
+++ /dev/null
@@ -1,63 +0,0 @@
-/*
- * Copyright 2015 Google LLC
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *       http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.google.cloud.storage;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotEquals;
-import static org.junit.Assert.assertNull;
-
-import com.google.cloud.storage.spi.v1.StorageRpc;
-import org.junit.Assert;
-import org.junit.Test;
-
-public class OptionTest {
-
-  private static final StorageRpc.Option RPC_OPTION = StorageRpc.Option.DELIMITER;
-  private static final StorageRpc.Option ANOTHER_RPC_OPTION = StorageRpc.Option.FIELDS;
-  private static final String VALUE = "some value";
-  private static final String OTHER_VALUE = "another value";
-  private static final Option OPTION = new Option(RPC_OPTION, VALUE) {};
-  private static final Option OPTION_EQUALS = new Option(RPC_OPTION, VALUE) {};
-  private static final Option OPTION_NOT_EQUALS1 = new Option(RPC_OPTION, OTHER_VALUE) {};
-  private static final Option OPTION_NOT_EQUALS2 = new Option(ANOTHER_RPC_OPTION, VALUE) {};
-
-  @Test
-  public void testEquals() {
-    assertEquals(OPTION, OPTION_EQUALS);
-    assertNotEquals(OPTION, OPTION_NOT_EQUALS1);
-    assertNotEquals(OPTION, OPTION_NOT_EQUALS2);
-  }
-
-  @Test
-  public void testHashCode() {
-    assertEquals(OPTION.hashCode(), OPTION_EQUALS.hashCode());
-  }
-
-  @Test
-  public void testConstructor() {
-    assertEquals(RPC_OPTION, OPTION.getRpcOption());
-    assertEquals(VALUE, OPTION.getValue());
-    Option option = new Option(RPC_OPTION, null) {};
-    assertEquals(RPC_OPTION, option.getRpcOption());
-    assertNull(option.getValue());
-    try {
-      new Option(null, VALUE) {};
-      Assert.fail();
-    } catch (NullPointerException expected) {
-    }
-  }
-}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/SerializationTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/SerializationTest.java
index 06354d23..6420bd8a 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/SerializationTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/SerializationTest.java
@@ -16,6 +16,9 @@
 
 package com.google.cloud.storage;
 
+import static com.google.common.base.MoreObjects.firstNonNull;
+import static org.junit.Assert.assertEquals;
+
 import com.google.api.gax.retrying.ResultRetryAlgorithm;
 import com.google.cloud.BaseSerializationTest;
 import com.google.cloud.NoCredentials;
@@ -28,6 +31,7 @@ import com.google.common.collect.ImmutableMap;
 import java.io.Serializable;
 import java.util.Collections;
 import java.util.Map;
+import org.junit.Test;
 
 public class SerializationTest extends BaseSerializationTest {
 
@@ -111,4 +115,23 @@ public class SerializationTest extends BaseSerializationTest {
             options, BlobInfo.newBuilder(BlobId.of("b", "n")).build(), "upload-id", algorithm);
     return new Restorable<?>[] {reader, writer};
   }
+
+  /**
+   * Here we override the super classes implementation to remove the "assertNotSame".
+   *
+   * <p>We should not enforce that two instances are not the same. As long as they're equal and have
+   * the same hashCode that should be sufficient.
+   */
+  @Test
+  @Override
+  public void testSerializableObjects() throws Exception {
+    for (Serializable obj : firstNonNull(serializableObjects(), new Serializable[0])) {
+      Object copy = serializeAndDeserialize(obj);
+      assertEquals(obj, obj);
+      assertEquals(obj, copy);
+      assertEquals(obj.hashCode(), copy.hashCode());
+      assertEquals(obj.toString(), copy.toString());
+      assertEquals(copy, copy);
+    }
+  }
 }
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageBatchTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageBatchTest.java
index 1ea2412f..33c273db 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageBatchTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageBatchTest.java
@@ -16,6 +16,7 @@
 
 package com.google.cloud.storage;
 
+import static com.google.common.truth.Truth.assertThat;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertSame;
@@ -50,9 +51,6 @@ public class StorageBatchTest {
   private static final BlobSourceOption[] BLOB_SOURCE_OPTIONS = {
     BlobSourceOption.generationMatch(42L), BlobSourceOption.metagenerationMatch(42L)
   };
-  private static final BlobTargetOption[] BLOB_TARGET_OPTIONS = {
-    BlobTargetOption.generationMatch(), BlobTargetOption.metagenerationMatch()
-  };
   private static final GoogleJsonError GOOGLE_JSON_ERROR = new GoogleJsonError();
   private final HttpRetryAlgorithmManager retryAlgorithmManager =
       HttpStorageOptions.getDefaultInstance().getRetryAlgorithmManager();
@@ -130,10 +128,10 @@ public class StorageBatchTest {
     EasyMock.replay(batchMock);
     StorageBatchResult<Boolean> batchResult = storageBatch.delete(BLOB_ID, BLOB_SOURCE_OPTIONS);
     assertNotNull(callback.getValue());
-    assertEquals(2, capturedOptions.getValue().size());
-    for (BlobSourceOption option : BLOB_SOURCE_OPTIONS) {
-      assertEquals(option.getValue(), capturedOptions.getValue().get(option.getRpcOption()));
-    }
+    Map<StorageRpc.Option, Object> options = capturedOptions.getValue();
+    assertEquals(2, options.size());
+    assertThat(options).containsEntry(StorageRpc.Option.IF_GENERATION_MATCH, 42L);
+    assertThat(options).containsEntry(StorageRpc.Option.IF_METAGENERATION_MATCH, 42L);
     RpcBatch.Callback<Void> capturedCallback = callback.getValue();
     capturedCallback.onSuccess(null);
     assertTrue(batchResult.get());
@@ -183,11 +181,14 @@ public class StorageBatchTest {
         EasyMock.capture(capturedOptions));
     EasyMock.replay(batchMock, storage, optionsMock);
     StorageBatchResult<Blob> batchResult =
-        storageBatch.update(BLOB_INFO_COMPLETE, BLOB_TARGET_OPTIONS);
+        storageBatch.update(
+            BLOB_INFO_COMPLETE,
+            BlobTargetOption.generationMatch(),
+            BlobTargetOption.metagenerationMatch());
     assertNotNull(callback.getValue());
     assertEquals(2, capturedOptions.getValue().size());
-    assertEquals(42L, capturedOptions.getValue().get(BLOB_TARGET_OPTIONS[0].getRpcOption()));
-    assertEquals(42L, capturedOptions.getValue().get(BLOB_TARGET_OPTIONS[1].getRpcOption()));
+    assertEquals(42L, capturedOptions.getValue().get(StorageRpc.Option.IF_GENERATION_MATCH));
+    assertEquals(42L, capturedOptions.getValue().get(StorageRpc.Option.IF_METAGENERATION_MATCH));
     RpcBatch.Callback<StorageObject> capturedCallback = callback.getValue();
     capturedCallback.onSuccess(Conversions.apiary().blobInfo().encode(BLOB_INFO));
     assertEquals(new Blob(storage, new Blob.BuilderImpl(BLOB_INFO)), batchResult.get());
@@ -238,10 +239,10 @@ public class StorageBatchTest {
     EasyMock.replay(storage, batchMock, optionsMock);
     StorageBatchResult<Blob> batchResult = storageBatch.get(BLOB_ID, BLOB_GET_OPTIONS);
     assertNotNull(callback.getValue());
-    assertEquals(2, capturedOptions.getValue().size());
-    for (BlobGetOption option : BLOB_GET_OPTIONS) {
-      assertEquals(option.getValue(), capturedOptions.getValue().get(option.getRpcOption()));
-    }
+    Map<StorageRpc.Option, Object> options = capturedOptions.getValue();
+    assertEquals(2, options.size());
+    assertThat(options).containsEntry(StorageRpc.Option.IF_GENERATION_MATCH, 42L);
+    assertThat(options).containsEntry(StorageRpc.Option.IF_METAGENERATION_MATCH, 42L);
     RpcBatch.Callback<StorageObject> capturedCallback = callback.getValue();
     capturedCallback.onSuccess(Conversions.apiary().blobInfo().encode(BLOB_INFO));
     assertEquals(new Blob(storage, new Blob.BuilderImpl(BLOB_INFO)), batchResult.get());
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageImplMockitoTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageImplMockitoTest.java
index 4465d0e0..e3ec65c5 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageImplMockitoTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageImplMockitoTest.java
@@ -30,8 +30,6 @@ import static org.mockito.Mockito.mock;
 import com.google.api.core.ApiClock;
 import com.google.api.gax.paging.Page;
 import com.google.api.services.storage.model.StorageObject;
-import com.google.cloud.Identity;
-import com.google.cloud.Policy;
 import com.google.cloud.ReadChannel;
 import com.google.cloud.ServiceOptions;
 import com.google.cloud.Tuple;
@@ -68,7 +66,6 @@ import org.junit.BeforeClass;
 import org.junit.Test;
 import org.mockito.ArgumentCaptor;
 import org.mockito.Mockito;
-import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
 
 public class StorageImplMockitoTest {
@@ -92,24 +89,24 @@ public class StorageImplMockitoTest {
   private static final String KMS_KEY_NAME =
       "projects/gcloud-devel/locations/us/keyRings/gcs_kms_key_ring_us/cryptoKeys/key";
   private static final Long RETENTION_PERIOD = 10L;
-  private static final String USER_PROJECT = "test-project";
   private static final int DEFAULT_BUFFER_SIZE = 15 * 1024 * 1024;
   private static final int MIN_BUFFER_SIZE = 256 * 1024;
+  protected static final long PAGE_SIZE = 42L;
   // BucketInfo objects
   private static final BucketInfo BUCKET_INFO1 =
-      BucketInfo.newBuilder(BUCKET_NAME1).setMetageneration(42L).build();
+      BucketInfo.newBuilder(BUCKET_NAME1).setMetageneration(PAGE_SIZE).build();
   private static final BucketInfo BUCKET_INFO2 = BucketInfo.newBuilder(BUCKET_NAME2).build();
   private static final BucketInfo BUCKET_INFO3 =
       BucketInfo.newBuilder(BUCKET_NAME3)
           .setRetentionPeriod(RETENTION_PERIOD)
           .setRetentionPolicyIsLocked(true)
-          .setMetageneration(42L)
+          .setMetageneration(PAGE_SIZE)
           .build();
 
   // BlobInfo objects
   private static final BlobInfo BLOB_INFO1 =
       BlobInfo.newBuilder(BUCKET_NAME1, BLOB_NAME1, 24L)
-          .setMetageneration(42L)
+          .setMetageneration(PAGE_SIZE)
           .setContentType("application/json")
           .setMd5("md5string")
           .build();
@@ -131,22 +128,11 @@ public class StorageImplMockitoTest {
       Storage.BucketTargetOption.metagenerationMatch();
   private static final Storage.BucketTargetOption BUCKET_TARGET_PREDEFINED_ACL =
       Storage.BucketTargetOption.predefinedAcl(Storage.PredefinedAcl.PRIVATE);
-  private static final Storage.BucketTargetOption BUCKET_TARGET_USER_PROJECT =
-      Storage.BucketTargetOption.userProject(USER_PROJECT);
   private static final Map<StorageRpc.Option, ?> BUCKET_TARGET_OPTIONS =
       ImmutableMap.of(
           StorageRpc.Option.IF_METAGENERATION_MATCH, BUCKET_INFO1.getMetageneration(),
-          StorageRpc.Option.PREDEFINED_ACL, BUCKET_TARGET_PREDEFINED_ACL.getValue());
-  private static final Map<StorageRpc.Option, ?> BUCKET_TARGET_OPTIONS_LOCK_RETENTION_POLICY =
-      ImmutableMap.of(
-          StorageRpc.Option.IF_METAGENERATION_MATCH,
-          BUCKET_INFO3.getMetageneration(),
-          StorageRpc.Option.USER_PROJECT,
-          USER_PROJECT);
-
-  // Blob target options (create, update, compose)
-  private static final Storage.BlobTargetOption BLOB_TARGET_GENERATION =
-      Storage.BlobTargetOption.generationMatch();
+          StorageRpc.Option.PREDEFINED_ACL, Storage.PredefinedAcl.PRIVATE.getEntry());
+
   private static final Storage.BlobTargetOption BLOB_TARGET_METAGENERATION =
       Storage.BlobTargetOption.metagenerationMatch();
   private static final Storage.BlobTargetOption BLOB_TARGET_DISABLE_GZIP_CONTENT =
@@ -159,17 +145,9 @@ public class StorageImplMockitoTest {
       ImmutableMap.of(
           StorageRpc.Option.IF_METAGENERATION_MATCH, BLOB_INFO1.getMetageneration(),
           StorageRpc.Option.IF_GENERATION_MATCH, 0L,
-          StorageRpc.Option.PREDEFINED_ACL, BUCKET_TARGET_PREDEFINED_ACL.getValue());
+          StorageRpc.Option.PREDEFINED_ACL, Storage.PredefinedAcl.PRIVATE.getEntry());
   private static final Map<StorageRpc.Option, ?> BLOB_TARGET_OPTIONS_CREATE_DISABLE_GZIP_CONTENT =
       ImmutableMap.of(StorageRpc.Option.IF_DISABLE_GZIP_CONTENT, true);
-  private static final Map<StorageRpc.Option, ?> BLOB_TARGET_OPTIONS_UPDATE =
-      ImmutableMap.of(
-          StorageRpc.Option.IF_METAGENERATION_MATCH, BLOB_INFO1.getMetageneration(),
-          StorageRpc.Option.PREDEFINED_ACL, BUCKET_TARGET_PREDEFINED_ACL.getValue());
-  private static final Map<StorageRpc.Option, ?> BLOB_TARGET_OPTIONS_COMPOSE =
-      ImmutableMap.of(
-          StorageRpc.Option.IF_GENERATION_MATCH, BLOB_INFO1.getGeneration(),
-          StorageRpc.Option.IF_METAGENERATION_MATCH, BLOB_INFO1.getMetageneration());
 
   // Blob write options (create, writer)
   private static final Storage.BlobWriteOption BLOB_WRITE_METAGENERATION =
@@ -184,11 +162,6 @@ public class StorageImplMockitoTest {
       Storage.BlobWriteOption.crc32cMatch();
 
   // Bucket get/source options
-  private static final Storage.BucketSourceOption BUCKET_SOURCE_METAGENERATION =
-      Storage.BucketSourceOption.metagenerationMatch(BUCKET_INFO1.getMetageneration());
-  private static final Map<StorageRpc.Option, ?> BUCKET_SOURCE_OPTIONS =
-      ImmutableMap.of(
-          StorageRpc.Option.IF_METAGENERATION_MATCH, BUCKET_SOURCE_METAGENERATION.getValue());
   private static final Storage.BucketGetOption BUCKET_GET_METAGENERATION =
       Storage.BucketGetOption.metagenerationMatch(BUCKET_INFO1.getMetageneration());
   private static final Storage.BucketGetOption BUCKET_GET_FIELDS =
@@ -196,8 +169,7 @@ public class StorageImplMockitoTest {
   private static final Storage.BucketGetOption BUCKET_GET_EMPTY_FIELDS =
       Storage.BucketGetOption.fields();
   private static final Map<StorageRpc.Option, ?> BUCKET_GET_OPTIONS =
-      ImmutableMap.of(
-          StorageRpc.Option.IF_METAGENERATION_MATCH, BUCKET_SOURCE_METAGENERATION.getValue());
+      ImmutableMap.of(StorageRpc.Option.IF_METAGENERATION_MATCH, BUCKET_INFO1.getMetageneration());
 
   // Blob get/source options
   private static final Storage.BlobGetOption BLOB_GET_METAGENERATION =
@@ -211,8 +183,8 @@ public class StorageImplMockitoTest {
   private static final Storage.BlobGetOption BLOB_GET_EMPTY_FIELDS = Storage.BlobGetOption.fields();
   private static final Map<StorageRpc.Option, ?> BLOB_GET_OPTIONS =
       ImmutableMap.of(
-          StorageRpc.Option.IF_METAGENERATION_MATCH, BLOB_GET_METAGENERATION.getValue(),
-          StorageRpc.Option.IF_GENERATION_MATCH, BLOB_GET_GENERATION.getValue());
+          StorageRpc.Option.IF_METAGENERATION_MATCH, BLOB_INFO1.getMetageneration(),
+          StorageRpc.Option.IF_GENERATION_MATCH, BLOB_INFO1.getGeneration());
   private static final Storage.BlobSourceOption BLOB_SOURCE_METAGENERATION =
       Storage.BlobSourceOption.metagenerationMatch(BLOB_INFO1.getMetageneration());
   private static final Storage.BlobSourceOption BLOB_SOURCE_GENERATION =
@@ -221,16 +193,12 @@ public class StorageImplMockitoTest {
       Storage.BlobSourceOption.generationMatch();
   private static final Map<StorageRpc.Option, ?> BLOB_SOURCE_OPTIONS =
       ImmutableMap.of(
-          StorageRpc.Option.IF_METAGENERATION_MATCH, BLOB_SOURCE_METAGENERATION.getValue(),
-          StorageRpc.Option.IF_GENERATION_MATCH, BLOB_SOURCE_GENERATION.getValue());
-  private static final Map<StorageRpc.Option, ?> BLOB_SOURCE_OPTIONS_COPY =
-      ImmutableMap.of(
-          StorageRpc.Option.IF_SOURCE_METAGENERATION_MATCH, BLOB_SOURCE_METAGENERATION.getValue(),
-          StorageRpc.Option.IF_SOURCE_GENERATION_MATCH, BLOB_SOURCE_GENERATION.getValue());
+          StorageRpc.Option.IF_METAGENERATION_MATCH, BLOB_INFO1.getMetageneration(),
+          StorageRpc.Option.IF_GENERATION_MATCH, BLOB_INFO1.getGeneration());
 
   // Bucket list options
   private static final Storage.BucketListOption BUCKET_LIST_PAGE_SIZE =
-      Storage.BucketListOption.pageSize(42L);
+      Storage.BucketListOption.pageSize(PAGE_SIZE);
   private static final Storage.BucketListOption BUCKET_LIST_PREFIX =
       Storage.BucketListOption.prefix("prefix");
   private static final Storage.BucketListOption BUCKET_LIST_FIELDS =
@@ -238,13 +206,11 @@ public class StorageImplMockitoTest {
   private static final Storage.BucketListOption BUCKET_LIST_EMPTY_FIELDS =
       Storage.BucketListOption.fields();
   private static final Map<StorageRpc.Option, ?> BUCKET_LIST_OPTIONS =
-      ImmutableMap.of(
-          StorageRpc.Option.MAX_RESULTS, BUCKET_LIST_PAGE_SIZE.getValue(),
-          StorageRpc.Option.PREFIX, BUCKET_LIST_PREFIX.getValue());
+      ImmutableMap.of(StorageRpc.Option.MAX_RESULTS, PAGE_SIZE, StorageRpc.Option.PREFIX, "prefix");
 
   // Blob list options
   private static final Storage.BlobListOption BLOB_LIST_PAGE_SIZE =
-      Storage.BlobListOption.pageSize(42L);
+      Storage.BlobListOption.pageSize(PAGE_SIZE);
   private static final Storage.BlobListOption BLOB_LIST_PREFIX =
       Storage.BlobListOption.prefix("prefix");
   private static final Storage.BlobListOption BLOB_LIST_FIELDS =
@@ -255,14 +221,12 @@ public class StorageImplMockitoTest {
       Storage.BlobListOption.fields();
   private static final Map<StorageRpc.Option, ?> BLOB_LIST_OPTIONS =
       ImmutableMap.of(
-          StorageRpc.Option.MAX_RESULTS, BLOB_LIST_PAGE_SIZE.getValue(),
-          StorageRpc.Option.PREFIX, BLOB_LIST_PREFIX.getValue(),
-          StorageRpc.Option.VERSIONS, BLOB_LIST_VERSIONS.getValue());
-
-  // ACLs
-  private static final Acl ACL = Acl.of(Acl.User.ofAllAuthenticatedUsers(), Acl.Role.OWNER);
-  private static final Acl OTHER_ACL =
-      Acl.of(new Acl.Project(Acl.Project.ProjectRole.OWNERS, "p"), Acl.Role.READER);
+          StorageRpc.Option.MAX_RESULTS,
+          PAGE_SIZE,
+          StorageRpc.Option.PREFIX,
+          "prefix",
+          StorageRpc.Option.VERSIONS,
+          false);
 
   // Customer supplied encryption key options
   private static final Map<StorageRpc.Option, ?> ENCRYPTION_KEY_OPTIONS =
@@ -271,34 +235,6 @@ public class StorageImplMockitoTest {
   // Customer managed encryption key options
   private static final Map<StorageRpc.Option, ?> KMS_KEY_NAME_OPTIONS =
       ImmutableMap.of(StorageRpc.Option.KMS_KEY_NAME, KMS_KEY_NAME);
-  // IAM policies
-  private static final String POLICY_ETAG1 = "CAE=";
-  private static final String POLICY_ETAG2 = "CAI=";
-  private static final Policy LIB_POLICY1 =
-      Policy.newBuilder()
-          .addIdentity(StorageRoles.objectViewer(), Identity.allUsers())
-          .addIdentity(
-              StorageRoles.objectAdmin(),
-              Identity.user("test1@gmail.com"),
-              Identity.user("test2@gmail.com"))
-          .setEtag(POLICY_ETAG1)
-          .setVersion(1)
-          .build();
-
-  private static final ServiceAccount SERVICE_ACCOUNT = ServiceAccount.of("test@google.com");
-
-  private static final com.google.api.services.storage.model.Policy API_POLICY1 =
-      new com.google.api.services.storage.model.Policy()
-          .setBindings(
-              ImmutableList.of(
-                  new com.google.api.services.storage.model.Policy.Bindings()
-                      .setMembers(ImmutableList.of("allUsers"))
-                      .setRole("roles/storage.objectViewer"),
-                  new com.google.api.services.storage.model.Policy.Bindings()
-                      .setMembers(ImmutableList.of("user:test1@gmail.com", "user:test2@gmail.com"))
-                      .setRole("roles/storage.objectAdmin")))
-          .setEtag(POLICY_ETAG1)
-          .setVersion(1);
 
   private static final String PRIVATE_KEY_STRING =
       "MIICdwIBADANBgkqhkiG9w0BAQEFAASCAmEwggJdAgEAAoG"
@@ -318,9 +254,6 @@ public class StorageImplMockitoTest {
           + "EkPPhszldvQTY486uPxyD/D7HdfnGW/Nbw5JUhfvecAdudDEhNAQ3PNabyDMI+TpiHy4NTWOrgdcWrzj6VXcdc"
           + "+uuABnPwRCdcyJ1xl2kOrPksRnp1auNGMLOe4IpEBjGY7baX9UG8+A45MbG0aHmkR59Op/aR9XowIDAQAB";
 
-  private static final String SIGNED_URL =
-      "http://www.test.com/test-bucket/test1.txt?GoogleAccessId=testClient-test@test.com&Expires=1553839761&Signature=MJUBXAZ7";
-
   private static final ApiClock TIME_SOURCE =
       new ApiClock() {
         @Override
@@ -334,39 +267,6 @@ public class StorageImplMockitoTest {
         }
       };
 
-  // List of chars under test were taken from
-  // https://en.wikipedia.org/wiki/Percent-encoding#Percent-encoding_reserved_characters
-  private static final Map<Character, String> RFC3986_URI_ENCODING_MAP =
-      ImmutableMap.<Character, String>builder()
-          .put('!', "%21")
-          .put('#', "%23")
-          .put('$', "%24")
-          .put('&', "%26")
-          .put('\'', "%27")
-          .put('(', "%28")
-          .put(')', "%29")
-          .put('*', "%2A")
-          .put('+', "%2B")
-          .put(',', "%2C")
-          // NOTE: Whether the forward slash character should be encoded depends on the URI segment
-          // being encoded. The path segment should not encode forward slashes, but others (e.g.
-          // query parameter keys and values) should encode them. Tests verifying encoding behavior
-          // in path segments should make a copy of this map and replace the mapping for '/' to "/".
-          .put('/', "%2F")
-          .put(':', "%3A")
-          .put(';', "%3B")
-          .put('=', "%3D")
-          .put('?', "%3F")
-          .put('@', "%40")
-          .put('[', "%5B")
-          .put(']', "%5D")
-          // In addition to [a-zA-Z0-9], these chars should not be URI-encoded:
-          .put('-', "-")
-          .put('_', "_")
-          .put('.', ".")
-          .put('~', "~")
-          .build();
-
   // Notification
   private static final String ETAG = "0xFF00";
   private static final String GENERATED_ID = "B/N:1";
@@ -398,7 +298,6 @@ public class StorageImplMockitoTest {
           .setPayloadFormat(PAYLOAD_FORMAT)
           .build();
 
-  private static final String ACCOUNT = "account";
   private static PrivateKey privateKey;
   private static PublicKey publicKey;
 
@@ -426,16 +325,13 @@ public class StorageImplMockitoTest {
 
   private static final RuntimeException UNEXPECTED_CALL_EXCEPTION =
       new RuntimeException("Unexpected call");
-  private static final Answer UNEXPECTED_CALL_ANSWER =
-      new Answer<Object>() {
-        @Override
-        public Object answer(InvocationOnMock invocation) {
-          throw new IllegalArgumentException(
-              "Unexpected call of "
-                  + invocation.getMethod()
-                  + " with "
-                  + Arrays.toString(invocation.getArguments()));
-        };
+  private static final Answer<?> UNEXPECTED_CALL_ANSWER =
+      invocation -> {
+        throw new IllegalArgumentException(
+            "Unexpected call of "
+                + invocation.getMethod()
+                + " with "
+                + Arrays.toString(invocation.getArguments()));
       };
 
   @Before
@@ -549,9 +445,9 @@ public class StorageImplMockitoTest {
     initializeService();
     Bucket bucket = storage.get(BUCKET_NAME1, BUCKET_GET_METAGENERATION, BUCKET_GET_FIELDS);
     assertEquals(
-        BUCKET_GET_METAGENERATION.getValue(),
-        capturedOptions.getValue().get(BUCKET_GET_METAGENERATION.getRpcOption()));
-    String selector = (String) capturedOptions.getValue().get(BLOB_GET_FIELDS.getRpcOption());
+        BUCKET_INFO1.getMetageneration(),
+        capturedOptions.getValue().get(StorageRpc.Option.IF_METAGENERATION_MATCH));
+    String selector = (String) capturedOptions.getValue().get(StorageRpc.Option.FIELDS);
     assertTrue(selector.contains("name"));
     assertTrue(selector.contains("location"));
     assertTrue(selector.contains("acl"));
@@ -572,9 +468,9 @@ public class StorageImplMockitoTest {
     initializeService();
     Bucket bucket = storage.get(BUCKET_NAME1, BUCKET_GET_METAGENERATION, BUCKET_GET_EMPTY_FIELDS);
     assertEquals(
-        BUCKET_GET_METAGENERATION.getValue(),
-        capturedOptions.getValue().get(BUCKET_GET_METAGENERATION.getRpcOption()));
-    String selector = (String) capturedOptions.getValue().get(BLOB_GET_FIELDS.getRpcOption());
+        BUCKET_INFO1.getMetageneration(),
+        capturedOptions.getValue().get(StorageRpc.Option.IF_METAGENERATION_MATCH));
+    String selector = (String) capturedOptions.getValue().get(StorageRpc.Option.FIELDS);
     assertTrue(selector.contains("name"));
     assertEquals(4, selector.length());
     assertEquals(BUCKET_INFO1.getName(), bucket.getName());
@@ -654,12 +550,12 @@ public class StorageImplMockitoTest {
             BLOB_GET_GENERATION,
             BLOB_GET_FIELDS);
     assertEquals(
-        BLOB_GET_METAGENERATION.getValue(),
-        capturedOptions.getValue().get(BLOB_GET_METAGENERATION.getRpcOption()));
+        BLOB_INFO1.getMetageneration(),
+        capturedOptions.getValue().get(StorageRpc.Option.IF_METAGENERATION_MATCH));
     assertEquals(
-        BLOB_GET_GENERATION.getValue(),
-        capturedOptions.getValue().get(BLOB_GET_GENERATION.getRpcOption()));
-    String selector = (String) capturedOptions.getValue().get(BLOB_GET_FIELDS.getRpcOption());
+        BLOB_INFO1.getGeneration(),
+        capturedOptions.getValue().get(StorageRpc.Option.IF_GENERATION_MATCH));
+    String selector = (String) capturedOptions.getValue().get(StorageRpc.Option.FIELDS);
     assertTrue(selector.contains("bucket"));
     assertTrue(selector.contains("name"));
     assertTrue(selector.contains("contentType"));
@@ -687,12 +583,12 @@ public class StorageImplMockitoTest {
             BLOB_GET_GENERATION,
             BLOB_GET_EMPTY_FIELDS);
     assertEquals(
-        BLOB_GET_METAGENERATION.getValue(),
-        capturedOptions.getValue().get(BLOB_GET_METAGENERATION.getRpcOption()));
+        BLOB_INFO1.getMetageneration(),
+        capturedOptions.getValue().get(StorageRpc.Option.IF_METAGENERATION_MATCH));
     assertEquals(
-        BLOB_GET_GENERATION.getValue(),
-        capturedOptions.getValue().get(BLOB_GET_GENERATION.getRpcOption()));
-    String selector = (String) capturedOptions.getValue().get(BLOB_GET_FIELDS.getRpcOption());
+        BLOB_INFO1.getGeneration(),
+        capturedOptions.getValue().get(StorageRpc.Option.IF_GENERATION_MATCH));
+    String selector = (String) capturedOptions.getValue().get(StorageRpc.Option.FIELDS);
     assertTrue(selector.contains("bucket"));
     assertTrue(selector.contains("name"));
     assertEquals(11, selector.length());
@@ -942,6 +838,7 @@ public class StorageImplMockitoTest {
 
     ByteArrayInputStream fileStream = new ByteArrayInputStream(BLOB_CONTENT);
 
+    // verify that md5 and crc32c are cleared if present when calling create
     doReturn(Conversions.apiary().blobInfo().encode(BLOB_INFO1))
         .doThrow(UNEXPECTED_CALL_EXCEPTION)
         .when(storageRpcMock)
@@ -963,6 +860,7 @@ public class StorageImplMockitoTest {
     ArgumentCaptor<ByteArrayInputStream> capturedStream =
         ArgumentCaptor.forClass(ByteArrayInputStream.class);
 
+    // verify that md5 and crc32c are cleared if present when calling create
     ByteArrayInputStream fileStream = new ByteArrayInputStream(BLOB_CONTENT);
     doReturn(Conversions.apiary().blobInfo().encode(BLOB_INFO1))
         .doThrow(UNEXPECTED_CALL_EXCEPTION)
@@ -986,6 +884,7 @@ public class StorageImplMockitoTest {
   public void testCreateBlobFromStreamWithEncryptionKey() throws IOException {
     ByteArrayInputStream fileStream = new ByteArrayInputStream(BLOB_CONTENT);
 
+    // verify that md5 and crc32c are cleared if present when calling create
     doReturn(Conversions.apiary().blobInfo().encode(BLOB_INFO1))
         .doReturn(Conversions.apiary().blobInfo().encode(BLOB_INFO1))
         .doThrow(UNEXPECTED_CALL_EXCEPTION)
@@ -1011,6 +910,7 @@ public class StorageImplMockitoTest {
 
     ByteArrayInputStream fileStream = new ByteArrayInputStream(BLOB_CONTENT);
 
+    // verify that md5 and crc32c are cleared if present when calling create
     Exception internalErrorException = new StorageException(500, "internalError");
     doThrow(internalErrorException)
         .when(storageRpcMock)
@@ -1289,7 +1189,7 @@ public class StorageImplMockitoTest {
     initializeService();
     ImmutableList<Bucket> bucketList = ImmutableList.of(expectedBucket1, expectedBucket2);
     Page<Bucket> page = storage.list(BUCKET_LIST_FIELDS);
-    String selector = (String) capturedOptions.getValue().get(BUCKET_LIST_FIELDS.getRpcOption());
+    String selector = (String) capturedOptions.getValue().get(StorageRpc.Option.FIELDS);
     assertTrue(selector.contains("items("));
     assertTrue(selector.contains("name"));
     assertTrue(selector.contains("acl"));
@@ -1318,8 +1218,7 @@ public class StorageImplMockitoTest {
     initializeService();
     ImmutableList<Bucket> bucketList = ImmutableList.of(expectedBucket1, expectedBucket2);
     Page<Bucket> page = storage.list(BUCKET_LIST_EMPTY_FIELDS);
-    String selector =
-        (String) capturedOptions.getValue().get(BUCKET_LIST_EMPTY_FIELDS.getRpcOption());
+    String selector = (String) capturedOptions.getValue().get(StorageRpc.Option.FIELDS);
     assertTrue(selector.contains("items("));
     assertTrue(selector.contains("name"));
     assertTrue(selector.contains("nextPageToken"));
@@ -1414,13 +1313,9 @@ public class StorageImplMockitoTest {
     ImmutableList<Blob> blobList = ImmutableList.of(expectedBlob1, expectedBlob2);
     Page<Blob> page =
         storage.list(BUCKET_NAME1, BLOB_LIST_PAGE_SIZE, BLOB_LIST_PREFIX, BLOB_LIST_FIELDS);
-    assertEquals(
-        BLOB_LIST_PAGE_SIZE.getValue(),
-        capturedOptions.getValue().get(BLOB_LIST_PAGE_SIZE.getRpcOption()));
-    assertEquals(
-        BLOB_LIST_PREFIX.getValue(),
-        capturedOptions.getValue().get(BLOB_LIST_PREFIX.getRpcOption()));
-    String selector = (String) capturedOptions.getValue().get(BLOB_LIST_FIELDS.getRpcOption());
+    assertEquals(PAGE_SIZE, capturedOptions.getValue().get(StorageRpc.Option.MAX_RESULTS));
+    assertEquals("prefix", capturedOptions.getValue().get(StorageRpc.Option.PREFIX));
+    String selector = (String) capturedOptions.getValue().get(StorageRpc.Option.FIELDS);
     assertTrue(selector.contains("prefixes"));
     assertTrue(selector.contains("items("));
     assertTrue(selector.contains("bucket"));
@@ -1452,14 +1347,9 @@ public class StorageImplMockitoTest {
     ImmutableList<Blob> blobList = ImmutableList.of(expectedBlob1, expectedBlob2);
     Page<Blob> page =
         storage.list(BUCKET_NAME1, BLOB_LIST_PAGE_SIZE, BLOB_LIST_PREFIX, BLOB_LIST_EMPTY_FIELDS);
-    assertEquals(
-        BLOB_LIST_PAGE_SIZE.getValue(),
-        capturedOptions.getValue().get(BLOB_LIST_PAGE_SIZE.getRpcOption()));
-    assertEquals(
-        BLOB_LIST_PREFIX.getValue(),
-        capturedOptions.getValue().get(BLOB_LIST_PREFIX.getRpcOption()));
-    String selector =
-        (String) capturedOptions.getValue().get(BLOB_LIST_EMPTY_FIELDS.getRpcOption());
+    assertEquals(PAGE_SIZE, capturedOptions.getValue().get(StorageRpc.Option.MAX_RESULTS));
+    assertEquals("prefix", capturedOptions.getValue().get(StorageRpc.Option.PREFIX));
+    String selector = (String) capturedOptions.getValue().get(StorageRpc.Option.FIELDS);
     assertTrue(selector.contains("prefixes"));
     assertTrue(selector.contains("items("));
     assertTrue(selector.contains("bucket"));
@@ -1661,6 +1551,7 @@ public class StorageImplMockitoTest {
 
   @Test
   public void testWriter() {
+    // verify that md5 and crc32c are cleared if present when calling create
     doReturn("upload-id")
         .doThrow(UNEXPECTED_CALL_EXCEPTION)
         .when(storageRpcMock)
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageImplTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageImplTest.java
index 69dc8568..f4e497be 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageImplTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageImplTest.java
@@ -51,7 +51,6 @@ import com.google.cloud.storage.Acl.Role;
 import com.google.cloud.storage.Acl.User;
 import com.google.cloud.storage.Storage.BlobSourceOption;
 import com.google.cloud.storage.Storage.BlobTargetOption;
-import com.google.cloud.storage.Storage.BlobWriteOption;
 import com.google.cloud.storage.Storage.BucketSourceOption;
 import com.google.cloud.storage.Storage.CopyRequest;
 import com.google.cloud.storage.spi.StorageRpcFactory;
@@ -103,12 +102,6 @@ public class StorageImplTest {
   private static final String BLOB_NAME2 = "n2";
   private static final String BLOB_NAME3 = "n3";
   private static final byte[] BLOB_CONTENT = {0xD, 0xE, 0xA, 0xD};
-  private static final byte[] BLOB_SUB_CONTENT = {0xE, 0xA};
-  private static final String CONTENT_MD5 = "O1R4G1HJSDUISJjoIYmVhQ==";
-  private static final String CONTENT_CRC32C = "9N3EPQ==";
-  private static final String SUB_CONTENT_MD5 = "5e7c7CdasUiOn3BO560jPg==";
-  private static final String SUB_CONTENT_CRC32C = "bljNYA==";
-  private static final int DEFAULT_CHUNK_SIZE = 2 * 1024 * 1024;
   private static final String BASE64_KEY = "JVzfVl8NLD9FjedFuStegjRfES5ll5zc59CIXw572OA=";
   private static final Key KEY =
       new SecretKeySpec(BaseEncoding.base64().decode(BASE64_KEY), "AES256");
@@ -116,8 +109,6 @@ public class StorageImplTest {
       "projects/gcloud-devel/locations/us/keyRings/gcs_kms_key_ring_us/cryptoKeys/key";
   private static final Long RETENTION_PERIOD = 10L;
   private static final String USER_PROJECT = "test-project";
-  private static final int DEFAULT_BUFFER_SIZE = 15 * 1024 * 1024;
-  private static final int MIN_BUFFER_SIZE = 256 * 1024;
   // BucketInfo objects
   private static final BucketInfo BUCKET_INFO1 =
       BucketInfo.newBuilder(BUCKET_NAME1).setMetageneration(42L).build();
@@ -152,7 +143,7 @@ public class StorageImplTest {
   private static final Map<StorageRpc.Option, ?> BUCKET_TARGET_OPTIONS =
       ImmutableMap.of(
           StorageRpc.Option.IF_METAGENERATION_MATCH, BUCKET_INFO1.getMetageneration(),
-          StorageRpc.Option.PREDEFINED_ACL, BUCKET_TARGET_PREDEFINED_ACL.getValue());
+          StorageRpc.Option.PREDEFINED_ACL, Storage.PredefinedAcl.PRIVATE.getEntry());
   private static final Map<StorageRpc.Option, ?> BUCKET_TARGET_OPTIONS_LOCK_RETENTION_POLICY =
       ImmutableMap.of(
           StorageRpc.Option.IF_METAGENERATION_MATCH,
@@ -164,66 +155,23 @@ public class StorageImplTest {
   private static final BlobTargetOption BLOB_TARGET_GENERATION = BlobTargetOption.generationMatch();
   private static final BlobTargetOption BLOB_TARGET_METAGENERATION =
       BlobTargetOption.metagenerationMatch();
-  private static final BlobTargetOption BLOB_TARGET_DISABLE_GZIP_CONTENT =
-      BlobTargetOption.disableGzipContent();
-  private static final BlobTargetOption BLOB_TARGET_NOT_EXIST = BlobTargetOption.doesNotExist();
   private static final BlobTargetOption BLOB_TARGET_PREDEFINED_ACL =
       BlobTargetOption.predefinedAcl(Storage.PredefinedAcl.PRIVATE);
-  private static final Map<StorageRpc.Option, ?> BLOB_TARGET_OPTIONS_CREATE =
-      ImmutableMap.of(
-          StorageRpc.Option.IF_METAGENERATION_MATCH, BLOB_INFO1.getMetageneration(),
-          StorageRpc.Option.IF_GENERATION_MATCH, 0L,
-          StorageRpc.Option.PREDEFINED_ACL, BUCKET_TARGET_PREDEFINED_ACL.getValue());
-  private static final Map<StorageRpc.Option, ?> BLOB_TARGET_OPTIONS_CREATE_DISABLE_GZIP_CONTENT =
-      ImmutableMap.of(StorageRpc.Option.IF_DISABLE_GZIP_CONTENT, true);
   private static final Map<StorageRpc.Option, ?> BLOB_TARGET_OPTIONS_UPDATE =
       ImmutableMap.of(
           StorageRpc.Option.IF_METAGENERATION_MATCH, BLOB_INFO1.getMetageneration(),
-          StorageRpc.Option.PREDEFINED_ACL, BUCKET_TARGET_PREDEFINED_ACL.getValue());
+          StorageRpc.Option.PREDEFINED_ACL, Storage.PredefinedAcl.PRIVATE.getEntry());
   private static final Map<StorageRpc.Option, ?> BLOB_TARGET_OPTIONS_COMPOSE =
       ImmutableMap.of(
           StorageRpc.Option.IF_GENERATION_MATCH, BLOB_INFO1.getGeneration(),
           StorageRpc.Option.IF_METAGENERATION_MATCH, BLOB_INFO1.getMetageneration());
 
-  // Blob write options (create, writer)
-  private static final BlobWriteOption BLOB_WRITE_METAGENERATION =
-      BlobWriteOption.metagenerationMatch();
-  private static final BlobWriteOption BLOB_WRITE_NOT_EXIST = BlobWriteOption.doesNotExist();
-  private static final BlobWriteOption BLOB_WRITE_PREDEFINED_ACL =
-      BlobWriteOption.predefinedAcl(Storage.PredefinedAcl.PRIVATE);
-  private static final BlobWriteOption BLOB_WRITE_MD5_HASH = BlobWriteOption.md5Match();
-  private static final BlobWriteOption BLOB_WRITE_CRC2C = BlobWriteOption.crc32cMatch();
-
   // Bucket get/source options
   private static final BucketSourceOption BUCKET_SOURCE_METAGENERATION =
       BucketSourceOption.metagenerationMatch(BUCKET_INFO1.getMetageneration());
   private static final Map<StorageRpc.Option, ?> BUCKET_SOURCE_OPTIONS =
-      ImmutableMap.of(
-          StorageRpc.Option.IF_METAGENERATION_MATCH, BUCKET_SOURCE_METAGENERATION.getValue());
-  private static final Storage.BucketGetOption BUCKET_GET_METAGENERATION =
-      Storage.BucketGetOption.metagenerationMatch(BUCKET_INFO1.getMetageneration());
-  private static final Storage.BucketGetOption BUCKET_GET_FIELDS =
-      Storage.BucketGetOption.fields(Storage.BucketField.LOCATION, Storage.BucketField.ACL);
-  private static final Storage.BucketGetOption BUCKET_GET_EMPTY_FIELDS =
-      Storage.BucketGetOption.fields();
-  private static final Map<StorageRpc.Option, ?> BUCKET_GET_OPTIONS =
-      ImmutableMap.of(
-          StorageRpc.Option.IF_METAGENERATION_MATCH, BUCKET_SOURCE_METAGENERATION.getValue());
-
-  // Blob get/source options
-  private static final Storage.BlobGetOption BLOB_GET_METAGENERATION =
-      Storage.BlobGetOption.metagenerationMatch(BLOB_INFO1.getMetageneration());
-  private static final Storage.BlobGetOption BLOB_GET_GENERATION =
-      Storage.BlobGetOption.generationMatch(BLOB_INFO1.getGeneration());
-  private static final Storage.BlobGetOption BLOB_GET_GENERATION_FROM_BLOB_ID =
-      Storage.BlobGetOption.generationMatch();
-  private static final Storage.BlobGetOption BLOB_GET_FIELDS =
-      Storage.BlobGetOption.fields(Storage.BlobField.CONTENT_TYPE, Storage.BlobField.CRC32C);
-  private static final Storage.BlobGetOption BLOB_GET_EMPTY_FIELDS = Storage.BlobGetOption.fields();
-  private static final Map<StorageRpc.Option, ?> BLOB_GET_OPTIONS =
-      ImmutableMap.of(
-          StorageRpc.Option.IF_METAGENERATION_MATCH, BLOB_GET_METAGENERATION.getValue(),
-          StorageRpc.Option.IF_GENERATION_MATCH, BLOB_GET_GENERATION.getValue());
+      ImmutableMap.of(StorageRpc.Option.IF_METAGENERATION_MATCH, BUCKET_INFO1.getMetageneration());
+
   private static final BlobSourceOption BLOB_SOURCE_METAGENERATION =
       BlobSourceOption.metagenerationMatch(BLOB_INFO1.getMetageneration());
   private static final BlobSourceOption BLOB_SOURCE_GENERATION =
@@ -232,43 +180,12 @@ public class StorageImplTest {
       BlobSourceOption.generationMatch();
   private static final Map<StorageRpc.Option, ?> BLOB_SOURCE_OPTIONS =
       ImmutableMap.of(
-          StorageRpc.Option.IF_METAGENERATION_MATCH, BLOB_SOURCE_METAGENERATION.getValue(),
-          StorageRpc.Option.IF_GENERATION_MATCH, BLOB_SOURCE_GENERATION.getValue());
+          StorageRpc.Option.IF_METAGENERATION_MATCH, BLOB_INFO1.getMetageneration(),
+          StorageRpc.Option.IF_GENERATION_MATCH, BLOB_INFO1.getGeneration());
   private static final Map<StorageRpc.Option, ?> BLOB_SOURCE_OPTIONS_COPY =
       ImmutableMap.of(
-          StorageRpc.Option.IF_SOURCE_METAGENERATION_MATCH, BLOB_SOURCE_METAGENERATION.getValue(),
-          StorageRpc.Option.IF_SOURCE_GENERATION_MATCH, BLOB_SOURCE_GENERATION.getValue());
-
-  // Bucket list options
-  private static final Storage.BucketListOption BUCKET_LIST_PAGE_SIZE =
-      Storage.BucketListOption.pageSize(42L);
-  private static final Storage.BucketListOption BUCKET_LIST_PREFIX =
-      Storage.BucketListOption.prefix("prefix");
-  private static final Storage.BucketListOption BUCKET_LIST_FIELDS =
-      Storage.BucketListOption.fields(Storage.BucketField.LOCATION, Storage.BucketField.ACL);
-  private static final Storage.BucketListOption BUCKET_LIST_EMPTY_FIELDS =
-      Storage.BucketListOption.fields();
-  private static final Map<StorageRpc.Option, ?> BUCKET_LIST_OPTIONS =
-      ImmutableMap.of(
-          StorageRpc.Option.MAX_RESULTS, BUCKET_LIST_PAGE_SIZE.getValue(),
-          StorageRpc.Option.PREFIX, BUCKET_LIST_PREFIX.getValue());
-
-  // Blob list options
-  private static final Storage.BlobListOption BLOB_LIST_PAGE_SIZE =
-      Storage.BlobListOption.pageSize(42L);
-  private static final Storage.BlobListOption BLOB_LIST_PREFIX =
-      Storage.BlobListOption.prefix("prefix");
-  private static final Storage.BlobListOption BLOB_LIST_FIELDS =
-      Storage.BlobListOption.fields(Storage.BlobField.CONTENT_TYPE, Storage.BlobField.MD5HASH);
-  private static final Storage.BlobListOption BLOB_LIST_VERSIONS =
-      Storage.BlobListOption.versions(false);
-  private static final Storage.BlobListOption BLOB_LIST_EMPTY_FIELDS =
-      Storage.BlobListOption.fields();
-  private static final Map<StorageRpc.Option, ?> BLOB_LIST_OPTIONS =
-      ImmutableMap.of(
-          StorageRpc.Option.MAX_RESULTS, BLOB_LIST_PAGE_SIZE.getValue(),
-          StorageRpc.Option.PREFIX, BLOB_LIST_PREFIX.getValue(),
-          StorageRpc.Option.VERSIONS, BLOB_LIST_VERSIONS.getValue());
+          StorageRpc.Option.IF_SOURCE_METAGENERATION_MATCH, BUCKET_INFO1.getMetageneration(),
+          StorageRpc.Option.IF_SOURCE_GENERATION_MATCH, BLOB_INFO1.getGeneration());
 
   // ACLs
   private static final Acl ACL = Acl.of(User.ofAllAuthenticatedUsers(), Role.OWNER);
@@ -328,9 +245,6 @@ public class StorageImplTest {
           + "EkPPhszldvQTY486uPxyD/D7HdfnGW/Nbw5JUhfvecAdudDEhNAQ3PNabyDMI+TpiHy4NTWOrgdcWrzj6VXcdc"
           + "+uuABnPwRCdcyJ1xl2kOrPksRnp1auNGMLOe4IpEBjGY7baX9UG8+A45MbG0aHmkR59Op/aR9XowIDAQAB";
 
-  private static final String SIGNED_URL =
-      "http://www.test.com/test-bucket/test1.txt?GoogleAccessId=testClient-test@test.com&Expires=1553839761&Signature=MJUBXAZ7";
-
   private static final ApiClock TIME_SOURCE =
       new ApiClock() {
         @Override
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/UnifiedOptsTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/UnifiedOptsTest.java
new file mode 100644
index 00000000..82a2d9b0
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/UnifiedOptsTest.java
@@ -0,0 +1,38 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage;
+
+import static com.google.common.truth.Truth.assertThat;
+import static org.junit.Assert.assertThrows;
+
+import com.google.cloud.storage.UnifiedOpts.Opts;
+import org.junit.Test;
+
+public final class UnifiedOptsTest {
+
+  @Test
+  public void opts_validation_uniqueKeys() {
+    IllegalArgumentException iae =
+        assertThrows(
+            IllegalArgumentException.class,
+            () ->
+                Opts.from(UnifiedOpts.generationMatch(1), UnifiedOpts.generationMatch(2))
+                    .getRpcOptions());
+
+    assertThat(iae).hasMessageThat().contains("GENERATION_MATCH");
+  }
+}