feat: Initial Preview of Transfer Manager #2105

Merge pull request from googleapis/feat/transfer-manager

google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/ChunkedDownloadCallable.java

@@ -0,0 +1,98 @@
+/*
+ * Copyright 2023 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.transfermanager;
+
+import com.google.cloud.ReadChannel;
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.Storage.BlobSourceOption;
+import com.google.cloud.storage.StorageException;
+import com.google.common.io.ByteStreams;
+import java.nio.channels.FileChannel;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+import java.util.concurrent.Callable;
+
+final class ChunkedDownloadCallable implements Callable<DownloadSegment> {
+
+  private final BlobInfo originalBlob;
+
+  private final Storage storage;
+
+  private final Storage.BlobSourceOption[] opts;
+
+  private final long startPosition;
+
+  private final long endPosition;
+  private final Path destPath;
+
+  ChunkedDownloadCallable(
+      Storage storage,
+      BlobInfo originalBlob,
+      BlobSourceOption[] opts,
+      Path destPath,
+      long startPosition,
+      long endPosition) {
+    this.originalBlob = originalBlob;
+    this.storage = storage;
+    this.opts = opts;
+    this.startPosition = startPosition;
+    this.endPosition = endPosition;
+    this.destPath = destPath;
+  }
+
+  @Override
+  public DownloadSegment call() {
+    long bytesCopied = -1L;
+    try (ReadChannel rc = storage.reader(originalBlob.getBlobId(), opts)) {
+      FileChannel wc =
+          FileChannel.open(destPath, StandardOpenOption.WRITE, StandardOpenOption.CREATE);
+      rc.seek(startPosition);
+      rc.limit(endPosition);
+      wc.position(startPosition);
+      bytesCopied = ByteStreams.copy(rc, wc);
+      long bytesExpected = endPosition - startPosition;
+      if (bytesCopied != bytesExpected) {
+        return DownloadSegment.newBuilder(originalBlob, TransferStatus.FAILED_TO_FINISH)
+            .setException(
+                new StorageException(
+                    0,
+                    "Unexpected end of stream, read "
+                        + bytesCopied
+                        + " expected "
+                        + bytesExpected
+                        + " from object "
+                        + originalBlob.getBlobId().toGsUtilUriWithGeneration()))
+            .build();
+      }
+    } catch (Exception e) {
+      if (bytesCopied == -1) {
+        return DownloadSegment.newBuilder(originalBlob, TransferStatus.FAILED_TO_START)
+            .setException(e)
+            .build();
+      }
+      return DownloadSegment.newBuilder(originalBlob, TransferStatus.FAILED_TO_FINISH)
+          .setException(e)
+          .build();
+    }
+    DownloadSegment result =
+        DownloadSegment.newBuilder(originalBlob, TransferStatus.SUCCESS)
+            .setOutputDestination(destPath)
+            .build();
+    return result;
+  }
+}

google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/DefaultQos.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright 2023 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.transfermanager;
+
+final class DefaultQos implements Qos {
+
+  private final long divideAndConquerThreshold;
+
+  private DefaultQos(long divideAndConquerThreshold) {
+    this.divideAndConquerThreshold = divideAndConquerThreshold;
+  }
+
+  @Override
+  public boolean divideAndConquer(long objectSize) {
+    return objectSize > divideAndConquerThreshold;
+  }
+
+  static DefaultQos of() {
+    return of(128L * 1024 * 1024);
+  }
+
+  static DefaultQos of(long divideAndConquerThreshold) {
+    return new DefaultQos(divideAndConquerThreshold);
+  }
+}

google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/DirectDownloadCallable.java

@@ -0,0 +1,95 @@
+/*
+ * Copyright 2023 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.transfermanager;
+
+import com.google.cloud.ReadChannel;
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.Storage.BlobSourceOption;
+import com.google.cloud.storage.StorageException;
+import com.google.common.io.ByteStreams;
+import java.nio.channels.FileChannel;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+import java.util.concurrent.Callable;
+
+final class DirectDownloadCallable implements Callable<DownloadResult> {
+  private final BlobInfo originalBlob;
+
+  private final ParallelDownloadConfig parallelDownloadConfig;
+  private final Storage storage;
+
+  private final Storage.BlobSourceOption[] opts;
+
+  DirectDownloadCallable(
+      Storage storage,
+      BlobInfo originalBlob,
+      ParallelDownloadConfig parallelDownloadConfig,
+      BlobSourceOption[] opts) {
+    this.originalBlob = originalBlob;
+    this.parallelDownloadConfig = parallelDownloadConfig;
+    this.storage = storage;
+    this.opts = opts;
+  }
+
+  @Override
+  public DownloadResult call() {
+    Path path = TransferManagerUtils.createDestPath(parallelDownloadConfig, originalBlob);
+    long bytesCopied = -1L;
+    try (ReadChannel rc =
+        storage.reader(
+            BlobId.of(parallelDownloadConfig.getBucketName(), originalBlob.getName()), opts)) {
+      FileChannel wc =
+          FileChannel.open(
+              path,
+              StandardOpenOption.WRITE,
+              StandardOpenOption.CREATE,
+              StandardOpenOption.TRUNCATE_EXISTING);
+      bytesCopied = ByteStreams.copy(rc, wc);
+      if (originalBlob.getSize() != null) {
+        if (bytesCopied != originalBlob.getSize()) {
+          return DownloadResult.newBuilder(originalBlob, TransferStatus.FAILED_TO_FINISH)
+              .setException(
+                  new StorageException(
+                      0,
+                      "Unexpected end of stream, read "
+                          + bytesCopied
+                          + " expected "
+                          + originalBlob.getSize()
+                          + " from object "
+                          + originalBlob.getBlobId().toGsUtilUriWithGeneration()))
+              .build();
+        }
+      }
+    } catch (Exception e) {
+      if (bytesCopied == -1) {
+        return DownloadResult.newBuilder(originalBlob, TransferStatus.FAILED_TO_START)
+            .setException(e)
+            .build();
+      }
+      return DownloadResult.newBuilder(originalBlob, TransferStatus.FAILED_TO_FINISH)
+          .setException(e)
+          .build();
+    }
+    DownloadResult result =
+        DownloadResult.newBuilder(originalBlob, TransferStatus.SUCCESS)
+            .setOutputDestination(path.toAbsolutePath())
+            .build();
+    return result;
+  }
+}

google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/DownloadJob.java

@@ -0,0 +1,156 @@
+/*
+ * Copyright 2023 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.transfermanager;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.api.core.ApiFuture;
+import com.google.api.core.ApiFutures;
+import com.google.api.core.BetaApi;
+import com.google.api.gax.rpc.ApiExceptions;
+import com.google.common.base.MoreObjects;
+import com.google.common.collect.ImmutableList;
+import java.util.List;
+import java.util.Objects;
+import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
+import org.checkerframework.checker.nullness.qual.NonNull;
+
+/**
+ * A parallel download job sent to Transfer Manager.
+ *
+ * @see Builder
+ */
+@BetaApi
+public final class DownloadJob {
+
+  @NonNull private final List<ApiFuture<DownloadResult>> downloadResults;
+
+  @NonNull private final ParallelDownloadConfig parallelDownloadConfig;
+
+  private DownloadJob(
+      @NonNull List<ApiFuture<DownloadResult>> downloadResults,
+      @NonNull ParallelDownloadConfig parallelDownloadConfig) {
+    this.downloadResults = downloadResults;
+    this.parallelDownloadConfig = parallelDownloadConfig;
+  }
+
+  /**
+   * The list of {@link DownloadResult DownloadResults} for each download request Transfer Manager
+   * executed for this job. Note calling this method will block the invoking thread until all
+   * download requests are complete.
+   *
+   * @see Builder#setDownloadResults(List)
+   */
+  @BetaApi
+  public @NonNull List<DownloadResult> getDownloadResults() {
+    return ApiExceptions.callAndTranslateApiException(ApiFutures.allAsList(downloadResults));
+  }
+
+  /**
+   * The {@link ParallelDownloadConfig} used for this DownloadJob.
+   *
+   * @see Builder#setParallelDownloadConfig(ParallelDownloadConfig)
+   */
+  @BetaApi
+  public @NonNull ParallelDownloadConfig getParallelDownloadConfig() {
+    return parallelDownloadConfig;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (!(o instanceof DownloadJob)) {
+      return false;
+    }
+    DownloadJob that = (DownloadJob) o;
+    return downloadResults.equals(that.downloadResults)
+        && parallelDownloadConfig.equals(that.parallelDownloadConfig);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(downloadResults, parallelDownloadConfig);
+  }
+
+  @Override
+  public String toString() {
+    return MoreObjects.toStringHelper(this)
+        .add("downloadResults", downloadResults)
+        .add("parallelDownloadConfig", parallelDownloadConfig)
+        .toString();
+  }
+
+  @BetaApi
+  public static Builder newBuilder() {
+    return new Builder();
+  }
+
+  /**
+   * Builds an instance of DownloadJob
+   *
+   * @see DownloadJob
+   */
+  @BetaApi
+  public static final class Builder {
+
+    private @NonNull List<ApiFuture<DownloadResult>> downloadResults;
+    private @MonotonicNonNull ParallelDownloadConfig parallelDownloadConfig;
+
+    private Builder() {
+      this.downloadResults = ImmutableList.of();
+    }
+
+    /**
+     * Sets the results for a DownloadJob being performed by Transfer Manager.
+     *
+     * @return the instance of the Builder with DownloadResults modified.
+     * @see DownloadJob#getDownloadResults()
+     */
+    @BetaApi
+    public Builder setDownloadResults(@NonNull List<ApiFuture<DownloadResult>> downloadResults) {
+      this.downloadResults = ImmutableList.copyOf(downloadResults);
+      return this;
+    }
+
+    /**
+     * Sets the {@link ParallelDownloadConfig} used for this DownloadJob.
+     *
+     * @return the instance of the Builder with ParallelDownloadConfig modified.
+     * @see DownloadJob#getParallelDownloadConfig()
+     */
+    @BetaApi
+    public Builder setParallelDownloadConfig(
+        @NonNull ParallelDownloadConfig parallelDownloadConfig) {
+      this.parallelDownloadConfig = parallelDownloadConfig;
+      return this;
+    }
+
+    /**
+     * Creates a DownloadJob object.
+     *
+     * @return {@link DownloadJob}
+     */
+    @BetaApi
+    public DownloadJob build() {
+      checkNotNull(downloadResults);
+      checkNotNull(parallelDownloadConfig);
+      return new DownloadJob(downloadResults, parallelDownloadConfig);
+    }
+  }
+}