feat(storage)!: return stream of ByteBuffer

packages/gax/Sources/GoogleCloudGax/HTTPClientResponse.swift

@@ -43,6 +43,10 @@ import struct NIOCore.ByteBuffer
     self.response.headers
   }
 
+  public var body: _HTTPResponseBody {
+    _HTTPResponseBody(self.response.body)
+  }
+
   public func data(upTo: Int) async throws -> Data {
     let buffer = try await self.response.body.collect(upTo: upTo)
     return Data(buffer: buffer)

packages/gax/Sources/GoogleCloudGax/HTTPResponseBody.swift

@@ -0,0 +1,43 @@
+// Copyright 2026 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
+//
+//     https://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.
+
+import struct AsyncHTTPClient.HTTPClientResponse
+import struct NIOCore.ByteBuffer
+
+/// An asynchronous sequence of response body chunks.
+@_spi(GoogleCloudInternal) public struct _HTTPResponseBody: AsyncSequence, Sendable {
+  public typealias Element = NIOCore.ByteBuffer
+
+  let body: AsyncHTTPClient.HTTPClientResponse.Body
+
+  public init(_ body: AsyncHTTPClient.HTTPClientResponse.Body) {
+    self.body = body
+  }
+
+  public struct AsyncIterator: AsyncIteratorProtocol {
+    var iterator: AsyncHTTPClient.HTTPClientResponse.Body.AsyncIterator
+
+    public init(_ iterator: AsyncHTTPClient.HTTPClientResponse.Body.AsyncIterator) {
+      self.iterator = iterator
+    }
+
+    public mutating func next() async throws -> NIOCore.ByteBuffer? {
+      try await self.iterator.next()
+    }
+  }
+
+  public func makeAsyncIterator() -> AsyncIterator {
+    AsyncIterator(self.body.makeAsyncIterator())
+  }
+}

packages/storage/Package.swift

@@ -84,7 +84,10 @@ let package = Package(
     .testTarget(
       name: "GoogleCloudStorageIntegrationTests",
       dependencies: [
-        "GoogleCloudStorage"
+        "GoogleCloudStorage",
+        .product(name: "GoogleCloudAuth", package: "auth"),
+        .product(name: "GoogleCloudGax", package: "gax"),
+        .product(name: "NIOCore", package: "swift-nio"),
       ],
       path: "Tests/IntegrationTests"
     ),

packages/storage/Sources/GoogleCloudStorage/DownloadOptions.swift

@@ -13,6 +13,8 @@
 // limitations under the License.
 
 import Foundation
+@_spi(GoogleCloudInternal) import GoogleCloudGax
+import struct NIOCore.ByteBuffer
 
 /// Specifies a byte range for ranged reads.
 public enum ReadObjectRange: Sendable, Hashable, Equatable {
@@ -168,7 +170,7 @@ public struct ReadObjectOptions: Sendable {
   /// Flag to enable automatic decompressive transcoding by GCS. Defaults to `true`.
   public var enableDecompressiveTranscoding: Bool = true
 
-  /// Configuration options for download checksum validation.
+  /// Checksum options for validating data integrity.
   public var checksums: ChecksumOptions = .default
 
   /// Flag to enable transparent auto-resumption on transient network failures. Defaults to `true`.
@@ -240,9 +242,9 @@ public struct ReadObjectMetadata: Sendable, Hashable, Equatable {
   }
 }
 
-/// An asynchronous sequence of `Data` chunks representing an object payload being downloaded.
+/// An asynchronous sequence of `ByteBuffer` chunks representing an object payload being downloaded.
 public struct ReadObjectSequence: AsyncSequence, Sendable {
-  public typealias Element = Data
+  public typealias Element = NIOCore.ByteBuffer
 
   /// Name of the bucket containing the object being read.
   public var bucket: String = ""
@@ -253,7 +255,11 @@ public struct ReadObjectSequence: AsyncSequence, Sendable {
   /// Configuration options used for this object download.
   public var options: ReadObjectOptions = .init()
 
-  internal var stream: AsyncThrowingStream<Data, Error> = AsyncThrowingStream { $0.finish() }
+  /// Object metadata extracted from initial HTTP response headers.
+  public var metadata: ReadObjectMetadata = .init()
+
+  package var initialBody: _HTTPResponseBody?
+  package var stream: AsyncThrowingStream<NIOCore.ByteBuffer, Error>?
 
   /// Creates a new `ReadObjectSequence` instance.
   public init() {}
@@ -267,27 +273,77 @@ public struct ReadObjectSequence: AsyncSequence, Sendable {
 
   /// An asynchronous iterator for iterating over chunks of downloaded object payload data.
   public struct AsyncIterator: AsyncIteratorProtocol, Sendable {
-    public typealias Element = Data
-
-    private struct Storage: @unchecked Sendable {
-      var iterator: AsyncThrowingStream<Data, Error>.AsyncIterator
+    public typealias Element = NIOCore.ByteBuffer
+
+    package final class Storage: @unchecked Sendable {
+      let options: ReadObjectOptions
+      var bodyIterator: _HTTPResponseBody.AsyncIterator?
+      var streamIterator: AsyncThrowingStream<NIOCore.ByteBuffer, Error>.AsyncIterator?
+      var isFinished: Bool = false
+
+      init(
+        options: ReadObjectOptions,
+        initialBody: _HTTPResponseBody?,
+        stream: AsyncThrowingStream<NIOCore.ByteBuffer, Error>?
+      ) {
+        self.options = options
+        self.bodyIterator = initialBody?.makeAsyncIterator()
+        self.streamIterator = stream?.makeAsyncIterator()
+      }
+
+      func next() async throws -> NIOCore.ByteBuffer? {
+        guard !isFinished else { return nil }
+
+        if case .prefix(0) = options.range {
+          isFinished = true
+          return nil
+        }
+        if case .suffix(0) = options.range {
+          isFinished = true
+          return nil
+        }
+
+        if var it = streamIterator {
+          let chunk = try await it.next()
+          self.streamIterator = it
+          if chunk == nil {
+            isFinished = true
+          }
+          return chunk
+        } else if var it = bodyIterator {
+          let chunk = try await it.next()
+          self.bodyIterator = it
+          if chunk == nil {
+            isFinished = true
+          }
+          return chunk
+        } else {
+          isFinished = true
+          return nil
+        }
+      }
     }
 
-    private var storage: Storage
+    package var storage: Storage
 
-    internal init(iterator: AsyncThrowingStream<Data, Error>.AsyncIterator) {
-      self.storage = Storage(iterator: iterator)
+    package init(storage: Storage) {
+      self.storage = storage
     }
 
-    /// Advances to the next `Data` chunk in the downloaded object payload stream.
-    public mutating func next() async throws -> Data? {
-      try await storage.iterator.next()
+    /// Advances to the next `ByteBuffer` chunk in the downloaded object payload stream.
+    public mutating func next() async throws -> NIOCore.ByteBuffer? {
+      try await storage.next()
     }
   }
 
   /// Creates an asynchronous iterator for iterating over object payload chunks.
   public func makeAsyncIterator() -> AsyncIterator {
-    AsyncIterator(iterator: stream.makeAsyncIterator())
+    let storage = AsyncIterator.Storage(
+      options: options,
+      initialBody: initialBody,
+      stream: stream
+    )
+    return AsyncIterator(storage: storage)
   }
 }
 

packages/storage/Sources/GoogleCloudStorage/StorageClient+Download.swift

@@ -40,9 +40,9 @@ extension StorageClient {
       bucket: bucket, object: object, options: options)
     let response = try await request.execute()
     let statusCode = Int(response.status.code)
-    let data = try await response.data()
 
     guard (200..<300).contains(statusCode) else {
+      let data = try await response.data()
       let message = String(data: data, encoding: .utf8) ?? ""
       throw DownloadError.unexpectedServerResponse(
         statusCode: statusCode, message: message)
@@ -51,22 +51,12 @@ extension StorageClient {
     let metadata = try Self.parseReadObjectMetadata(
       from: response.headers, bucket: bucket, object: object)
 
-    let stream = AsyncThrowingStream<Data, Error> { continuation in
-      if case .prefix(0) = options.range {
-        // Requested 0 bytes
-      } else if case .suffix(0) = options.range {
-        // Requested 0 bytes
-      } else if !data.isEmpty {
-        continuation.yield(data)
-      }
-      continuation.finish()
-    }
-
     let sequence = ReadObjectSequence().with {
       $0.bucket = bucket
       $0.object = object
       $0.options = options
-      $0.stream = stream
+      $0.metadata = metadata
+      $0.initialBody = response.body
     }
     return ReadObjectResult().with {
       $0.metadata = metadata
@@ -169,7 +159,7 @@ extension StorageClient {
 }
 
 extension GoogleCloudGax._HTTPClient {
-  fileprivate func buildReadObjectRequest(
+  package func buildReadObjectRequest(
     bucket: String,
     object: String,
     options: ReadObjectOptions

packages/storage/Sources/GoogleCloudStorage/StorageClientOptions.swift

@@ -22,12 +22,17 @@ public struct StorageClientOptions: Sendable {
   /// Default configuration inherited by data-plane operations (e.g., Uploads).
   public var upload: UploadOptions
 
+  /// Default configuration inherited by data-plane download operations (e.g., ReadObject).
+  public var download: ReadObjectOptions
+
   public init(
     client: GoogleCloudGax.ClientOptions = .init(),
-    upload: UploadOptions = .default
+    upload: UploadOptions = .default,
+    download: ReadObjectOptions = .default
   ) {
     self.client = client
     self.upload = upload
+    self.download = download
   }
 
   /// Override specific values using closure modification.

packages/storage/Tests/DownloadOptionsTests.swift

@@ -35,8 +35,8 @@ import Testing
     #expect(defaultOptions.customerEncryptionKey == nil)
     #expect(defaultOptions.range == .entire)
     #expect(defaultOptions.enableDecompressiveTranscoding == true)
-    #expect(defaultOptions.checksums == .default)
     #expect(defaultOptions.autoResume == true)
+    #expect(defaultOptions.checksums == .default)
   }
 
   @Test func readObjectOptionsWithBuilder() throws {
@@ -50,17 +50,17 @@ import Testing
       $0.customerEncryptionKey = csek
       $0.range = .bounded(start: 0, end: 1024)
       $0.enableDecompressiveTranscoding = false
-      $0.checksums = .none
       $0.autoResume = false
+      $0.checksums = .none
     }
 
     #expect(options.generation == 456)
     #expect(options.preconditions?.ifGenerationMatch == 123)
     #expect(options.customerEncryptionKey == csek)
     #expect(options.range == .bounded(start: 0, end: 1024))
     #expect(options.enableDecompressiveTranscoding == false)
-    #expect(options.checksums == .none)
     #expect(options.autoResume == false)
+    #expect(options.checksums == .none)
   }
 
   @Test func readObjectMetadataProperties() {
@@ -101,16 +101,13 @@ import Testing
       $0.bucket = "bkt"
       $0.object = "obj"
     }
-    let options = ReadObjectOptions().with { $0.autoResume = false }
     let sequence = ReadObjectSequence().with {
       $0.bucket = "bkt"
       $0.object = "obj"
-      $0.options = options
     }
 
     #expect(sequence.bucket == "bkt")
     #expect(sequence.object == "obj")
-    #expect(sequence.options.autoResume == false)
 
     var iterator = sequence.makeAsyncIterator()
     let firstChunk = try await iterator.next()