feat: Pub/sub by sigurdm · Pull Request #244 · googleapis/google-cloud-dart

sigurdm · 2026-04-20T12:52:00Z

Initial implementation. Only basic functionality implemented (see todos).
Streaming pull is implemented.

Rely on the package:grpc BaseAuthenticator abstraction for authentication. Rexports those.

No built-in retry yet.

…issues

gemini-code-assist

Code Review

This pull request introduces a new Google Cloud Pub/Sub client library, including core client functionality, topic and subscription management, and support for the Pub/Sub emulator. The review identified several critical issues: a race condition and token refresh problem in the authentication logic, a resource leak where the auth client is not closed, an overly restrictive project ID requirement for the emulator, an incorrect parsing logic for IPv6 emulator hosts, and code duplication in message mapping. I have recommended addressing these issues to improve the reliability and maintainability of the client.

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

…add license/changelog

…tage test deletion

…ll integration tests

…ogle_cloud_pubsub

sigurdm · 2026-05-22T12:32:11Z

/gemini review

gemini-code-assist

Code Review

This pull request adds an experimental Google Cloud Pub/Sub client for Dart, supporting topic and subscription management along with message publishing and pull operations. Key feedback includes supporting full resource names for cross-project access, optimizing StreamingPull with bidirectional communication, and correcting Windows-specific environment variable retrieval logic. A minor style correction for the package description was also noted.

sigurdm · 2026-06-09T14:38:09Z

/gcbrun

jonasfj

Mostly just nits.

We could consider landing all the generated files first, if you want a more detailed review of the public API.

Introduce the handwritten client code (PubSub, Topic, Subscription, Message) on top of the generated gRPC code. - Implement PubSub client with emulator support. - Implement Topic and Subscription resource management. - Implement streaming pull and unary fallback for message acknowledgments. - Add unit and integration tests. - Add usage examples in README and example/example.dart. TAG=agy CONV=06ddc172-e8da-4e30-bab8-139c539520f9

- Make ReceivedMessage constructor private (ReceivedMessage._). - Introduce package-private createReceivedMessage helper annotated with @internal. - Update client.dart and unit_test.dart to use the helper. TAG=agy CONV=06ddc172-e8da-4e30-bab8-139c539520f9

Simplify ReceivedMessage by removing ack() and modifyAckDeadline() methods, which removes the need for storing internal callback handlers. - Make ReceivedMessage constructor public and simple. - Update client.dart to map ReceivedMessage without callbacks. - Update unit tests and integration tests to use Subscription.acknowledge() and Subscription.modifyAckDeadline() instead of message methods. - Fix linter warning in client.dart (use tearoff). TAG=agy CONV=06ddc172-e8da-4e30-bab8-139c539520f9

Add inline comment to _calculateProjectId explaining that we only fall back to 'test-project' when the emulator is active. TAG=agy CONV=06ddc172-e8da-4e30-bab8-139c539520f9

…tegy to resolve divergence

…with latest API design - Changed acknowledge and modifyAckDeadline to take ReceivedMessage and return void (fire-and-forget). - Added acknowledgeNow and modifyAckDeadlineNow to support immediate, awaitable operations on List<ReceivedMessage>. - Updated tests and examples accordingly. TAG=agy CONV=bc7c6164-e9cb-4a0f-b326-7752c814e1ce

brianquinlan · 2026-06-23T22:00:40Z

@@ -0,0 +1,93 @@
+// Copyright 2026 Google LLC


This looks copied from pkgs/google_cloud_storage/lib/src/storage_emulator_host_vm.dart

Maybe this functionality should be moved into package:google_cloud?

Agreed it's duplicated from package:google_cloud_storage. I'd like to propose punting that refactoring for now and addressing it in a follow up across both packages.

brianquinlan · 2026-06-23T22:08:42Z

+    final resolvedProjectId = _calculateProjectId(projectId, emulatorHost);
+    if (resolvedProjectId == null) {
+      throw ArgumentError(
+        'A project ID is required, but none was provided or could be '


Isn't this a StateError?

Good point. I've updated this to throw an ArgumentError from the constructor if the project ID is missing.

brianquinlan · 2026-06-23T22:12:46Z

+  /// For authentication, an [authenticator] can be supplied to obtain
+  /// and refresh access credentials for authenticating gRPC requests.
+  ///
+  /// If no [authenticator] is provided, requests are made without


Would it make sense to use ADC like the GCS and the other languages' pub sub clients?

Done. The client now falls back to using applicationDefaultCredentialsAuthenticator if the emulator is not in use.

brianquinlan · 2026-06-23T22:27:11Z

+      await _publisher.deleteTopic(request, options: await _callOptions);
+      return true;
+    } on GrpcError catch (e) {
+      if (e.code == StatusCode.notFound) {


In package:google_cloud_storage, I throw in this case. The other client libraries with exceptions also throw here. I think that we should be consistent, at least within our own packages.

Does an api expert (@natebosch ?) have an opinion?

I think either design is OK. Since we have precedent I lean towards matching what we already do elsewhere.

brianquinlan · 2026-06-23T22:29:31Z

+        options: await _callOptions,
+      );
+      return subscriptionName(name);
+    } on GrpcError catch (e) {


Maybe you want to handle this exception mapping in a function?

Done. I've extracted this into a _mapGrpcError helper function.

brianquinlan · 2026-06-23T22:31:45Z

+// limitations under the License.
+
+/// Base class for all exceptions thrown by the google_cloud_pubsub package.
+abstract final class PubSubException implements Exception {}


Do you want to define your own exception hierarchy? The other languages' clients use their generic rpc exception types (as does package:google_cloud_storage).

I've defined a custom PubSubException hierarchy to map from the underlying gRPC exceptions (matching what we do in packages like google_cloud_storage).

google_cloud_storage doesn't define it's own exceptions, it uses the exceptions defined in package:google_cloud_rpc.

I just checked and the Python pubsub client throws google.api_core.exceptions.NotFound if a topic is not found.

If we want to diverge from other languages' clients and a client-specific exception hierarchy for each client, then I think that we need a reason why.

Yeah, using google_cloud_rpc exceptions is a good idea.

However, when I went to integrate it, ServiceException (and all its subclasses like BadRequestException, NotFoundException, etc.) requires a non-optional http.BaseResponse response. Because google_cloud_pubsub uses gRPC over HTTP/2, we don't have an http.BaseResponse to pass.

Making response and responseBody optional in google_cloud_rpc would be a breaking change for existing google_cloud_rpc users (changing final http.BaseResponse response to http.BaseResponse?).

Are you open to making that breaking change in google_cloud_rpc (e.g. bumping to 0.6.0), or do you have another preferred way to integrate gRPC clients with the google_cloud_rpc exception hierarchy?

brianquinlan · 2026-06-23T22:36:47Z

@@ -0,0 +1,334 @@
+@TestOn('vm')


Do you want to have one test file per method under test? This is going to get pretty big quickly.

Done. Split integration_test.dart into individual test files per operation under test (create_topic_test.dart, delete_topic_test.dart, publish_test.dart, create_subscription_test.dart, pull_test.dart, streaming_pull_test.dart, acknowledge_test.dart, modify_ack_deadline_test.dart) and created test_utils.dart for shared setup.

Yes, I've gone ahead and split integration_test.dart into multiple smaller test files by method so it stays manageable.

natebosch · 2026-06-24T23:53:48Z

+    if (_isEmulator) {
+      return CallOptions();
+    }
+    final authenticator = _authenticator;
+    if (authenticator == null) {
+      return CallOptions();
+    }
+    return authenticator.toCallOptions;


[optional] I like to collapse early return guard clauses when they fit on one line. Also we can use ?? for the second null check.

Suggested change

if (_isEmulator) {

return CallOptions();

}

final authenticator = _authenticator;

if (authenticator == null) {

return CallOptions();

}

return authenticator.toCallOptions;

if (_isEmulator) return CallOptions();

return authenticator?.toCallOptions ?? CallOptions();

natebosch · 2026-06-24T23:58:14Z

+      await _publisher.deleteTopic(request, options: await _callOptions);
+      return true;
+    } on GrpcError catch (e) {
+      if (e.code == StatusCode.notFound) {


I think either design is OK. Since we have precedent I lean towards matching what we already do elsewhere.

natebosch · 2026-06-25T00:32:12Z

+import 'package:grpc/grpc.dart';
+import 'package:meta/meta.dart';
+import '../google_cloud_pubsub.dart';
+import 'generated/google/pubsub/v1/pubsub.pbgrpc.dart' as grpc;
+import 'pubsub_emulator_host_vm.dart';
+import 'subscription.dart' show newSubscription, newSubscriptionName;
+import 'topic.dart' show newTopic, newTopicName;


[nit]

Suggested change

import 'package:grpc/grpc.dart';

import 'package:meta/meta.dart';

import '../google_cloud_pubsub.dart';

import 'generated/google/pubsub/v1/pubsub.pbgrpc.dart' as grpc;

import 'pubsub_emulator_host_vm.dart';

import 'subscription.dart' show newSubscription, newSubscriptionName;

import 'topic.dart' show newTopic, newTopicName;

import 'package:grpc/grpc.dart';

import 'package:meta/meta.dart';

import '../google_cloud_pubsub.dart';

import 'generated/google/pubsub/v1/pubsub.pbgrpc.dart' as grpc;

import 'pubsub_emulator_host_vm.dart';

import 'subscription.dart' show newSubscription, newSubscriptionName;

import 'topic.dart' show newTopic, newTopicName;

natebosch · 2026-06-25T00:33:52Z

+      );
+    }
+
+    if (emulatorHost case String host) {


[nit] Avoid the appearance of inputs that may be other types by only using the syntax for the null check.

Suggested change

if (emulatorHost case String host) {

if (emulatorHost case final host?) {

natebosch · 2026-06-25T00:37:54Z

+/// and the emulator configuration might not be available until the application
+/// is launched.
+@internal
+String? get pubsubEmulatorHost {


[nit] inconsistent capitalization - other code treats "pub" and "sub" as different words but they are joined here.

natebosch · 2026-06-25T01:15:38Z

+    if (invocation.memberName == #cancel) {
+      return Future<void>.value();
+    }


How about implementing the cancel method instead?

natebosch · 2026-06-25T01:16:08Z

+class FakeResponseStream<T> extends Stream<T>
+    implements grpc.ResponseStream<T> {
+  final Stream<T> _stream;


Why are we both extending Stream and including a stream member?

natebosch · 2026-06-25T01:17:31Z

+    // Listen to request stream to prevent sender from hanging on close()
+    request.listen((_) {}, onDone: () {});


[nit]

Suggested change

// Listen to request stream to prevent sender from hanging on close()

request.listen((_) {}, onDone: () {});

// Listen to request stream to prevent sender from hanging on close()

request.drain();

natebosch · 2026-06-25T01:19:02Z

+    if (acknowledgeBehavior != null) {
+      acknowledgeBehavior!(request.ackIds)


[optional]

Suggested change

if (acknowledgeBehavior != null) {

acknowledgeBehavior!(request.ackIds)

if (acknowledgeBehavior case final acknowledge?) {

acknowledge(request.ackIds)

natebosch · 2026-06-25T01:21:58Z

+class FakeClientChannel implements grpc.ClientChannel {
+  @override
+  Future<void> shutdown() async {
+    // No-op for testing
+  }
+
+  @override
+  dynamic noSuchMethod(Invocation invocation) => super.noSuchMethod(invocation);
+}


You can extend Fake from package:test/test.dart to get this override

Suggested change

class FakeClientChannel implements grpc.ClientChannel {

@override

Future<void> shutdown() async {

// No-op for testing

}

@override

dynamic noSuchMethod(Invocation invocation) => super.noSuchMethod(invocation);

}

class FakeClientChannel extends Fake implements grpc.ClientChannel {

@override

Future<void> shutdown() async {

// No-op for testing

}

}

Similar in the other fakes.

- Fix finalSize check in Windows environment variable reader to handle errors/race conditions. - Update documentation to follow style guidelines for programmer errors (use "It is an error if..." instead of "Throws..."). TAG=agy CONV=ad5bf873-d53c-4338-9b86-1b6a21376d4d

- Default to ADC when not using emulator - Default emulator project ID to test-project - Parse emulator host as Uri? with validation - Update deleteTopic and deleteSubscription to throw on notFound - Rename name parameter to topic/subscription in PubSub client methods - Centralize GrpcError mapping - Rename topicId and subscriptionId to id - Split integration tests into individual files per operation - Rename unit_test.dart to error_propagation_test.dart and refactor fakes - Expand and clarify doc comments across public APIs

sigurdm added 2 commits April 13, 2026 07:45

Implement google_cloud_pubsub package with gRPC and resolve protobuf …

77d01f4

…issues

Use grpc

4be47d9

sigurdm changed the title ~~Pub sub~~ Pub/sub Apr 20, 2026

sigurdm changed the title ~~Pub/sub~~ feat: Pub/sub Apr 20, 2026

gemini-code-assist Bot reviewed Apr 20, 2026

View reviewed changes

sigurdm and others added 7 commits April 20, 2026 13:02

lints

aa7c181

update deps diagram

eda1eaa

Update pkgs/google_cloud_pubsub/lib/src/client.dart

bd77866

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

work

75b1543

PubSub: Remove web support, add common exception, improve tests, and …

89030f6

…add license/changelog

PubSub: Commit forgotten files (example and credentials helper) and s…

f5cb39d

…tage test deletion

Fix credentials helper race condition, HTTP client leak, and flaky pu…

5640799

…ll integration tests

sigurdm marked this pull request as ready for review May 21, 2026 14:44

sigurdm requested a review from brianquinlan May 21, 2026 14:44

sigurdm added 2 commits May 21, 2026 14:45

Merge branch 'origin/main' into pub_sub

b5bdc72

Regenerate workspace dependency diagram and update docs to include go…

ed9b387

…ogle_cloud_pubsub

sigurdm requested a review from jonasfj May 21, 2026 14:47

gemini-code-assist Bot reviewed May 22, 2026

View reviewed changes

jonasfj reviewed Jun 9, 2026

View reviewed changes

sigurdm requested a review from jonasfj June 12, 2026 12:17

jonasfj approved these changes Jun 18, 2026

View reviewed changes

Comment thread pkgs/google_cloud_pubsub/lib/src/message.dart Outdated

Comment thread pkgs/google_cloud_pubsub/lib/src/message.dart Outdated

Comment thread pkgs/google_cloud_pubsub/lib/src/client.dart Outdated

sigurdm force-pushed the pub_sub branch from 84433d9 to c34002d Compare June 19, 2026 08:33

sigurdm added 3 commits June 19, 2026 08:40

docs(pubsub): document project ID fallback for emulator

1468628

Add inline comment to _calculateProjectId explaining that we only fall back to 'test-project' when the emulator is active. TAG=agy CONV=06ddc172-e8da-4e30-bab8-139c539520f9

sigurdm mentioned this pull request Jun 22, 2026

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

Open

Merge origin/pub_sub (old history) into clean pub_sub using ours stra…

f407966

…tegy to resolve divergence

sigurdm added 2 commits June 22, 2026 10:25

chore(pubsub): add missing copyright headers to existing files

fe7ba0b

brianquinlan reviewed Jun 23, 2026

View reviewed changes

natebosch approved these changes Jun 25, 2026

View reviewed changes

sigurdm added 2 commits June 25, 2026 13:33

sigurdm requested a review from brianquinlan August 14, 2026 13:39

	if (emulatorHost case String host) {
	if (emulatorHost case final host?) {

		// Listen to request stream to prevent sender from hanging on close()
		request.listen((_) {}, onDone: () {});

		if (acknowledgeBehavior != null) {
		acknowledgeBehavior!(request.ackIds)

Uh oh!

Conversation

sigurdm commented Apr 20, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

gemini-code-assist Bot left a comment

Choose a reason for hiding this comment

Code Review

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

sigurdm commented May 22, 2026

Uh oh!

gemini-code-assist Bot left a comment

Choose a reason for hiding this comment

Code Review

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

sigurdm commented Jun 9, 2026

Uh oh!

jonasfj left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

sigurdm Aug 19, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

sigurdm commented Apr 20, 2026 •

edited

Loading

sigurdm Aug 19, 2026 •

edited

Loading

natebosch Jun 25, 2026 •

edited

Loading