class StorageRequestAuthorizer {

  public void authorize(StorageRequest<?> request) throws AccessDeniedException {
    logger.atFine().log("authorizeStorageRequest(%s)", request);

    // Objects
    if (request instanceof Storage.Objects.List) {
      Storage.Objects.List listRequest = (Storage.Objects.List) request;
      authorizationHandler.handleListObjects(
          getGcsUri(listRequest.getBucket(), listRequest.getPrefix()));
    } else if (request instanceof Storage.Objects.Insert) {
      Storage.Objects.Insert insertRequest = (Storage.Objects.Insert) request;
      authorizationHandler.handleInsertObject(
          getGcsUri(insertRequest.getBucket(), ((StorageObject) getData(request)).getName()));
    } else if (request instanceof Storage.Objects.Compose) {
      Storage.Objects.Compose composeRequest = (Storage.Objects.Compose) request;
      String bucket = composeRequest.getDestinationBucket();
      URI destination = getGcsUri(bucket, composeRequest.getDestinationObject());
      List<URI> sources =
          ((ComposeRequest) getData(request))
              .getSourceObjects().stream()
                  .map(source -> getGcsUri(bucket, source.getName()))
                  .collect(toImmutableList());
      authorizationHandler.handleComposeObject(destination, sources);
    } else if (request instanceof Storage.Objects.Get) {
      Storage.Objects.Get getRequest = (Storage.Objects.Get) request;
      authorizationHandler.handleGetObject(
          getGcsUri(getRequest.getBucket(), getRequest.getObject()));
    } else if (request instanceof Storage.Objects.Delete) {
      Storage.Objects.Delete deleteRequest = (Storage.Objects.Delete) request;
      authorizationHandler.handleDeleteObject(
          getGcsUri(deleteRequest.getBucket(), deleteRequest.getObject()));
    } else if (request instanceof Storage.Objects.Rewrite) {
      Storage.Objects.Rewrite rewriteRequest = (Storage.Objects.Rewrite) request;
      authorizationHandler.handleRewriteObject(
          getGcsUri(rewriteRequest.getSourceBucket(), rewriteRequest.getSourceObject()),
          getGcsUri(rewriteRequest.getDestinationBucket(), rewriteRequest.getDestinationObject()));

    } else if (request instanceof Storage.Objects.Copy) {
      Storage.Objects.Copy copyRequest = (Storage.Objects.Copy) request;
      authorizationHandler.handleCopyObject(
          getGcsUri(copyRequest.getSourceBucket(), copyRequest.getSourceObject()),
          getGcsUri(copyRequest.getDestinationBucket(), copyRequest.getDestinationObject()));
    } else if (request instanceof Storage.Objects.Patch) {
      Storage.Objects.Patch patchRequest = (Storage.Objects.Patch) request;
      authorizationHandler.handlePatchObject(
          getGcsUri(patchRequest.getBucket(), patchRequest.getObject()));
    }

    // Buckets
    else if (request instanceof Storage.Buckets.List) {
      authorizationHandler.handleListBuckets(((Storage.Buckets.List) request).getProject());
    } else if (request instanceof Storage.Buckets.Insert) {
      authorizationHandler.handleInsertBucket(
          ((Storage.Buckets.Insert) request).getProject(),
          getGcsUri(((Bucket) getData(request)).getName(), /* objectPath= */ null));
    } else if (request instanceof Storage.Buckets.Get) {
      authorizationHandler.handleGetBucket(
          getGcsUri(((Storage.Buckets.Get) request).getBucket(), /* objectPath= */ null));
    } else if (request instanceof Storage.Buckets.Delete) {
      authorizationHandler.handleDeleteBucket(
          getGcsUri(((Storage.Buckets.Delete) request).getBucket(), /* objectPath= */ null));
    }

    // Unhandled request types. All storage request should be covered by the previous checks.
    else {
      throw new RuntimeException("Unhandled storage request type. Request: " + request);
    }
  }

  public  StorageRequestAuthorizer(
      Class<? extends AuthorizationHandler> handlerClass, Map<String, String> properties);
  @VisibleForTesting  StorageRequestAuthorizer(AuthorizationHandler authorizationHandler);

}