samples(storage): add a sample for downloading chunks concurrently using Transfer Manager (#5402)

* Add sample download_chunks_concurrently with test.

* Improve test and comments.

* Improve cleanup in download_chunks_concurrently_test.go

* Improve comments and tests for download_chunks_concurrently

* Remove unnecessary space.

* Resolve comments.

Author: cpriti-os

storage/transfer_manager/download_chunks_concurrently.go

@@ -0,0 +1,93 @@
+// Copyright 2025 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.
+
+// [START storage_download_chunks_concurrently]
+package transfermanager
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"os"
+
+	"cloud.google.com/go/storage"
+	"cloud.google.com/go/storage/transfermanager"
+)
+
+// downloadChunksConcurrently downloads a single file in chunks, concurrently in a process pool.
+func downloadChunksConcurrently(w io.Writer, bucketName, blobName, filename string) error {
+	// bucketName := "your-bucket-name"
+	// blobName := "target-file"
+	// filename := "path/to/your/local/file.txt"
+
+	// The chunkSize is the size of each chunk to be downloaded.
+	// The performance impact of this value depends on the use case.
+	// For example, for a slow network, using a smaller chunkSize may be better.
+	// Providing this parameter is optional and the default value is 32 MiB.
+	chunkSize := 16 * 1024 * 1024 // 16 MiB
+
+	// The maximum number of workers to use for the operation.
+	// Please note, providing this parameter is optional.
+	// The performance impact of this value depends on the use case.
+	// To download one large file, the default value: NumCPU / 2 is usually fine.
+	workers := 8
+
+	ctx := context.Background()
+	client, err := storage.NewClient(ctx)
+	if err != nil {
+		return fmt.Errorf("storage.NewClient: %w", err)
+	}
+	defer client.Close()
+
+	d, err := transfermanager.NewDownloader(client, transfermanager.WithPartSize(int64(chunkSize)), transfermanager.WithWorkers(workers))
+	if err != nil {
+		return fmt.Errorf("transfermanager.NewDownloader: %w", err)
+	}
+
+	f, err := os.Create(filename)
+	if err != nil {
+		return fmt.Errorf("os.Create: %w", err)
+	}
+
+	in := &transfermanager.DownloadObjectInput{
+		Bucket:      bucketName,
+		Object:      blobName,
+		Destination: f,
+	}
+
+	if err := d.DownloadObject(ctx, in); err != nil {
+		return fmt.Errorf("d.DownloadObject: %w", err)
+	}
+
+	// Wait for all downloads to complete and close the downloader.
+	// This allows to synchronize the download processes.
+	results, err := d.WaitAndClose()
+	if err != nil {
+		return fmt.Errorf("d.WaitAndClose: %w", err)
+	}
+
+	// Process the downloader result.
+	if len(results) != 1 {
+		return fmt.Errorf("expected 1 result, got %d", len(results))
+	}
+	result := results[0]
+	if result.Err != nil {
+		fmt.Fprintf(w, "download of %v failed with error %v\n", result.Object, result.Err)
+	}
+	fmt.Fprintf(w, "Downloaded %v to %v.\n", blobName, filename)
+
+	return nil
+}
+
+// [END storage_download_chunks_concurrently]

storage/transfer_manager/download_chunks_concurrently_test.go

@@ -0,0 +1,116 @@
+// Copyright 2025 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 transfermanager
+
+import (
+	"bytes"
+	"context"
+	"crypto/rand"
+	"fmt"
+	"io"
+	"log"
+	"os"
+	"strings"
+	"testing"
+
+	"cloud.google.com/go/storage"
+	"github.com/GoogleCloudPlatform/golang-samples/internal/testutil"
+	"github.com/google/uuid"
+)
+
+const (
+	testPrefix     = "storage-objects-test"
+	downloadObject = "tm-obj-download"
+)
+
+var (
+	tmBucketName  string
+	storageClient *storage.Client
+	downloadData  []byte
+)
+
+func TestMain(m *testing.M) {
+	ctx := context.Background()
+	tc, _ := testutil.ContextMain(m)
+
+	var err error
+
+	// Create fixture client & bucket to use across tests.
+	storageClient, err = storage.NewClient(ctx)
+	if err != nil {
+		log.Fatalf("storage.NewClient: %v", err)
+	}
+	tmBucketName = fmt.Sprintf("%s-%s", testPrefix, uuid.New().String())
+	bucket := storageClient.Bucket(tmBucketName)
+	if err := bucket.Create(ctx, tc.ProjectID, nil); err != nil {
+		log.Fatalf("Bucket(%q).Create: %v", tmBucketName, err)
+	}
+
+	// Create object fixture for download tests.
+	w := bucket.Object(downloadObject).NewWriter(ctx)
+	downloadData = make([]byte, 2*1024*1024) // 2 MiB
+	if _, err := rand.Read(downloadData); err != nil {
+		log.Fatalf("rand.Read: %v", err)
+	}
+	if _, err := io.Copy(w, bytes.NewReader(downloadData)); err != nil {
+		log.Fatalf("uploading object: %v", err)
+	}
+	if err := w.Close(); err != nil {
+		log.Fatalf("closing writer: %v", err)
+	}
+
+	// Run tests.
+	exitCode := m.Run()
+
+	// Cleanup bucket and objects.
+	if err := testutil.DeleteBucketIfExists(ctx, storageClient, tmBucketName); err != nil {
+		log.Printf("deleting bucket: %v", err)
+	}
+	os.Exit(exitCode)
+}
+
+func TestDownloadChunksConcurrently(t *testing.T) {
+	bucketName := tmBucketName
+	blobName := downloadObject
+
+	// Create a temporary file to download to, ensuring we have permissions
+	// and the file is cleaned up.
+	f, err := os.CreateTemp("", "tm-file-test-")
+	if err != nil {
+		t.Fatalf("os.CreateTemp: %v", err)
+	}
+	fileName := f.Name()
+	f.Close() // Close the file so the download can write to it.
+	defer os.Remove(fileName)
+
+	var buf bytes.Buffer
+	if err := downloadChunksConcurrently(&buf, bucketName, blobName, fileName); err != nil {
+		t.Errorf("downloadChunksConcurrently: %v", err)
+	}
+
+	if got, want := buf.String(), fmt.Sprintf("Downloaded %v to %v", blobName, fileName); !strings.Contains(got, want) {
+		t.Errorf("got %q, want to contain %q", got, want)
+	}
+
+	// Verify that the downloaded data is the same as the uploaded data.
+	downloadedBytes, err := os.ReadFile(fileName)
+	if err != nil {
+		t.Fatalf("os.ReadFile(%q): %v", fileName, err)
+	}
+
+	if !bytes.Equal(downloadedBytes, downloadData) {
+		t.Errorf("downloaded data does not match uploaded data. got %d bytes, want %d bytes", len(downloadedBytes), len(downloadData))
+	}
+}