feat(pubsub): Add exponential backoff, batching, and parallel streaming pull

- Exponential Backoff: Configurable retry parameters for transient network failures.
- Background Batching: Automatic batching for publish, acknowledge, and modifyAckDeadline.
- Immediate RPCs: acknowledgeNow and modifyAckDeadlineNow for explicit awaitable RPCs.
- Parallel Streaming Pull: Added maxConcurrentStreams to streamingPull to increase throughput.
- Added comprehensive unit tests and documentation examples.

TAG=agy
CONV=bc7c6164-e9cb-4a0f-b326-7752c814e1ce

Author: sigurdm

pkgs/google_cloud_pubsub/README.md

@@ -30,15 +30,30 @@ void main() async {
   // Application Default Credentials (ADC).
   final pubSub = PubSub(projectId: 'your-project-id');
 
-  // Create a topic.
-  final topic = await pubSub.topic('put-your-topic-name-here').create();
+  // Create a topic (with optional batching and retry settings).
+  final topic = await pubSub
+      .topic(
+        'put-your-topic-name-here',
+        publishSettings: const PublishSettings(
+          batching: BatchingSettings(
+            maxMessages: 100,
+            maxDelay: Duration(milliseconds: 10),
+          ),
+        ),
+      )
+      .create();
 
   // Create a subscription to that topic.
   final subscription = await pubSub
-      .subscription('put-your-subscription-name-here')
+      .subscription(
+        'put-your-subscription-name-here',
+        ackSettings: const AckSettings(
+          batching: BatchingSettings(maxDelay: Duration(milliseconds: 50)),
+        ),
+      )
       .create(topic: topic.name);
 
-  // Publish a message.
+  // Publish a message. This is automatically batched and retried.
   await topic.publish(utf8.encode('message 1'));
 
   // Pull messages from the subscription.
@@ -47,16 +62,19 @@ void main() async {
   for (final receivedMessage in messages) {
     print('Received message: ${utf8.decode(receivedMessage.data)}');
 
-    // Acknowledge the message.
-    await subscription.acknowledgeNow([receivedMessage]);
+    // Acknowledge the message in the background.
+    subscription.acknowledge(receivedMessage);
   }
 
   print(
     'Your topic is available at:\n'
     'https://pubsub.googleapis.com/v1/${topic.name}',
   );
 
-  // Clean up.
+  // Clean up and flush any pending batches.
+  subscription.close();
+  topic.close();
+
   await subscription.delete();
   await topic.delete();
 

pkgs/google_cloud_pubsub/example/doc_examples.dart

@@ -0,0 +1,74 @@
+// 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 'dart:async';
+import 'package:google_cloud_pubsub/google_cloud_pubsub.dart';
+
+Future<void> streamingPullExample(PubSub pubsub) async {
+  // #docregion streaming_pull_example
+  final subscription = pubsub.subscription('my-subscription');
+
+  // Establish a streaming pull with 3 parallel streams for high throughput
+  // and custom retry settings for handling transient connection drops.
+  final stream = subscription.streamingPull(
+    maxConcurrentStreams: 3,
+    streamAckDeadlineSeconds: 30,
+    retry: const RetrySettings(
+      maxRetries: 10,
+      initialDelay: Duration(seconds: 1),
+      maxDelay: Duration(seconds: 30),
+    ),
+  );
+
+  final listener = stream.listen(
+    (ReceivedMessage message) {
+      print('Received message: ${message.message.data}');
+
+      // Acknowledge the message. This buffers the ACK in the background
+      // and batches it with others, sending it over one of the active
+      // gRPC streams or falling back to a unary RPC if streams are down.
+      subscription.acknowledge(message);
+    },
+    onError: (Object error) {
+      print('Stream encountered a permanent error: $error');
+    },
+    onDone: () {
+      print('Stream closed.');
+    },
+  );
+
+  // Later, to stop receiving messages and clean up resources:
+  await listener.cancel();
+  // #enddocregion streaming_pull_example
+}
+
+Future<void> acknowledgeNowExample(PubSub pubsub) async {
+  // #docregion acknowledge_now_example
+  final subscription = pubsub.subscription('my-subscription');
+
+  // Pull up to 10 messages from the subscription.
+  final messages = await subscription.pull(maxMessages: 10);
+
+  if (messages.isNotEmpty) {
+    try {
+      // Acknowledge all pulled messages immediately and wait for the RPC
+      // to complete. This bypasses background batching.
+      await subscription.acknowledgeNow(messages);
+      print('Successfully acknowledged ${messages.length} messages.');
+    } on Exception catch (e) {
+      print('Failed to acknowledge messages: $e');
+    }
+  }
+  // #enddocregion acknowledge_now_example
+}

pkgs/google_cloud_pubsub/example/example.dart

@@ -23,8 +23,30 @@ Future<void> main() async {
   final pubsub = PubSub(projectId: 'my-project-id');
 
   try {
-    final topic = pubsub.topic('my-topic');
+    final topic = pubsub.topic(
+      'my-topic',
+      publishSettings: const PublishSettings(
+        batching: BatchingSettings(
+          maxMessages: 50,
+          maxDelay: Duration(milliseconds: 20),
+        ),
+      ),
+    );
     print('Successfully initialized client for topic: ${topic.id}');
+
+    final subscription = pubsub.subscription(
+      'my-subscription',
+      ackSettings: const AckSettings(
+        batching: BatchingSettings(
+          maxMessages: 50,
+          maxDelay: Duration(milliseconds: 20),
+        ),
+      ),
+    );
+    print('Successfully initialized subscription: ${subscription.id}');
+
+    topic.close();
+    subscription.close();
   } finally {
     await pubsub.close();
   }

pkgs/google_cloud_pubsub/lib/google_cloud_pubsub.dart

@@ -20,7 +20,9 @@ export 'package:grpc/grpc.dart'
         ServiceAccountAuthenticator,
         applicationDefaultCredentialsAuthenticator;
 
+export 'src/batching.dart' show BatchingSettings;
 export 'src/client.dart' show PubSub;
 export 'src/message.dart' show Message, ReceivedMessage;
-export 'src/subscription.dart' show Subscription;
-export 'src/topic.dart' show Topic;
+export 'src/retry.dart' show RetrySettings;
+export 'src/subscription.dart' show AckSettings, Subscription;
+export 'src/topic.dart' show PublishSettings, Topic;

pkgs/google_cloud_pubsub/lib/src/batching.dart

@@ -0,0 +1,89 @@
+// 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 'dart:async';
+
+import 'package:meta/meta.dart';
+
+/// Settings for batching operations.
+final class BatchingSettings {
+  /// The maximum number of items in a batch.
+  final int maxMessages;
+
+  /// The maximum size in bytes for a batch.
+  final int maxBytes;
+
+  /// The maximum time to wait before sending a batch.
+  final Duration maxDelay;
+
+  const BatchingSettings({
+    this.maxMessages = 100,
+    this.maxBytes = 1024 * 1024, // 1 MB
+    this.maxDelay = const Duration(milliseconds: 10),
+  });
+}
+
+/// Generic batcher that accumulates items of type [T] and fires batches of [T]
+/// according to [BatchingSettings].
+@internal
+class Batcher<T> {
+  final BatchingSettings settings;
+  final int Function(T) itemSize;
+  final Future<void> Function(List<T>) onBatch;
+
+  final List<T> _buffer = [];
+  int _currentSizeBytes = 0;
+  Timer? _timer;
+
+  Batcher({
+    required this.settings,
+    required this.itemSize,
+    required this.onBatch,
+  });
+
+  /// Adds an item to the batch.
+  void add(T item) {
+    _buffer.add(item);
+    _currentSizeBytes += itemSize(item);
+
+    if (_buffer.length >= settings.maxMessages ||
+        _currentSizeBytes >= settings.maxBytes) {
+      _flush();
+    } else {
+      _timer ??= Timer(settings.maxDelay, _flush);
+    }
+  }
+
+  void _flush() {
+    _timer?.cancel();
+    _timer = null;
+
+    if (_buffer.isEmpty) return;
+
+    final batch = List<T>.from(_buffer);
+    _buffer.clear();
+    _currentSizeBytes = 0;
+
+    // Fire and forget
+    onBatch(batch).catchError((_) {
+      // Errors should be handled by onBatch (e.g. failing the completers for
+      // the items).
+    });
+  }
+
+  /// Closes the batcher, flushing any remaining items immediately.
+  void close() {
+    _flush();
+  }
+}