feat(pubsub): add message and acknowledgment batching

[sigurdm]

This is Part 2 of 3 in a stacked set of PRs:

1. feat(pubsub): add message and acknowledgment batching with retry infrastructure #292: Core retry and batching infrastructure (RetrySettings, Batcher, BatchingSettings)
2. feat(pubsub): add message and acknowledgment batching #338 (this PR): Topic publishing batching & lifecycle management (PublishSettings, Topic.publish, Topic.close)
3. feat(pubsub): add resilient parallel streaming pull #339: Subscription batching, resilient parallel streaming pull, and lifecycle management (AckSettings, Subscription.streamingPull, Subscription.close)

---

Key Features in Part 2

• Topic Batching: Added PublishSettings controlling background batching (BatchingSettings) and unary retry configuration (RetrySettings) for Topic.publish.
• Accurate Byte Counting: Computes payload and attribute key/value byte overhead when batching messages against byte limits.
• Completer Safety: Ensures all completers in a batch complete or fail exactly once even on mismatched server responses or backend errors.
• Lifecycle Management: Async Topic.close() that flushes pending batches and awaits in-flight RPCs, preventing subsequent publishes with StateError.
• Instance Settings Preservation: Topic.create() returns this rather than constructing a new default instance, preserving configured publishSettings.
• String Representations: Informative Message.toString() and ReceivedMessage.toString().

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

---

_{Stack created with GitHub Stacks CLI • Give Feedback 💬}

[sigurdm]

/gcbrun

[gemini-code-assist]

Code Review

This pull request introduces message batching and retry settings (PublishSettings) for Topic publishing, along with lifecycle improvements such as an asynchronous close() method and safer batch completion handling. Review feedback suggests optimizing the message size calculation by avoiding utf8.encode() allocations to reduce memory churn, and simplifying a redundant Future.value return in an async method.

[gemini-code-assist]

In an 'async' function, returning 'Future.value(...)' is redundant because any returned value is automatically wrapped in a 'Future'. You can simplify this by returning the list literal directly.

Suggested change

	if (messages.isEmpty) return Future.value(<String>[]);
	if (messages.isEmpty) return <String>[];

[sigurdm]

Fixed: Removed the redundant Future.value(...) wrapping in publishMessages (and similarly in acknowledge and modifyAckDeadline).

[sigurdm]

/gcbrun

[sigurdm]

/gcbrun

[sigurdm]

/gcbrun

[sigurdm]

Collapsed into PR #292 (retries + batching combined).