Read and append to objects in zonal buckets

namespace gcs = google::cloud::storage;
auto coro = [](gcs::AsyncClient& client, std::string bucket_name,
               std::string object_name)
    -> google::cloud::future<google::storage::v2::Object> {
  auto [writer, token] =
      (co_await client.StartAppendableObjectUpload(
           gcs::BucketName(std::move(bucket_name)), std::move(object_name)))
          .value();
  std::cout << "Appendable upload started for object " << object_name << "\n";

  token = (co_await writer.Write(std::move(token),
                                 gcs::WritePayload("Some data\n")))
              .value();
  std::cout << "Wrote initial data.\n";

  // Flush the buffered data to the service. This is not a terminal
  // operation. The writer can be used after the flush completes.
  // After a flush, the data is visible to readers.
  auto flush_status = co_await writer.Flush();
  if (!flush_status.ok()) throw std::runtime_error(flush_status.message());

  std::cout << "Flush completed. Persisted size is now "
            << std::get<std::int64_t>(writer.PersistedState()) << "\n";

  // The writer is still open. We can write more data.
  token = (co_await writer.Write(std::move(token),
                                 gcs::WritePayload("Some more data\n")))
              .value();
  std::cout << "Wrote more data.\n";

  // Finalize the upload to make it a regular object.
  co_return (co_await writer.Finalize(std::move(token))).value();
};

import (
	"context"
	"fmt"
	"io"
	"time"

	"cloud.google.com/go/storage"
)

// createAndWriteAppendableObject creates and uploads a new appendable object in
// a bucket with the Rapid storage class. The object will not be finalized.
func createAndWriteAppendableObject(w io.Writer, bucket, object string) error {
	// bucket := "bucket-name"
	// object := "object-name"
	ctx := context.Background()
	client, err := storage.NewGRPCClient(ctx, storage.WithAppendableUploads())
	if err != nil {
		return fmt.Errorf("storage.NewGRPCClient: %w", err)
	}
	defer client.Close()

	ctx, cancel := context.WithTimeout(ctx, time.Second*10)
	defer cancel()

	// Create a Writer and write some data.
	writer := client.Bucket(bucket).Object(object).NewWriter(ctx)

	if _, err := writer.Write([]byte("Some data\n")); err != nil {
		return fmt.Errorf("Writer.Write: %w", err)
	}

	// Flush the buffered data to the service. This is not a terminal
	// operation. The Writer can be used after the flush completes.
	// After a flush, the data is visible to readers.
	size, err := writer.Flush()
	if err != nil {
		return fmt.Errorf("Writer.Flush: %w", err)
	}
	fmt.Fprintf(w, "Flush completed. Persisted size is now %d\n", size)

	// The Writer is still open. We can write more data.
	if _, err := writer.Write([]byte("Some more data\n")); err != nil {
		return fmt.Errorf("Writer.Write: %w", err)
	}

	// Close the Writer to flush any remaining buffered data.
	// The object will be unfinalized, which means another writer can
	// later append to the object.
	if err := writer.Close(); err != nil {
		return fmt.Errorf("Writer.Close: %w", err)
	}
	fmt.Fprintf(w, "Uploaded object %v\n", object)

	return nil
}

import com.google.cloud.storage.BlobAppendableUpload;
import com.google.cloud.storage.BlobAppendableUpload.AppendableUploadWriteableByteChannel;
import com.google.cloud.storage.BlobAppendableUploadConfig;
import com.google.cloud.storage.BlobAppendableUploadConfig.CloseAction;
import com.google.cloud.storage.BlobId;
import com.google.cloud.storage.BlobInfo;
import com.google.cloud.storage.FlushPolicy;
import com.google.cloud.storage.Storage;
import com.google.cloud.storage.StorageOptions;
import com.google.common.io.ByteStreams;
import java.io.IOException;
import java.nio.channels.FileChannel;
import java.nio.channels.ReadableByteChannel;
import java.nio.file.Paths;
import java.util.Locale;

public class CreateAndWriteAppendableObject {
  public static void createAndWriteAppendableObject(
      String bucketName, String objectName, String filePath) throws Exception {
    // The ID of your GCS bucket
    // String bucketName = "your-unique-bucket-name";

    // The ID of your GCS object
    // String objectName = "your-object-name";

    // The path to the file to upload
    // String filePath = "path/to/your/file";

    try (Storage storage = StorageOptions.grpc().build().getService()) {
      BlobId blobId = BlobId.of(bucketName, objectName);
      BlobInfo blobInfo = BlobInfo.newBuilder(blobId).build();

      int flushSize = 64 * 1000;
      FlushPolicy.MaxFlushSizeFlushPolicy flushPolicy = FlushPolicy.maxFlushSize(flushSize);
      BlobAppendableUploadConfig config =
          BlobAppendableUploadConfig.of()
              .withCloseAction(CloseAction.FINALIZE_WHEN_CLOSING)
              .withFlushPolicy(flushPolicy);
      BlobAppendableUpload uploadSession = storage.blobAppendableUpload(blobInfo, config);
      try (AppendableUploadWriteableByteChannel channel = uploadSession.open();
          ReadableByteChannel readableByteChannel = FileChannel.open(Paths.get(filePath))) {
        ByteStreams.copy(readableByteChannel, channel);
        // Since the channel is in a try-with-resources block, channel.close()
        // will be implicitly called here, which triggers the finalization.
      } catch (IOException ex) {
        throw new IOException("Failed to upload to object " + blobId.toGsUtilUri(), ex);
      }
      BlobInfo result = storage.get(blobId);
      System.out.printf(
          Locale.US,
          "Object %s successfully uploaded",
          result.getBlobId().toGsUtilUriWithGeneration());
    }
  }
}

async def storage_create_and_write_appendable_object(
    bucket_name, object_name, grpc_client=None
):
    """Uploads an appendable object to zonal bucket.

    grpc_client: an existing grpc_client to use, this is only for testing.
    """

    if grpc_client is None:
        grpc_client = AsyncGrpcClient()
    writer = AsyncAppendableObjectWriter(
        client=grpc_client,
        bucket_name=bucket_name,
        object_name=object_name,
        generation=0,  # throws `FailedPrecondition` if object already exists.
    )
    # This creates a new appendable object of size 0 and opens it for appending.
    await writer.open()

    # appends data to the object
    # you can perform `.append` multiple times as needed. Data will be appended
    # to the end of the object.
    await writer.append(b"Some data")

    # Once all appends are done, close the gRPC bidirectional stream.
    await writer.close()

    print(
        f"Appended object {object_name} created of size {writer.persisted_size} bytes."
    )

// The appendable upload API is unstable. Build with
// `RUSTFLAGS="--cfg google_cloud_unstable_storage_bidi"` to enable it.
use bytes::Bytes;
use google_cloud_storage::client::Storage;

pub async fn sample(client: &Storage, bucket: &str, object: &str) -> Result<(), anyhow::Error> {
    let mut writer = client
        .open_appendable_object(format!("projects/_/buckets/{bucket}"), object)
        .send()
        .await?;

    writer.append(Bytes::from("Some data\n")).await?;

    // Flush the buffered data to the service. After a flush, the data is visible to readers.
    let persisted_size = writer.flush().await?;
    println!("Flush completed. Persisted size is now {persisted_size}");

    // The writer is still open. We can write more data.
    writer.append(Bytes::from("Some more data\n")).await?;

    // Close the writer without finalizing the object.
    let persisted_size = writer.close().await?;
    println!("Created appendable object {object} in bucket {bucket} with {persisted_size} bytes");
    Ok(())
}

namespace gcs = google::cloud::storage;

// Helper coroutine to count newlines returned by an AsyncReader.
// This helps consume the data from the read operation.
auto count_newlines =
    [](gcs::AsyncReader reader,
       gcs::AsyncToken token) -> google::cloud::future<std::uint64_t> {
  std::uint64_t count = 0;
  while (token.valid()) {
    auto [payload, t] = (co_await reader.Read(std::move(token))).value();
    token = std::move(t);
    for (auto const& buffer : payload.contents()) {
      count += std::count(buffer.begin(), buffer.end(), '\n');
    }
  }
  co_return count;
};

auto coro =
    [&count_newlines](
        gcs::AsyncClient& client, std::string bucket_name,
        std::string object_name) -> google::cloud::future<std::uint64_t> {
  auto descriptor =
      (co_await client.Open(gcs::BucketName(std::move(bucket_name)),
                            std::move(object_name)))
          .value();

  auto [reader, token] = descriptor.Read(0, 1024);

  co_return co_await count_newlines(std::move(reader), std::move(token));
};

namespace gcs = google::cloud::storage;

// Helper coroutine to count newlines returned by an AsyncReader.
// This helps consume the data from the read operation.
auto count_newlines =
    [](gcs::AsyncReader reader,
       gcs::AsyncToken token) -> google::cloud::future<std::uint64_t> {
  std::uint64_t count = 0;
  while (token.valid()) {
    auto [payload, t] = (co_await reader.Read(std::move(token))).value();
    token = std::move(t);
    for (auto const& buffer : payload.contents()) {
      count += std::count(buffer.begin(), buffer.end(), '\n');
    }
  }
  co_return count;
};

auto coro =
    [&count_newlines](
        gcs::AsyncClient& client, std::string bucket_name,
        std::string object_name) -> google::cloud::future<std::uint64_t> {
  auto descriptor =
      (co_await client.Open(gcs::BucketName(std::move(bucket_name)),
                            std::move(object_name)))
          .value();

  auto [reader, token] = descriptor.ReadFromOffset(0);

  co_return co_await count_newlines(std::move(reader), std::move(token));
};

namespace gcs = google::cloud::storage;

// Helper coroutine to count newlines returned by an AsyncReader.
auto count_newlines =
    [](gcs::AsyncReader reader,
       gcs::AsyncToken token) -> google::cloud::future<std::uint64_t> {
  std::uint64_t count = 0;
  while (token.valid()) {
    auto [payload, t] = (co_await reader.Read(std::move(token))).value();
    token = std::move(t);
    for (auto const& buffer : payload.contents()) {
      count += std::count(buffer.begin(), buffer.end(), '\n');
    }
  }
  co_return count;
};

auto coro =
    [&count_newlines](
        gcs::AsyncClient& client, std::string bucket_name,
        std::string object_name) -> google::cloud::future<std::uint64_t> {
  auto descriptor =
      (co_await client.Open(gcs::BucketName(std::move(bucket_name)),
                            std::move(object_name)))
          .value();

  auto [r1, t1] = descriptor.Read(0, 1024);
  auto [r2, t2] = descriptor.Read(0, 1024);
  auto [r3, t3] = descriptor.Read(1024, 1024);

  auto c1 = count_newlines(std::move(r1), std::move(t1));
  auto c2 = count_newlines(std::move(r2), std::move(t2));
  auto c3 = count_newlines(std::move(r3), std::move(t3));
  co_return (co_await std::move(c1)) + (co_await std::move(c2)) +
      (co_await std::move(c3));
};

namespace gcs = google::cloud::storage;

// Helper coroutine to count newlines returned by an AsyncReader.
auto count_newlines =
    [](gcs::AsyncReader reader,
       gcs::AsyncToken token) -> google::cloud::future<std::uint64_t> {
  std::uint64_t count = 0;
  while (token.valid()) {
    auto [payload, t] = (co_await reader.Read(std::move(token))).value();
    token = std::move(t);
    for (auto const& buffer : payload.contents()) {
      count += std::count(buffer.begin(), buffer.end(), '\n');
    }
  }
  co_return count;
};

auto coro = [&count_newlines](
                gcs::AsyncClient& client, std::string bucket_name,
                std::string object_name1, std::string object_name2,
                std::string object_name3) -> google::cloud::future<void> {
  // List of object names to read (passed as arguments)
  std::vector<std::string> object_names = {object_name1, object_name2,
                                           object_name3};
  std::vector<google::cloud::future<std::uint64_t>> futures;

  // Start ranged reads for all objects and collect futures
  // This example opens multiple objects, not one object multiple times.
  for (auto const& name : object_names) {
    auto descriptor =
        (co_await client.Open(gcs::BucketName(bucket_name), name)).value();
    auto [reader, token] = descriptor.Read(0, 1024);
    futures.push_back(count_newlines(std::move(reader), std::move(token)));
  }

  // Process futures as they become ready and print results
  while (!futures.empty()) {
    bool progress_made = false;
    for (std::size_t i = 0; i < futures.size(); ++i) {
      if (futures[i].is_ready()) {  // Check if the future is ready
        auto count = futures[i].get();
        std::cout << "Object " << object_names[i] << " read returned "
                  << count << " newlines\n";
        futures.erase(futures.begin() + i);
        object_names.erase(object_names.begin() + i);
        progress_made = true;
        break;  // Restart the loop after modifying the vectors
      }
    }
    if (!progress_made) {
      std::this_thread::sleep_for(
          std::chrono::milliseconds(10));  // Avoid busy spin
    }
  }
};

import (
	"bytes"
	"context"
	"fmt"
	"io"
	"time"

	"cloud.google.com/go/storage"
)

// openObjectSingleRangedRead reads a single range from an object.
func openObjectSingleRangedRead(w io.Writer, bucket, object string) ([]byte, error) {
	// bucket := "bucket-name"
	// object := "object-name"
	ctx := context.Background()
	client, err := storage.NewGRPCClient(ctx, storage.WithGRPCBidiReads())
	if err != nil {
		return nil, fmt.Errorf("storage.NewGRPCClient: %w", err)
	}
	defer client.Close()

	ctx, cancel := context.WithTimeout(ctx, time.Second*10)
	defer cancel()

	// Read the first KiB of the file and copy into a buffer.
	r, err := client.Bucket(bucket).Object(object).NewRangeReader(ctx, 0, 1024)
	if err != nil {
		return nil, fmt.Errorf("NewRangeReader: %w", err)
	}
	defer r.Close()
	buf := new(bytes.Buffer)
	if _, err := io.Copy(buf, r); err != nil {
		return nil, fmt.Errorf("copying data: %w", err)
	}

	fmt.Fprintf(w, "Read the first 1024 bytes of %v into a buffer\n", object)

	return buf.Bytes(), nil
}

import (
	"bytes"
	"context"
	"fmt"
	"io"
	"time"

	"cloud.google.com/go/storage"
)

// openObjectReadFullObject reads a full object's data.
func openObjectReadFullObject(w io.Writer, bucket, object string) ([]byte, error) {
	// bucket := "bucket-name"
	// object := "object-name"
	ctx := context.Background()
	client, err := storage.NewGRPCClient(ctx, storage.WithGRPCBidiReads())
	if err != nil {
		return nil, fmt.Errorf("storage.NewGRPCClient: %w", err)
	}
	defer client.Close()

	ctx, cancel := context.WithTimeout(ctx, time.Second*10)
	defer cancel()

	// Read the full object and copy it into a buffer.
	r, err := client.Bucket(bucket).Object(object).NewReader(ctx)
	if err != nil {
		return nil, fmt.Errorf("NewReader: %w", err)
	}
	defer r.Close()
	buf := new(bytes.Buffer)
	if _, err := io.Copy(buf, r); err != nil {
		return nil, fmt.Errorf("copying data: %w", err)
	}

	fmt.Fprintf(w, "Read the data of %v into a buffer\n", object)

	return buf.Bytes(), nil
}

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"time"

	"cloud.google.com/go/storage"
)

// openObjectMultipleRangedRead opens a single object using
// MultiRangeDownloader to download multiple ranges.
func openObjectMultipleRangedRead(w io.Writer, bucket, object string) ([][]byte, error) {
	// bucket := "bucket-name"
	// object := "object-name"
	ctx := context.Background()
	client, err := storage.NewGRPCClient(ctx, storage.WithGRPCBidiReads())
	if err != nil {
		return nil, fmt.Errorf("storage.NewGRPCClient: %w", err)
	}
	defer client.Close()

	ctx, cancel := context.WithTimeout(ctx, time.Second*10)
	defer cancel()

	// Create the MultiRangeDownloader, which opens a stream to the object.
	mrd, err := client.Bucket(bucket).Object(object).NewMultiRangeDownloader(ctx)
	if err != nil {
		return nil, fmt.Errorf("NewMultiRangeDownloader: %w", err)
	}

	// Add some 1 KiB ranges to download. This call is non-blocking. The
	// provided callback is invoked when the range download is complete.
	// Each range writes to its own buffer and its own error slot, so the
	// callbacks can run concurrently without any further synchronization.
	startOffsets := []int64{0, 1024, 2048}
	var dataBufs [3]bytes.Buffer
	var errs [3]error
	for i, off := range startOffsets {
		mrd.Add(&dataBufs[i], off, 1024, func(off, length int64, err error) {
			errs[i] = err
		})
	}

	// Wait for all downloads to complete.
	mrd.Wait()
	if err := errors.Join(errs[:]...); err != nil {
		return nil, fmt.Errorf("one or more downloads failed: %w", err)
	}
	if err := mrd.Close(); err != nil {
		return nil, fmt.Errorf("MultiRangeDownloader.Close: %w", err)
	}

	fmt.Fprintf(w, "Read the ranges of %v into memory\n", object)

	// Collect the byte slices
	var byteSlices [][]byte
	for i := range dataBufs {
		byteSlices = append(byteSlices, dataBufs[i].Bytes())
	}

	return byteSlices, nil
}

import com.google.api.core.ApiFuture;
import com.google.cloud.storage.BlobId;
import com.google.cloud.storage.BlobReadSession;
import com.google.cloud.storage.RangeSpec;
import com.google.cloud.storage.ReadProjectionConfigs;
import com.google.cloud.storage.Storage;
import com.google.cloud.storage.StorageOptions;
import java.util.concurrent.TimeUnit;

public class OpenObjectSingleRangedRead {
  public static void openObjectSingleRangedRead(
      String bucketName, String objectName, long offset, int length) throws Exception {
    // The ID of your GCS bucket
    // String bucketName = "your-unique-bucket-name";

    // The ID of your GCS object
    // String objectName = "your-object-name";

    // The beginning of the range
    // long offset = 0

    // The maximum number of bytes to read from the object.
    // int length = 64;

    try (Storage storage = StorageOptions.grpc().build().getService()) {
      BlobId blobId = BlobId.of(bucketName, objectName);
      ApiFuture<BlobReadSession> futureBlobReadSession = storage.blobReadSession(blobId);

      try (BlobReadSession blobReadSession = futureBlobReadSession.get(10, TimeUnit.SECONDS)) {
        // Define the range of bytes to read.
        RangeSpec rangeSpec = RangeSpec.of(offset, length);
        ApiFuture<byte[]> future =
            blobReadSession.readAs(ReadProjectionConfigs.asFutureBytes().withRangeSpec(rangeSpec));

        // Wait for the read to complete.
        byte[] bytes = future.get();

        System.out.println(
            "Successfully read "
                + bytes.length
                + " bytes from object "
                + objectName
                + " in bucket "
                + bucketName);
      }
    }
  }
}

import com.google.api.core.ApiFuture;
import com.google.cloud.storage.BlobId;
import com.google.cloud.storage.BlobReadSession;
import com.google.cloud.storage.ReadAsChannel;
import com.google.cloud.storage.ReadProjectionConfigs;
import com.google.cloud.storage.Storage;
import com.google.cloud.storage.StorageOptions;
import java.nio.ByteBuffer;
import java.nio.channels.ScatteringByteChannel;
import java.util.Locale;
import java.util.concurrent.TimeUnit;

public class OpenObjectReadFullObject {
  public static void openObjectReadFullObject(String bucketName, String objectName)
      throws Exception {
    // The ID of your GCS bucket
    // String bucketName = "your-unique-bucket-name";

    // The ID of your GCS object to read
    // String objectName = "your-object-name";

    try (Storage storage = StorageOptions.grpc().build().getService()) {
      BlobId blobId = BlobId.of(bucketName, objectName);
      ApiFuture<BlobReadSession> futureBlobReadSession = storage.blobReadSession(blobId);

      try (BlobReadSession blobReadSession = futureBlobReadSession.get(10, TimeUnit.SECONDS)) {

        ReadAsChannel readAsChannelConfig = ReadProjectionConfigs.asChannel();
        try (ScatteringByteChannel channel = blobReadSession.readAs(readAsChannelConfig)) {
          long totalBytesRead = 0;
          ByteBuffer buffer = ByteBuffer.allocate(64 * 1024);
          int bytesRead;

          while ((bytesRead = channel.read(buffer)) != -1) {
            totalBytesRead += bytesRead;
            buffer.clear();
          }

          System.out.printf(
              Locale.US,
              "Successfully read a total of %d bytes from object %s%n",
              totalBytesRead,
              blobId.toGsUtilUri());
        }
      }
    }
  }
}

import com.google.api.core.ApiFuture;
import com.google.api.core.ApiFutures;
import com.google.cloud.storage.BlobId;
import com.google.cloud.storage.BlobReadSession;
import com.google.cloud.storage.RangeSpec;
import com.google.cloud.storage.ReadProjectionConfigs;
import com.google.cloud.storage.Storage;
import com.google.cloud.storage.StorageOptions;
import com.google.common.collect.ImmutableList;
import java.util.List;
import java.util.concurrent.TimeUnit;

public class OpenObjectMultipleRangedRead {
  public static void openObjectMultipleRangedRead(
      String bucketName, String objectName, long offset1, int length1, long offset2, int length2)
      throws Exception {
    // The ID of your GCS bucket
    // String bucketName = "your-unique-bucket-name";

    // The ID of your GCS object
    // String objectName = "your-object-name";

    // The beginning of the range 1
    // long offset = 0

    // The maximum number of bytes to read in range 1
    // int length = 16;

    // The beginning of the range 2
    // long offset = 16

    // The maximum number of bytes to read in range 2
    // int length = 32;

    try (Storage storage = StorageOptions.grpc().build().getService()) {
      BlobId blobId = BlobId.of(bucketName, objectName);
      ApiFuture<BlobReadSession> futureBlobReadSession = storage.blobReadSession(blobId);
      RangeSpec rangeSpec1 = RangeSpec.of(offset1, length1);
      RangeSpec rangeSpec2 = RangeSpec.of(offset2, length2);

      try (BlobReadSession blobReadSession = futureBlobReadSession.get(10, TimeUnit.SECONDS)) {
        ApiFuture<byte[]> future1 =
            blobReadSession.readAs(ReadProjectionConfigs.asFutureBytes().withRangeSpec(rangeSpec1));
        ApiFuture<byte[]> future2 =
            blobReadSession.readAs(ReadProjectionConfigs.asFutureBytes().withRangeSpec(rangeSpec2));

        List<byte[]> allBytes = ApiFutures.allAsList(ImmutableList.of(future1, future2)).get();

        byte[] bytes1 = allBytes.get(0);
        byte[] bytes2 = allBytes.get(1);

        System.out.println(
            "Successfully read "
                + bytes1.length
                + " bytes from range 1 and "
                + bytes2.length
                + " bytes from range 2.");
      }
    }
  }
}

import com.google.api.core.ApiFuture;
import com.google.api.core.ApiFutures;
import com.google.cloud.storage.BlobId;
import com.google.cloud.storage.BlobReadSession;
import com.google.cloud.storage.RangeSpec;
import com.google.cloud.storage.ReadAsFutureBytes;
import com.google.cloud.storage.ReadProjectionConfigs;
import com.google.cloud.storage.Storage;
import com.google.cloud.storage.StorageOptions;
import com.google.common.util.concurrent.MoreExecutors;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

public class OpenMultipleObjectsRangedRead {
  public static void multipleObjectsSingleRangedRead(
      String bucketName, List<String> objectNames, long startOffset, int length) throws Exception {
    // The ID of your GCS bucket
    // String bucketName = "your-unique-bucket-name";

    // The ID of your GCS objects to read
    // List<String> objectName = Arrays.asList("object-1", "object-2", "object-3");

    RangeSpec singleRange = RangeSpec.of(startOffset, length);
    ReadAsFutureBytes rangeConfig =
        ReadProjectionConfigs.asFutureBytes().withRangeSpec(singleRange);

    try (Storage storage = StorageOptions.grpc().build().getService()) {
      List<ApiFuture<byte[]>> futuresToWaitOn = new ArrayList<>();

      System.out.printf(
          "Initiating single ranged read [%d, %d] on %d objects...%n",
          startOffset, startOffset + length - 1, objectNames.size());

      for (String objectName : objectNames) {
        BlobId blobId = BlobId.of(bucketName, objectName);
        ApiFuture<BlobReadSession> futureReadSession = storage.blobReadSession(blobId);

        ApiFuture<byte[]> readAndCloseFuture =
            ApiFutures.transformAsync(
                futureReadSession,
                (BlobReadSession session) -> {
                  ApiFuture<byte[]> readFuture = session.readAs(rangeConfig);

                  readFuture.addListener(
                      () -> {
                        try {
                          session.close();
                        } catch (java.io.IOException e) {
                          System.err.println(
                              "WARN: Background error while closing session: " + e.getMessage());
                        }
                      },
                      MoreExecutors.directExecutor());
                  return readFuture;
                },
                MoreExecutors.directExecutor());

        futuresToWaitOn.add(readAndCloseFuture);
      }
      ApiFutures.allAsList(futuresToWaitOn).get(30, TimeUnit.SECONDS);

      System.out.println("All concurrent single-ranged read operations are complete.");
    }
  }
}

async def storage_open_object_single_ranged_read(
    bucket_name, object_name, start_byte, size, grpc_client=None
):
    """Downloads a range of bytes from an object.

    grpc_client: an existing grpc_client to use, this is only for testing.
    """
    if grpc_client is None:
        grpc_client = AsyncGrpcClient()

    mrd = AsyncMultiRangeDownloader(grpc_client, bucket_name, object_name)

    try:
        # Open the object, mrd always opens in read mode.
        await mrd.open()

        # requested range will be downloaded into this buffer, user may provide
        # their own buffer or file-like object.
        output_buffer = BytesIO()
        await mrd.download_ranges([(start_byte, size, output_buffer)])
    finally:
        if mrd.is_stream_open:
            await mrd.close()

    # Downloaded size can differ from requested size if object is smaller.
    # mrd will download at most up to the end of the object.
    downloaded_size = output_buffer.getbuffer().nbytes
    print(f"Downloaded {downloaded_size} bytes from {object_name}")

async def storage_open_object_read_full_object(
    bucket_name, object_name, grpc_client=None
):
    """Downloads the entire content of an object using a multi-range downloader.

    grpc_client: an existing grpc_client to use, this is only for testing.
    """
    if grpc_client is None:
        grpc_client = AsyncGrpcClient()

    # mrd = Multi-Range-Downloader
    mrd = AsyncMultiRangeDownloader(grpc_client, bucket_name, object_name)

    try:
        # Open the object, mrd always opens in read mode.
        await mrd.open()

        # This could be any buffer or file-like object.
        output_buffer = BytesIO()
        # A download range of (0, 0) means to read from the beginning to the end.
        await mrd.download_ranges([(0, 0, output_buffer)])
    finally:
        if mrd.is_stream_open:
            await mrd.close()

    downloaded_bytes = output_buffer.getvalue()
    print(
        f"Downloaded all {len(downloaded_bytes)} bytes from object {object_name} in bucket {bucket_name}."
    )

async def storage_open_object_multiple_ranged_read(
    bucket_name, object_name, grpc_client=None
):
    """Downloads multiple ranges of bytes from a single object into different buffers.

    grpc_client: an existing grpc_client to use, this is only for testing.
    """
    if grpc_client is None:
        grpc_client = AsyncGrpcClient()

    mrd = AsyncMultiRangeDownloader(grpc_client, bucket_name, object_name)

    try:
        # Open the object, mrd always opens in read mode.
        await mrd.open()

        # Specify four different buffers to download ranges into.
        buffers = [BytesIO(), BytesIO(), BytesIO(), BytesIO()]

        # Define the ranges to download. Each range is a tuple of (start_byte, size, buffer).
        # All ranges will download 10 bytes from different starting positions.
        # We choose arbitrary start bytes for this example. An object should be large enough.
        # A user can choose any start byte between 0 and `object_size`.
        # If `start_bytes` is greater than `object_size`, mrd will throw an error.
        ranges = [
            (0, 10, buffers[0]),
            (20, 10, buffers[1]),
            (40, 10, buffers[2]),
            (60, 10, buffers[3]),
        ]

        await mrd.download_ranges(ranges)

    finally:
        await mrd.close()

    # Print the downloaded content from each buffer.
    for i, output_buffer in enumerate(buffers):
        downloaded_size = output_buffer.getbuffer().nbytes
        print(
            f"Downloaded {downloaded_size} bytes into buffer {i + 1} from start byte {ranges[i][0]}: {output_buffer.getvalue()}"
        )

async def storage_open_multiple_objects_ranged_read(
    bucket_name, object_names, grpc_client=None
):
    """Downloads a range of bytes from multiple objects concurrently.

    grpc_client: an existing grpc_client to use, this is only for testing.
    """
    if grpc_client is None:
        grpc_client = AsyncGrpcClient()

    async def _download_range(object_name):
        """Helper coroutine to download a range from a single object."""
        mrd = AsyncMultiRangeDownloader(grpc_client, bucket_name, object_name)
        try:
            # Open the object, mrd always opens in read mode.
            await mrd.open()

            # Each object downloads the first 100 bytes.
            start_byte = 0
            size = 100

            # requested range will be downloaded into this buffer, user may provide
            # their own buffer or file-like object.
            output_buffer = BytesIO()
            await mrd.download_ranges([(start_byte, size, output_buffer)])
        finally:
            if mrd.is_stream_open:
                await mrd.close()

        # Downloaded size can differ from requested size if object is smaller.
        # mrd will download at most up to the end of the object.
        downloaded_size = output_buffer.getbuffer().nbytes
        print(f"Downloaded {downloaded_size} bytes from {object_name}")

    download_tasks = [_download_range(name) for name in object_names]
    await asyncio.gather(*download_tasks)

use google_cloud_storage::client::Storage;
use google_cloud_storage::model_ext::ReadRange;

pub async fn sample(
    client: &Storage,
    bucket: &str,
    object: &str,
    start: u64,
    end: u64,
    destination_file_name: &str,
) -> anyhow::Result<()> {
    let count = end - start + 1;
    let (metadata, mut reader) = client
        .open_object(format!("projects/_/buckets/{bucket}"), object)
        .send_and_read(ReadRange::segment(start, count))
        .await?;

    let mut content = Vec::new();
    while let Some(data) = reader.next().await.transpose()? {
        content.extend_from_slice(&data);
    }

    tokio::fs::write(destination_file_name, content).await?;

    println!(
        "Downloaded bytes from {start} to {end} of object {object} in bucket {bucket} to {destination_file_name}."
    );
    println!("Metadata: {:?}", metadata.object());
    Ok(())
}

use google_cloud_storage::client::Storage;
use google_cloud_storage::model_ext::ReadRange;

pub async fn sample(
    client: &Storage,
    bucket: &str,
    object: &str,
    destination_file_name: &str,
) -> anyhow::Result<()> {
    let (metadata, mut reader) = client
        .open_object(format!("projects/_/buckets/{bucket}"), object)
        .send_and_read(ReadRange::all())
        .await?;

    let mut content = Vec::new();
    while let Some(data) = reader.next().await.transpose()? {
        content.extend_from_slice(&data);
    }

    tokio::fs::write(destination_file_name, content).await?;

    println!("Downloaded object {object} in bucket {bucket} to {destination_file_name}.");
    println!("Metadata: {:?}", metadata.object());
    Ok(())
}

use google_cloud_storage::client::Storage;
use google_cloud_storage::model_ext::ReadRange;

pub async fn sample(client: &Storage, bucket: &str, object: &str) -> anyhow::Result<()> {
    let bucket_name = format!("projects/_/buckets/{bucket}");

    // 1. Open the connection and read the first range.
    let (descriptor, mut reader) = client
        .open_object(bucket_name, object)
        .send_and_read(ReadRange::head(5))
        .await?;

    let mut content = Vec::new();
    while let Some(data) = reader.next().await.transpose()? {
        content.extend_from_slice(&data);
    }
    println!(
        "Read first range from {object} in {bucket}. Content: {:?}",
        String::from_utf8_lossy(&content)
    );

    // 2. Read another range.
    let mut reader = descriptor.read_range(ReadRange::segment(5, 5)).await;

    let mut content = Vec::new();
    while let Some(data) = reader.next().await.transpose()? {
        content.extend_from_slice(&data);
    }
    println!(
        "Read second range from {object} in {bucket}. Content: {:?}",
        String::from_utf8_lossy(&content)
    );

    println!("Metadata: {:?}", descriptor.object());
    Ok(())
}

use google_cloud_storage::client::Storage;
use google_cloud_storage::model_ext::ReadRange;
use google_cloud_storage::object_descriptor::ObjectDescriptor;
use std::collections::HashMap;
use tokio::task::JoinSet;

pub async fn sample(client: &Storage, bucket: &str, object_names: &[&str]) -> anyhow::Result<()> {
    let bucket_name = format!("projects/_/buckets/{bucket}");

    async fn read_footer(
        client: &Storage,
        bucket_name: &str,
        object_name: &str,
    ) -> anyhow::Result<(ObjectDescriptor, Vec<u8>, String)> {
        // Reading the last 8 bytes is common in Parquet and other column-oriented formats
        // with a footer.
        let (descriptor, mut reader) = client
            .open_object(bucket_name, object_name)
            .send_and_read(ReadRange::tail(8))
            .await?;

        let mut footer = Vec::new();
        while let Some(data) = reader.next().await.transpose()? {
            footer.extend_from_slice(&data);
        }
        Ok((descriptor, footer, object_name.to_string()))
    }

    let mut set = JoinSet::new();
    for o in object_names {
        let client = client.clone();
        let bucket_name = bucket_name.clone();
        let object_name = o.to_string();
        set.spawn(async move { read_footer(&client, &bucket_name, &object_name).await });
    }

    let mut descriptors = HashMap::new();
    while let Some(result) = set.join_next().await {
        let (descriptor, footer, object_name) = result??;
        println!("The footer for object {object_name} is {footer:?}");
        descriptors.insert(object_name, (descriptor, footer));
    }

    // Use the descriptors in the map to read more ranges.

    Ok(())
}

namespace gcs = google::cloud::storage;
auto coro = [](gcs::AsyncClient& client, std::string bucket_name,
               std::string object_name)
    -> google::cloud::future<google::storage::v2::Object> {
  // Start an appendable upload and write some data.
  auto [writer, token] = (co_await client.StartAppendableObjectUpload(
                              gcs::BucketName(bucket_name), object_name))
                             .value();
  std::cout << "Appendable upload started.\n";
  token = (co_await writer.Write(std::move(token),
                                 gcs::WritePayload("paused data\n")))
              .value();

  // The writer is closed, but the upload is not finalized. This "pauses" the
  // upload, as the object remains appendable.
  auto close_status = co_await writer.Close();
  if (!close_status.ok()) throw std::runtime_error(close_status.message());
  std::cout << "Upload paused.\n";

  // To resume the upload we need the object's generation. We can use the
  // regular GCS client to get the latest metadata.
  auto regular_client = gcs::Client();
  auto metadata =
      regular_client.GetObjectMetadata(bucket_name, object_name).value();
  std::cout << "Object generation is " << metadata.generation() << "\n";

  // Now resume the upload.
  std::tie(writer, token) =
      (co_await client.ResumeAppendableObjectUpload(
           gcs::BucketName(bucket_name), object_name, metadata.generation()))
          .value();
  std::cout << "Upload resumed from offset "
            << std::get<std::int64_t>(writer.PersistedState()) << "\n";

  // Append the rest of the data.
  token = (co_await writer.Write(std::move(token),
                                 gcs::WritePayload("resumed data\n")))
              .value();

  // Finalize the upload and return the object metadata.
  co_return (co_await writer.Finalize(std::move(token))).value();
};

import (
	"context"
	"fmt"
	"io"
	"time"

	"cloud.google.com/go/storage"
)

// pauseAndResumeAppendableUpload creates a new unfinalized appendable object in
// a bucket with the Rapid storage class, closes the Writer, then re-opens the
// object for writing using NewWriterFromAppendableObject.
func pauseAndResumeAppendableUpload(w io.Writer, bucket, object string) error {
	// bucket := "bucket-name"
	// object := "object-name"
	ctx := context.Background()
	client, err := storage.NewGRPCClient(ctx, storage.WithAppendableUploads())
	if err != nil {
		return fmt.Errorf("storage.NewGRPCClient: %w", err)
	}
	defer client.Close()

	ctx, cancel := context.WithTimeout(ctx, time.Second*10)
	defer cancel()

	// Start an appendable upload and write some data.
	writer := client.Bucket(bucket).Object(object).NewWriter(ctx)

	if _, err := writer.Write([]byte("Some data\n")); err != nil {
		return fmt.Errorf("Writer.Write: %w", err)
	}

	// The writer is closed, but the upload is not finalized. This "pauses" the
	// upload, as the object remains appendable.
	if err := writer.Close(); err != nil {
		return fmt.Errorf("Writer.Close: %w", err)
	}
	fmt.Fprintf(w, "Uploaded unfinalized object %v\n", object)

	// To resume the upload we need the object's generation. We can get this
	// from the previous Writer after close.
	gen := writer.Attrs().Generation

	// Now resume the upload. Writer options including finalization can be
	// passed on calling this constructor.
	appendWriter, offset, err := client.Bucket(bucket).Object(object).Generation(gen).NewWriterFromAppendableObject(
		ctx, &storage.AppendableWriterOpts{
			FinalizeOnClose: true,
		},
	)
	if err != nil {
		return fmt.Errorf("NewWriterFromAppendableObject: %w", err)
	}
	fmt.Fprintf(w, "Resuming upload from offset %v\n", offset)

	// Append the rest of the data and close the Writer to finalize.
	if _, err := appendWriter.Write([]byte("resumed data\n")); err != nil {
		return fmt.Errorf("appendWriter.Write: %w", err)
	}
	if err := appendWriter.Close(); err != nil {
		return fmt.Errorf("Writer.Close: %w", err)
	}
	fmt.Fprintf(w, "Uploaded and finalized object %v\n", object)
	return nil
}

import com.google.cloud.storage.Blob;
import com.google.cloud.storage.BlobAppendableUpload;
import com.google.cloud.storage.BlobAppendableUpload.AppendableUploadWriteableByteChannel;
import com.google.cloud.storage.BlobAppendableUploadConfig;
import com.google.cloud.storage.BlobAppendableUploadConfig.CloseAction;
import com.google.cloud.storage.BlobId;
import com.google.cloud.storage.BlobInfo;
import com.google.cloud.storage.Storage;
import com.google.cloud.storage.StorageChannelUtils;
import com.google.cloud.storage.StorageOptions;
import com.google.common.io.ByteStreams;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.charset.StandardCharsets;
import java.nio.file.Paths;
import java.util.Locale;

public class PauseAndResumeAppendableObjectUpload {
  public static void pauseAndResumeAppendableObjectUpload(
      String bucketName, String objectName, String filePath) throws Exception {
    // The ID of your GCS bucket
    // String bucketName = "your-unique-bucket-name";

    // The ID of your GCS object
    // String objectName = "your-object-name";

    // The path to the file to upload
    // String filePath = "path/to/your/file";

    try (Storage storage = StorageOptions.grpc().build().getService()) {
      BlobId blobId = BlobId.of(bucketName, objectName);
      BlobInfo blobInfo = BlobInfo.newBuilder(blobId).build();

      // --- Step 1: Initial string write (PAUSE) ---
      // Default close action will be CLOSE_WITHOUT_FINALIZING
      BlobAppendableUploadConfig initialConfig = BlobAppendableUploadConfig.of();
      BlobAppendableUpload initialUploadSession =
          storage.blobAppendableUpload(blobInfo, initialConfig);

      try (AppendableUploadWriteableByteChannel channel = initialUploadSession.open()) {
        String initialData = "Initial data segment.\n";
        ByteBuffer buffer = ByteBuffer.wrap(initialData.getBytes(StandardCharsets.UTF_8));
        long totalBytesWritten = StorageChannelUtils.blockingEmptyTo(buffer, channel);
        channel.flush();

        System.out.printf(
            Locale.US, "Wrote %d bytes (initial string) in first segment.\n", totalBytesWritten);
      } catch (IOException ex) {
        throw new IOException("Failed initial upload to object " + blobId.toGsUtilUri(), ex);
      }

      Blob existingBlob = storage.get(blobId);
      long currentObjectSize = existingBlob.getSize();
      System.out.printf(
          Locale.US,
          "Initial upload paused. Currently uploaded size: %d bytes\n",
          currentObjectSize);

      // --- Step 2: Resume upload with file content and finalize ---
      // Use FINALIZE_WHEN_CLOSING to ensure the object is finalized on channel closure.
      BlobAppendableUploadConfig resumeConfig =
          BlobAppendableUploadConfig.of().withCloseAction(CloseAction.FINALIZE_WHEN_CLOSING);
      BlobAppendableUpload resumeUploadSession =
          storage.blobAppendableUpload(existingBlob.toBuilder().build(), resumeConfig);

      try (FileChannel fileChannel = FileChannel.open(Paths.get(filePath));
          AppendableUploadWriteableByteChannel channel = resumeUploadSession.open()) {
        long bytesToAppend = fileChannel.size();
        System.out.printf(
            Locale.US,
            "Appending the entire file (%d bytes) after the initial string.\n",
            bytesToAppend);

        ByteStreams.copy(fileChannel, channel);
      }

      BlobInfo result = storage.get(blobId);
      System.out.printf(
          Locale.US,
          "\nObject %s successfully resumed and finalized. Total size: %d bytes\n",
          result.getBlobId().toGsUtilUriWithGeneration(),
          result.getSize());
    }
  }
}