diff --git a/DEVELOPER_GUIDE.md b/DEVELOPER_GUIDE.md index eb0dcbba..c5ca4952 100644 --- a/DEVELOPER_GUIDE.md +++ b/DEVELOPER_GUIDE.md @@ -17,7 +17,6 @@ graph TD subgraph Tier0 ["Tier 0 (Publish First)"] google_cloud["google_cloud"] google_cloud_protobuf["protobuf"] - google_cloud_pubsub["pubsub"] end subgraph Tier1 ["Tier 1"] @@ -33,6 +32,7 @@ graph TD google_cloud_language_v2["language_v2"] google_cloud_location["location"] google_cloud_longrunning["longrunning"] + google_cloud_pubsub["pubsub"] google_cloud_storage["storage"] end @@ -93,6 +93,9 @@ graph TD google_cloud_logging_v2 --> google_cloud_rpc google_cloud_longrunning --> google_cloud_protobuf google_cloud_longrunning --> google_cloud_rpc + google_cloud_pubsub --> google_cloud + google_cloud_pubsub --> google_cloud_protobuf + google_cloud_pubsub --> google_cloud_rpc google_cloud_rpc --> google_cloud_protobuf google_cloud_secretmanager_v1 --> google_cloud_iam_v1 google_cloud_secretmanager_v1 --> google_cloud_location diff --git a/pkgs/google_cloud_pubsub/README.md b/pkgs/google_cloud_pubsub/README.md index 9162ab7d..7ee16c31 100644 --- a/pkgs/google_cloud_pubsub/README.md +++ b/pkgs/google_cloud_pubsub/README.md @@ -17,9 +17,59 @@ A Dart client for Google Cloud Pub/Sub. > feedback, suggestions, and comments, please file an issue in the > [bug tracker](https://github.com/googleapis/google-cloud-dart/issues). -## Usage +## Using Google Cloud PubSub -This package contains generated gRPC client code for Google Cloud Pub/Sub. +All access to Google Cloud Pub/Sub is made through the `PubSub` class. -For details on how to use the generated gRPC client, please refer to the gRPC and Protobuf documentation. +```dart +import 'dart:convert'; +import 'package:google_cloud_pubsub/google_cloud_pubsub.dart'; +void main() async { + // Note: You must provide an `authenticator` for authentication in production. + final pubSub = PubSub( + projectId: 'your-project-id', + authenticator: await applicationDefaultCredentialsAuthenticator( + ['https://www.googleapis.com/auth/pubsub'], + ), + ); + + // Create a topic. + final topic = await pubSub.topic('put-your-topic-name-here').create(); + + // Create a subscription to that topic. + final subscription = await pubSub + .subscription('put-your-subscription-name-here') + .create(topic: topic.name); + + // Publish a message. + await topic.publish(utf8.encode('message 1')); + + // Pull messages from the subscription. + final messages = await subscription.pull(maxMessages: 1); + + for (final receivedMessage in messages) { + print('Received message: ${utf8.decode(receivedMessage.data)}'); + + // Acknowledge the message. + await subscription.acknowledge([receivedMessage.ackId]); + } + + print( + 'Your topic is available at:\n' + 'https://pubsub.googleapis.com/v1/${topic.name}', + ); + + // Clean up. + await subscription.delete(); + await topic.delete(); + + pubSub.close(); +} +``` + +> [!NOTE] +> You must [set up authentication][] before using this package outside of +> Google Cloud. + +[set up authentication]: https://docs.cloud.google.com/pubsub/docs/reference/libraries#authentication diff --git a/pkgs/google_cloud_pubsub/example/example.dart b/pkgs/google_cloud_pubsub/example/example.dart new file mode 100644 index 00000000..92b81899 --- /dev/null +++ b/pkgs/google_cloud_pubsub/example/example.dart @@ -0,0 +1,40 @@ +// 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'; + +/// An example demonstrating how to initialize and use the [PubSub] client +/// with Google Application Default Credentials (ADC) authentication. +Future main() async { + // Asynchronously construct the ADC authenticator with the required Pub/Sub scope. + final authenticator = await applicationDefaultCredentialsAuthenticator([ + 'https://www.googleapis.com/auth/pubsub', + ]); + + // Pass the authenticator to the PubSub client constructor. + final pubsub = PubSub( + projectId: 'my-project-id', + authenticator: authenticator, + ); + + try { + // The client will use the authenticator to make authenticated requests. + final topic = pubsub.topic('my-topic'); + print('Successfully initialized client for topic: ${topic.id}'); + } finally { + await pubsub.close(); + } +} diff --git a/pkgs/google_cloud_pubsub/lib/google_cloud_pubsub.dart b/pkgs/google_cloud_pubsub/lib/google_cloud_pubsub.dart index 76d683ca..dd63f35f 100644 --- a/pkgs/google_cloud_pubsub/lib/google_cloud_pubsub.dart +++ b/pkgs/google_cloud_pubsub/lib/google_cloud_pubsub.dart @@ -12,7 +12,15 @@ // See the License for the specific language governing permissions and // limitations under the License. -export 'src/generated/google/pubsub/v1/pubsub.pb.dart'; -export 'src/generated/google/pubsub/v1/pubsub.pbgrpc.dart'; -export 'src/generated/google/pubsub/v1/schema.pb.dart'; -export 'src/generated/google/pubsub/v1/schema.pbgrpc.dart'; +export 'package:grpc/grpc.dart' + show + BaseAuthenticator, + ComputeEngineAuthenticator, + ServiceAccountAuthenticator, + applicationDefaultCredentialsAuthenticator; + +export 'src/client.dart' show PubSub; +export 'src/exceptions.dart'; +export 'src/message.dart' show Message, ReceivedMessage; +export 'src/subscription.dart' show Subscription; +export 'src/topic.dart' show Topic; diff --git a/pkgs/google_cloud_pubsub/lib/src/client.dart b/pkgs/google_cloud_pubsub/lib/src/client.dart new file mode 100644 index 00000000..db78ec3e --- /dev/null +++ b/pkgs/google_cloud_pubsub/lib/src/client.dart @@ -0,0 +1,550 @@ +// 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: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'; + +const _pubsubScopes = ['https://www.googleapis.com/auth/pubsub']; + +/// API for flexible, reliable, large-scale messaging. +/// +/// See [Google Cloud Pub/Sub](https://cloud.google.com/pubsub). +final class PubSub { + /// The project ID of this client. + final String projectId; + final ClientChannel _channel; + final bool _isEmulator; + grpc.PublisherClient? _publisherClient; + grpc.SubscriberClient? _subscriberClient; + final FutureOr? _authenticator; + + static String? _calculateProjectId( + String? projectId, + Uri? emulatorHost, + ) => switch ((projectId, emulatorHost)) { + (final String projectId, _) => projectId, + // When the emulator is active (emulatorHost is not null), we fall back + // to 'test-project' if GOOGLE_CLOUD_PROJECT is not set in the environment. + (null, _?) => projectFromEnvironment ?? 'test-project', + (null, null) => projectFromEnvironment, + }; + + static ClientChannel _calculateChannel( + String? apiEndpoint, + Uri? emulatorHost, + ) { + if (apiEndpoint != null) { + return ClientChannel( + apiEndpoint, + options: const ChannelOptions(credentials: ChannelCredentials.secure()), + ); + } + + if (emulatorHost case final uri?) { + return ClientChannel( + uri.host, + port: uri.hasPort ? uri.port : 8085, + options: const ChannelOptions( + credentials: ChannelCredentials.insecure(), + ), + ); + } + + return ClientChannel( + 'pubsub.googleapis.com', + options: const ChannelOptions(credentials: ChannelCredentials.secure()), + ); + } + + PubSub._( + this.projectId, + this._channel, + this._isEmulator, + this._authenticator, { + grpc.SubscriberClient? subscriberClient, + grpc.PublisherClient? publisherClient, + }) : _subscriberClient = subscriberClient, + _publisherClient = publisherClient; + + @visibleForTesting + factory PubSub.testing({ + required String projectId, + required ClientChannel channel, + grpc.SubscriberClient? subscriberClient, + grpc.PublisherClient? publisherClient, + }) => PubSub._( + projectId, + channel, + false, + null, + subscriberClient: subscriberClient, + publisherClient: publisherClient, + ); + + /// Turns the protobuf-generated [grpc.ReceivedMessage] into a + /// [ReceivedMessage]. + static ReceivedMessage _mapReceivedMessage(grpc.ReceivedMessage m) => + ReceivedMessage( + ackId: m.ackId, + messageId: m.message.messageId, + publishTime: m.message.publishTime.toDateTime(), + message: Message( + data: m.message.data, + attributes: m.message.attributes, + ), + ); + + /// Constructs a client used to communicate with [Google Cloud Pub/Sub][]. + /// + /// The [projectId] is the Google Cloud Project ID. If not provided, it will + /// be inferred from the environment. + /// + /// Project ID inference strategies: + /// 1. Reads the `GOOGLE_CLOUD_PROJECT` environment variable. + /// 2. If the `PUBSUB_EMULATOR_HOST` environment variable is set (indicating + /// the emulator is active), it defaults to `'test-project'`. + /// + /// It is an error if [projectId] is not provided and cannot be + /// inferred from the environment. + /// + /// 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 + /// authentication. + factory PubSub({ + String? projectId, + String? apiEndpoint, + BaseAuthenticator? authenticator, + }) { + final emulatorHost = pubSubEmulatorHost; + final resolvedProjectId = _calculateProjectId(projectId, emulatorHost); + if (resolvedProjectId == null) { + throw ArgumentError( + 'A project ID is required, but none was provided or could be ' + 'inferred from the environment.', + ); + } + return PubSub._( + resolvedProjectId, + _calculateChannel(apiEndpoint, emulatorHost), + emulatorHost != null, + authenticator ?? + (emulatorHost != null + ? null + : applicationDefaultCredentialsAuthenticator(_pubsubScopes)), + ); + } + + Future get _callOptions async { + if (_isEmulator) return CallOptions(); + final authenticator = await _authenticator; + return authenticator?.toCallOptions ?? CallOptions(); + } + + grpc.PublisherClient get _publisher => + _publisherClient ??= grpc.PublisherClient(_channel); + grpc.SubscriberClient get _subscriber => + _subscriberClient ??= grpc.SubscriberClient(_channel); + + /// Closes the client and cleans up any resources associated with it. + Future close() async { + await _channel.shutdown(); + } + + // Topic-related methods + + /// A [Topic] object with the given [unqualifiedName] in the client's project. + Topic topic(String unqualifiedName) => + Topic.unqualified(this, unqualifiedName); + + /// A [Topic] object with the given [name]. + /// + /// The [name] must be in the format `projects//topics/`. + /// Useful for cross-project access. + Topic topicName(String name) => Topic(this, name); + + /// A [Subscription] object with the given [unqualifiedName] in the client's + /// project. + Subscription subscription(String unqualifiedName) => + Subscription.unqualified(this, unqualifiedName); + + /// A [Subscription] object with the given [name]. + /// + /// The [name] must be in the format + /// `projects//subscriptions/`. + /// Useful for cross-project access. + Subscription subscriptionName(String name) => Subscription(this, name); + + /// Creates the given topic with the given [topic]. + /// + /// The [topic] must be in the format `projects//topics/`. + /// + /// Throws a [TopicAlreadyExistsException] if the topic already exists. + /// + /// See the [official documentation](https://cloud.google.com/pubsub/docs/reference/rpc/google.pubsub.v1#google.pubsub.v1.Publisher.CreateTopic). + // TODO(sigurdm): Support configuring topic options (labels, + // messageStoragePolicy, kmsKeyName, schemaSettings, + // messageRetentionDuration). + Future createTopic(String topic) async { + final t = grpc.Topic()..name = topic; + try { + await _publisher.createTopic(t, options: await _callOptions); + return topicName(topic); + } on GrpcError catch (e) { + throw _mapGrpcError( + e, + specificMappings: { + StatusCode.alreadyExists: (_) => TopicAlreadyExistsException(topic), + }, + ); + } + } + + /// Deletes the topic with the given [topic]. + /// + /// The [topic] must be in the format `projects//topics/`. + /// + /// Throws a [TopicNotFoundException] if the topic does not exist. + /// + /// See the [official documentation](https://cloud.google.com/pubsub/docs/reference/rpc/google.pubsub.v1#google.pubsub.v1.Publisher.DeleteTopic). + Future deleteTopic(String topic) async { + final request = grpc.DeleteTopicRequest()..topic = topic; + try { + await _publisher.deleteTopic(request, options: await _callOptions); + } on GrpcError catch (e) { + throw _mapGrpcError( + e, + specificMappings: { + StatusCode.notFound: (_) => TopicNotFoundException(topic), + }, + ); + } + } + + /// Adds one or more messages to the topic. + /// + /// The [topic] must be in the format `projects//topics/`. + /// + /// Throws a [TopicNotFoundException] if the topic does not exist. + /// + /// See the [official documentation](https://cloud.google.com/pubsub/docs/reference/rpc/google.pubsub.v1#google.pubsub.v1.Publisher.Publish). + // TODO(sigurdm): Support batch publishing (publishMany) for high-throughput. + Future publish( + String topic, + List data, { + Map? attributes, + }) async { + final message = grpc.PubsubMessage()..data = data; + if (attributes != null) { + message.attributes.addAll(attributes); + } + + final request = grpc.PublishRequest() + ..topic = topic + ..messages.add(message); + + try { + final response = await _publisher.publish( + request, + options: await _callOptions, + ); + return response.messageIds.first; + } on GrpcError catch (e) { + throw _mapGrpcError( + e, + specificMappings: { + StatusCode.notFound: (_) => TopicNotFoundException(topic), + }, + ); + } + } + + // TODO(sigurdm): Implement missing Publisher APIs: + // - GetTopic + // - UpdateTopic + // - ListTopics + // - ListTopicSubscriptions + // - ListTopicSnapshots + // - DetachSubscription + + // Subscription-related methods + + /// Creates a subscription to a given topic. + /// + /// The [subscription] must be in the format + /// `projects//subscriptions/`. + /// The [topic] must be in the format `projects//topics/`. + /// + /// Throws a [SubscriptionAlreadyExistsException] if the subscription already + /// exists. + /// Throws a [TopicNotFoundException] if the corresponding topic doesn't + /// exist. + /// + /// See the [official documentation](https://cloud.google.com/pubsub/docs/reference/rpc/google.pubsub.v1#google.pubsub.v1.Subscriber.CreateSubscription). + // TODO(sigurdm): Support configuring subscription options + // (ackDeadlineSeconds, pushConfig, deadLetterPolicy, retryPolicy, + // retainAckedMessages, enableExactlyOnceDelivery). + Future createSubscription( + String subscription, { + required String topic, + }) async { + final sub = grpc.Subscription() + ..name = subscription + ..topic = topic; + + try { + await _subscriber.createSubscription(sub, options: await _callOptions); + return subscriptionName(subscription); + } on GrpcError catch (e) { + throw _mapGrpcError( + e, + specificMappings: { + StatusCode.alreadyExists: (_) => + SubscriptionAlreadyExistsException(subscription), + StatusCode.notFound: (_) => TopicNotFoundException(topic), + }, + ); + } + } + + /// Deletes an existing subscription. + /// + /// The [subscription] must be in the format + /// `projects//subscriptions/`. + /// + /// Throws a [SubscriptionNotFoundException] if the subscription does not + /// exist. + /// + /// See the [official documentation](https://cloud.google.com/pubsub/docs/reference/rpc/google.pubsub.v1#google.pubsub.v1.Subscriber.DeleteSubscription). + Future deleteSubscription(String subscription) async { + final request = grpc.DeleteSubscriptionRequest() + ..subscription = subscription; + try { + await _subscriber.deleteSubscription( + request, + options: await _callOptions, + ); + } on GrpcError catch (e) { + throw _mapGrpcError( + e, + specificMappings: { + StatusCode.notFound: (_) => + SubscriptionNotFoundException(subscription), + }, + ); + } + } + + /// Pulls messages from the server. + /// + /// The [subscription] must be in the format + /// `projects//subscriptions/`. + /// + /// Throws a [SubscriptionNotFoundException] if the subscription does not + /// exist. + /// + /// See the [official documentation](https://cloud.google.com/pubsub/docs/reference/rpc/google.pubsub.v1#google.pubsub.v1.Subscriber.Pull). + Future> pull( + String subscription, { + int maxMessages = 1, + }) async { + final request = grpc.PullRequest() + ..subscription = subscription + ..maxMessages = maxMessages; + + try { + final response = await _subscriber.pull( + request, + options: await _callOptions, + ); + + return response.receivedMessages.map(_mapReceivedMessage).toList(); + } on GrpcError catch (e) { + throw _mapGrpcError( + e, + specificMappings: { + StatusCode.notFound: (_) => + SubscriptionNotFoundException(subscription), + }, + ); + } + } + + /// Establishes a stream with the server, which sends messages down to the + /// client. + /// + /// The [subscription] must be in the format + /// `projects//subscriptions/`. + /// + /// The client streams acknowledgments and ack deadline modifications + /// back to the server. If an error occurs (including when the server closes + /// the stream with status `UNAVAILABLE` to reassign resources), the stream + /// will throw a [StreamBrokenException]. In this case, the caller should + /// re-establish the stream. Flow control can be achieved by configuring the + /// underlying RPC channel. + /// + /// Throws a [StreamBrokenException] if the stream is broken by the server or + /// network. + /// + /// See the [official documentation](https://cloud.google.com/pubsub/docs/reference/rpc/google.pubsub.v1#google.pubsub.v1.Subscriber.StreamingPull). + Stream streamingPull( + String subscription, { + int streamAckDeadlineSeconds = 10, + }) async* { + final requestController = StreamController(); + try { + final options = await _callOptions; + requestController.add( + grpc.StreamingPullRequest() + ..subscription = subscription + ..streamAckDeadlineSeconds = streamAckDeadlineSeconds, + ); + // TODO(sigurdm): Retry on broken connections. + final responseStream = _subscriber.streamingPull( + requestController.stream, + options: options, + ); + await for (final response in responseStream) { + for (final m in response.receivedMessages) { + yield _mapReceivedMessage(m); + } + } + } on GrpcError catch (e) { + throw StreamBrokenException( + e.code, + e.message ?? 'Unknown error', + e.trailers ?? const {}, + ); + } finally { + await requestController.close(); + } + } + + /// Acknowledges the messages associated with the `ack_ids`. + /// + /// The [subscription] must be in the format + /// `projects//subscriptions/`. + /// + /// The Pub/Sub system can remove the relevant messages from the subscription. + /// + /// Acknowledging a message whose ack deadline has expired may succeed, + /// but such a message may be redelivered later. Acknowledging a message more + /// than once will not result in an error. + /// + /// Throws a [SubscriptionNotFoundException] if the subscription does not + /// exist. + /// + /// See the [official documentation](https://cloud.google.com/pubsub/docs/reference/rpc/google.pubsub.v1#google.pubsub.v1.Subscriber.Acknowledge). + Future acknowledge(String subscription, List ackIds) async { + final request = grpc.AcknowledgeRequest() + ..subscription = subscription + ..ackIds.addAll(ackIds); + + try { + await _subscriber.acknowledge(request, options: await _callOptions); + } on GrpcError catch (e) { + throw _mapGrpcError( + e, + specificMappings: { + StatusCode.notFound: (_) => + SubscriptionNotFoundException(subscription), + }, + ); + } + } + + /// Modifies the ack deadline for a list of specific messages. + /// + /// The [subscription] must be in the format + /// `projects//subscriptions/`. + /// + /// This method is useful to indicate that more time is needed to process a + /// message by the subscriber, or to make the message available for redelivery + /// if the processing was interrupted. Note that this does not modify the + /// subscription-level `ackDeadlineSeconds` used for subsequent messages. + /// + /// Modifying the ack deadline for messages whose deadline has already expired + /// may succeed, but those messages may have already been redelivered or + /// made available for redelivery. + /// + /// Throws a [SubscriptionNotFoundException] if the subscription does not + /// exist. + /// + /// See the [official documentation](https://cloud.google.com/pubsub/docs/reference/rpc/google.pubsub.v1#google.pubsub.v1.Subscriber.ModifyAckDeadline). + Future modifyAckDeadline( + String subscription, + List ackIds, + int ackDeadlineSeconds, + ) async { + final request = grpc.ModifyAckDeadlineRequest() + ..subscription = subscription + ..ackIds.addAll(ackIds) + ..ackDeadlineSeconds = ackDeadlineSeconds; + + try { + await _subscriber.modifyAckDeadline(request, options: await _callOptions); + } on GrpcError catch (e) { + throw _mapGrpcError( + e, + specificMappings: { + StatusCode.notFound: (_) => + SubscriptionNotFoundException(subscription), + }, + ); + } + } + + // TODO(sigurdm): Implement missing Subscriber APIs: + // - GetSubscription + // - UpdateSubscription + // - ListSubscriptions + // - ModifyPushConfig + // - GetSnapshot + // - ListSnapshots + // - CreateSnapshot + // - UpdateSnapshot + // - DeleteSnapshot + // - Seek + + // TODO(sigurdm): Implement missing Schema APIs: + // - CreateSchema + // - GetSchema + // - ListSchemas + // - ListSchemaRevisions + // - CommitSchema + // - RollbackSchema + // - DeleteSchemaRevision + // - DeleteSchema + // - ValidateSchema + // - ValidateMessage + Exception _mapGrpcError( + GrpcError e, { + Map specificMappings = const {}, + }) { + if (specificMappings[e.code] case final exceptionBuilder?) { + return exceptionBuilder(e); + } + return PubSubOperationException( + e.code, + e.message ?? 'Unknown error', + e.trailers ?? const {}, + ); + } +} diff --git a/pkgs/google_cloud_pubsub/lib/src/exceptions.dart b/pkgs/google_cloud_pubsub/lib/src/exceptions.dart new file mode 100644 index 00000000..12bac811 --- /dev/null +++ b/pkgs/google_cloud_pubsub/lib/src/exceptions.dart @@ -0,0 +1,93 @@ +// 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. + +/// Base class for all exceptions thrown by the google_cloud_pubsub package. +abstract final class PubSubException implements Exception {} + +/// Thrown when a topic is not found. +final class TopicNotFoundException implements PubSubException { + /// The name of the topic that was not found. + final String name; + + TopicNotFoundException(this.name); + + @override + String toString() => 'TopicNotFoundException: Topic "$name" not found.'; +} + +/// Thrown when a topic already exists. +final class TopicAlreadyExistsException implements PubSubException { + /// The name of the topic that already exists. + final String name; + + TopicAlreadyExistsException(this.name); + + @override + String toString() => + 'TopicAlreadyExistsException: Topic "$name" already exists.'; +} + +/// Thrown when a subscription already exists. +final class SubscriptionAlreadyExistsException implements PubSubException { + /// The name of the subscription that already exists. + final String name; + + SubscriptionAlreadyExistsException(this.name); + + @override + String toString() => + 'SubscriptionAlreadyExistsException: Subscription "$name" ' + 'already exists.'; +} + +/// Thrown when a subscription is not found. +final class SubscriptionNotFoundException implements PubSubException { + /// The name of the subscription that was not found. + final String name; + + SubscriptionNotFoundException(this.name); + + @override + String toString() => + 'SubscriptionNotFoundException: Subscription "$name" not found.'; +} + +/// Thrown when a streaming pull connection is broken. +final class StreamBrokenException implements PubSubException { + final int code; + final String message; + final Map trailers; + + StreamBrokenException(this.code, this.message, [this.trailers = const {}]); + + @override + String toString() => 'StreamBrokenException: [$code] $message'; +} + +/// Thrown when a Pub/Sub operation fails due to a gRPC error. +final class PubSubOperationException implements PubSubException { + /// The gRPC status code. + final int code; + + /// The error message. + final String message; + + /// The trailers from the gRPC error. + final Map trailers; + + PubSubOperationException(this.code, this.message, [this.trailers = const {}]); + + @override + String toString() => 'PubSubOperationException: [$code] $message'; +} diff --git a/pkgs/google_cloud_pubsub/lib/src/message.dart b/pkgs/google_cloud_pubsub/lib/src/message.dart new file mode 100644 index 00000000..87b49341 --- /dev/null +++ b/pkgs/google_cloud_pubsub/lib/src/message.dart @@ -0,0 +1,57 @@ +// 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:typed_data'; + +/// A Pub/Sub message. +final class Message { + /// The payload of this message. + final Uint8List data; + + /// Optional attributes for this message. + final Map attributes; + + Message({required List data, Map? attributes}) + : data = data is Uint8List ? data : Uint8List.fromList(data), + attributes = attributes ?? const {}; +} + +/// A message received from a subscription. +final class ReceivedMessage { + /// The acknowledgment ID, used to identify this message when acknowledging + /// it or modifying its acknowledgment deadline. + final String ackId; + + /// The message payload and attributes. + final Message message; + + /// The ID of this message, assigned by the server. + final String messageId; + + /// The time at which the message was published. + final DateTime publishTime; + + ReceivedMessage({ + required this.ackId, + required this.messageId, + required this.publishTime, + required this.message, + }); + + /// The message data. + Uint8List get data => message.data; + + /// Optional attributes for this message. + Map get attributes => message.attributes; +} diff --git a/pkgs/google_cloud_pubsub/lib/src/pubsub_emulator_host_vm.dart b/pkgs/google_cloud_pubsub/lib/src/pubsub_emulator_host_vm.dart new file mode 100644 index 00000000..ada3d65b --- /dev/null +++ b/pkgs/google_cloud_pubsub/lib/src/pubsub_emulator_host_vm.dart @@ -0,0 +1,106 @@ +// 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:ffi'; +import 'dart:io'; + +import 'package:ffi/ffi.dart'; +import 'package:meta/meta.dart'; + +final _getenv = DynamicLibrary.process() + .lookupFunction< + Pointer Function(Pointer), + Pointer Function(Pointer) + >('getenv'); + +final _getEnvironmentVariableW = DynamicLibrary.open('kernel32.dll') + .lookupFunction< + Int32 Function(Pointer, Pointer, Int32), + int Function(Pointer, Pointer, int) + >('GetEnvironmentVariableW'); + +String? _posixEnvironmentVariable(String name) => using((arena) { + final namePtr = name.toNativeUtf8(allocator: arena); + final valuePtr = _getenv(namePtr); + if (valuePtr == nullptr) { + return null; + } + return valuePtr.toDartString(); +}); + +String? _windowsEnvironmentVariable(String name) => using((arena) { + final namePtr = name.toNativeUtf16(allocator: arena); + // First call to determine size. + // size includes the terminating null character. + final size = _getEnvironmentVariableW(namePtr, nullptr, 0); + if (size == 0) { + return null; // Variable not found. + } + + final buffer = arena(size).cast(); + final finalSize = _getEnvironmentVariableW(namePtr, buffer, size); + if (finalSize == 0 || finalSize >= size) { + return null; + } + // finalSize is the number of characters stored, NOT including null. + // If finalSize is 0 and size was 1, it's an empty string. + return buffer.toDartString(); +}); + +String? _environmentVariable(String name) { + if (Platform.isWindows) { + return _windowsEnvironmentVariable(name); + } else { + return _posixEnvironmentVariable(name); + } +} + +/// The host and port of the Google Cloud Pub/Sub emulator, if set. +/// +/// Reads the `PUBSUB_EMULATOR_HOST` environment variable directly using native +/// code. [Platform.environment] is not used because it is cached and immutable +/// and the emulator configuration might not be available until the application +/// is launched. +@internal +Uri? get pubSubEmulatorHost { + final host = _environmentVariable('PUBSUB_EMULATOR_HOST'); + if (host == null || host.isEmpty) return null; + + final uri = host.startsWith('http://') || host.startsWith('https://') + ? Uri.tryParse(host) + : Uri.tryParse('http://$host'); + + if (uri == null) return null; + + // gRPC channels only require host and port. If the URI contains a path, + // query, or fragment, it is invalid for this purpose. + if ((uri.path.isNotEmpty && uri.path != '/') || + uri.query.isNotEmpty || + uri.fragment.isNotEmpty) { + return null; + } + + return uri; +} + +/// The project ID inferred from the environment. +/// +/// Reads the `GOOGLE_CLOUD_PROJECT` environment variable directly using native +/// code. +@internal +String? get projectFromEnvironment { + final project = _environmentVariable('GOOGLE_CLOUD_PROJECT'); + if (project?.isEmpty ?? true) return null; + return project; +} diff --git a/pkgs/google_cloud_pubsub/lib/src/subscription.dart b/pkgs/google_cloud_pubsub/lib/src/subscription.dart new file mode 100644 index 00000000..fa6722ea --- /dev/null +++ b/pkgs/google_cloud_pubsub/lib/src/subscription.dart @@ -0,0 +1,214 @@ +// 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 '../google_cloud_pubsub.dart'; + +/// A [Google Cloud Pub/Sub subscription](https://cloud.google.com/pubsub/docs/overview#subscriptions). +final class Subscription { + static final RegExp _subscriptionNameRegExp = RegExp( + r'^projects/[^/]+/subscriptions/[^/]+$', + ); + + /// The [PubSub] client associated with this subscription. + final PubSub pubsub; + + /// The fully qualified resource name of this subscription. + /// + /// It has the format `projects//subscriptions/`. + final String name; + + /// A subscription with the given [subscriptionId] in the client's project. + /// + /// It is an error if the constructed subscription name is invalid (e.g. if + /// [subscriptionId] contains slashes). + Subscription.unqualified(this.pubsub, String subscriptionId) + : name = 'projects/${pubsub.projectId}/subscriptions/$subscriptionId' { + _validateName(name); + } + + /// A subscription with the given [name]. + /// + /// Useful for cross-project access. + /// + /// It is an error if [name] is not in the format + /// `projects//subscriptions/`. + Subscription(this.pubsub, this.name) { + _validateName(name); + } + + static void _validateName(String name) { + if (!_subscriptionNameRegExp.hasMatch(name)) { + throw ArgumentError.value( + name, + 'name', + 'Must be in the format projects//subscriptions/', + ); + } + } + + /// The unqualified ID of this subscription. + String get id => name.split('/').last; + + /// Creates this subscription on the server, associating it with the [topic]. + /// + /// The subscription must not already exist on the server. + /// The [topic] must exist on the server. + /// + /// Throws a [SubscriptionAlreadyExistsException] if the subscription already + /// exists. + /// Throws a [TopicNotFoundException] if the corresponding topic doesn't + /// exist. + /// + /// Returns a [Subscription] instance representing the created subscription. + /// + /// See the [official documentation](https://cloud.google.com/pubsub/docs/reference/rpc/google.pubsub.v1#google.pubsub.v1.Subscriber.CreateSubscription). + Future create({required String topic}) => + pubsub.createSubscription(name, topic: topic); + + /// Deletes this subscription on the server. + /// + /// All messages retained in the subscription are immediately dropped. + /// + /// Throws a [SubscriptionNotFoundException] if the subscription does not + /// exist. + /// + /// See the [official documentation](https://cloud.google.com/pubsub/docs/reference/rpc/google.pubsub.v1#google.pubsub.v1.Subscriber.DeleteSubscription). + Future delete() => pubsub.deleteSubscription(name); + + /// Pulls messages from the server. + /// + /// Throws a [SubscriptionNotFoundException] if the subscription does not + /// exist. + /// + /// It is an error if [maxMessages] is less than or equal to 0. + /// + /// See the [official documentation](https://cloud.google.com/pubsub/docs/reference/rpc/google.pubsub.v1#google.pubsub.v1.Subscriber.Pull). + Future> pull({int maxMessages = 1}) { + if (maxMessages <= 0) { + throw ArgumentError.value( + maxMessages, + 'maxMessages', + 'Must be greater than 0', + ); + } + return pubsub.pull(name, maxMessages: maxMessages); + } + + /// Establishes a stream with the server, which sends messages down to the + /// client. + /// + /// The client streams acknowledgments and ack deadline modifications + /// back to the server. If an error occurs (including when the server closes + /// the stream with status `UNAVAILABLE` to reassign resources), the stream + /// will throw a [StreamBrokenException]. In this case, the caller should + /// re-establish the stream. Flow control can be achieved by configuring the + /// underlying RPC channel. + /// + /// Throws a [StreamBrokenException] if the stream is broken by the server or + /// network. + /// + /// See the [official documentation](https://cloud.google.com/pubsub/docs/reference/rpc/google.pubsub.v1#google.pubsub.v1.Subscriber.StreamingPull). + Stream streamingPull({int streamAckDeadlineSeconds = 10}) { + if (streamAckDeadlineSeconds < 10 || streamAckDeadlineSeconds > 600) { + throw ArgumentError.value( + streamAckDeadlineSeconds, + 'streamAckDeadlineSeconds', + 'Must be between 10 and 600 seconds', + ); + } + return pubsub.streamingPull( + name, + streamAckDeadlineSeconds: streamAckDeadlineSeconds, + ); + } + + /// Acknowledges the messages associated with the [messages] immediately. + /// + /// The Pub/Sub system can remove the relevant messages from the subscription. + /// + /// Acknowledging a message whose ack deadline has expired may succeed, + /// but such a message may be redelivered later. Acknowledging a message more + /// than once will not result in an error. + /// + /// Throws a [SubscriptionNotFoundException] if the subscription does not + /// exist. + /// + /// See the [official documentation](https://cloud.google.com/pubsub/docs/reference/rpc/google.pubsub.v1#google.pubsub.v1.Subscriber.Acknowledge). + Future acknowledgeNow(List messages) => + pubsub.acknowledge(name, messages.map((m) => m.ackId).toList()); + + /// Acknowledges a single message in the background. + /// + /// This operation is "fire-and-forget" and does not wait for server + /// confirmation. If you require explicit confirmation of acknowledgment, + /// use [acknowledgeNow]. + void acknowledge(ReceivedMessage message) { + unawaited(acknowledgeNow([message])); + } + + /// Modifies the ack deadline for a list of specific messages immediately. + /// + /// This method is useful to indicate that more time is needed to process a + /// message by the subscriber, or to make the message available for redelivery + /// if the processing was interrupted. Note that this does not modify the + /// subscription-level `ackDeadlineSeconds` used for subsequent messages. + /// + /// Modifying the ack deadline for messages whose deadline has already expired + /// may succeed, but those messages may have already been redelivered or + /// made available for redelivery. + /// + /// Throws a [SubscriptionNotFoundException] if the subscription does not + /// exist. + /// + /// It is an error if [ackDeadlineSeconds] is negative. + /// + /// See the [official documentation](https://cloud.google.com/pubsub/docs/reference/rpc/google.pubsub.v1#google.pubsub.v1.Subscriber.ModifyAckDeadline). + Future modifyAckDeadlineNow( + List messages, + int ackDeadlineSeconds, + ) { + if (ackDeadlineSeconds < 0) { + throw ArgumentError.value( + ackDeadlineSeconds, + 'ackDeadlineSeconds', + 'Must be non-negative', + ); + } + return pubsub.modifyAckDeadline( + name, + messages.map((m) => m.ackId).toList(), + ackDeadlineSeconds, + ); + } + + /// Modifies the ack deadline for a single message in the background. + /// + /// This operation is "fire-and-forget" and does not wait for server + /// confirmation. If you require explicit confirmation, use + /// [modifyAckDeadlineNow]. + /// + /// It is an error if [ackDeadlineSeconds] is negative. + void modifyAckDeadline(ReceivedMessage message, int ackDeadlineSeconds) { + if (ackDeadlineSeconds < 0) { + throw ArgumentError.value( + ackDeadlineSeconds, + 'ackDeadlineSeconds', + 'Must be non-negative', + ); + } + unawaited(modifyAckDeadlineNow([message], ackDeadlineSeconds)); + } +} diff --git a/pkgs/google_cloud_pubsub/lib/src/topic.dart b/pkgs/google_cloud_pubsub/lib/src/topic.dart new file mode 100644 index 00000000..4cad06cf --- /dev/null +++ b/pkgs/google_cloud_pubsub/lib/src/topic.dart @@ -0,0 +1,100 @@ +// 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 '../google_cloud_pubsub.dart'; + +/// A [Google Cloud Pub/Sub topic](https://cloud.google.com/pubsub/docs/overview#topics). +final class Topic { + static final RegExp _topicNameRegExp = RegExp( + r'^projects/[^/]+/topics/[^/]+$', + ); + + /// The [PubSub] client associated with this topic. + final PubSub pubsub; + + /// The fully qualified resource name of this topic. + /// + /// It has the format `projects//topics/`. + final String name; + + /// A topic with the given [topicId] in the client's project. + /// + /// It is an error if the constructed topic name is invalid (e.g. if [topicId] + /// contains slashes). + Topic.unqualified(this.pubsub, String topicId) + : name = 'projects/${pubsub.projectId}/topics/$topicId' { + _validateName(name); + } + + /// A topic with the given [name]. + /// + /// Useful for cross-project access. + /// + /// It is an error if [name] is not in the format + /// `projects//topics/`. + Topic(this.pubsub, this.name) { + _validateName(name); + } + + static void _validateName(String name) { + if (!_topicNameRegExp.hasMatch(name)) { + throw ArgumentError.value( + name, + 'name', + 'Must be in the format projects//topics/', + ); + } + } + + /// The unqualified ID of this topic. + String get id => name.split('/').last; + + /// Creates this topic on the server. + /// + /// The topic must not already exist on the server. + /// + /// A topic must exist on the server before you can publish messages to it + /// or create subscriptions for it. + /// + /// Throws a [TopicAlreadyExistsException] if the topic already exists. + /// + /// Returns a [Topic] instance representing the created topic. + /// + /// See the [official documentation](https://cloud.google.com/pubsub/docs/reference/rpc/google.pubsub.v1#google.pubsub.v1.Publisher.CreateTopic). + Future create() => pubsub.createTopic(name); + + /// Deletes this topic on the server. + /// + /// Throws a [TopicNotFoundException] if the topic does not exist. + /// + /// After a topic is deleted, a new topic may be created with the same name; + /// this is an entirely new topic with none of the old configuration or + /// subscriptions. Existing subscriptions to this topic are not deleted, but + /// their `topic` field is set to `_deleted-topic_`. + /// + /// See the [official documentation](https://cloud.google.com/pubsub/docs/reference/rpc/google.pubsub.v1#google.pubsub.v1.Publisher.DeleteTopic). + Future delete() => pubsub.deleteTopic(name); + + /// Adds one or more messages to the topic. + /// + /// Throws a [TopicNotFoundException] if the topic does not exist. + /// + /// [data] is the message content. + /// [attributes] are optional attributes for the message. + /// + /// See the [official documentation](https://cloud.google.com/pubsub/docs/reference/rpc/google.pubsub.v1#google.pubsub.v1.Publisher.Publish). + // TODO(sigurdm): Support batch publishing (publishMany) for high-throughput. + Future publish(List data, {Map? attributes}) => + pubsub.publish(name, data, attributes: attributes); +} diff --git a/pkgs/google_cloud_pubsub/pubspec.yaml b/pkgs/google_cloud_pubsub/pubspec.yaml index 084932c4..ac2c26f2 100644 --- a/pkgs/google_cloud_pubsub/pubspec.yaml +++ b/pkgs/google_cloud_pubsub/pubspec.yaml @@ -11,11 +11,18 @@ environment: resolution: workspace dependencies: + collection: ^1.19.1 + ffi: ^2.2.0 fixnum: ^1.1.1 + google_cloud: '>=0.3.0 <0.6.0' + google_cloud_protobuf: ^0.5.0 + google_cloud_rpc: ^0.5.0 + googleapis_auth: ^2.0.0 grpc: ^5.1.0 + http: ^1.6.0 + meta: ^1.18.1 protobuf: ^6.0.0 dev_dependencies: - http: ^1.6.0 protoc_plugin: ^25.0.0 test: ^1.24.0 diff --git a/pkgs/google_cloud_pubsub/test/acknowledge_test.dart b/pkgs/google_cloud_pubsub/test/acknowledge_test.dart new file mode 100644 index 00000000..9df1c9b9 --- /dev/null +++ b/pkgs/google_cloud_pubsub/test/acknowledge_test.dart @@ -0,0 +1,55 @@ +// 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. + +@TestOn('vm') +@Tags(['firebase-emulator']) +library; + +import 'package:google_cloud_pubsub/google_cloud_pubsub.dart'; +import 'package:test/test.dart'; + +import 'test_utils.dart'; + +void main() { + group('acknowledge', () { + late PubSub client; + + setUp(() async { + client = await createClient(); + }); + + tearDown(() async { + await client.close(); + }); + + test('acknowledge for non-existent subscription throws ' + 'SubscriptionNotFoundException', () async { + final subscriptionName = + 'non-existent-${DateTime.now().millisecondsSinceEpoch}'; + final subscription = client.subscription(subscriptionName); + + expect( + () => subscription.acknowledgeNow([ + ReceivedMessage( + ackId: 'ack-id', + messageId: 'msg-id', + publishTime: DateTime.now(), + message: Message(data: []), + ), + ]), + throwsA(isA()), + ); + }); + }); +} diff --git a/pkgs/google_cloud_pubsub/test/create_subscription_test.dart b/pkgs/google_cloud_pubsub/test/create_subscription_test.dart new file mode 100644 index 00000000..c619fb21 --- /dev/null +++ b/pkgs/google_cloud_pubsub/test/create_subscription_test.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. + +@TestOn('vm') +@Tags(['firebase-emulator']) +library; + +import 'package:google_cloud_pubsub/google_cloud_pubsub.dart'; +import 'package:test/test.dart'; + +import 'test_utils.dart'; + +void main() { + group('createSubscription', () { + late PubSub client; + + setUp(() async { + client = await createClient(); + }); + + tearDown(() async { + await client.close(); + }); + + test( + 'create existing subscription throws SubscriptionAlreadyExistsException', + () async { + final topicName = 'test-topic-${DateTime.now().millisecondsSinceEpoch}'; + final subscriptionName = + 'test-sub-${DateTime.now().millisecondsSinceEpoch}'; + final topic = client.topic(topicName); + final subscription = client.subscription(subscriptionName); + + await topic.create(); + addTearDown(() async => await topic.delete()); + await subscription.create(topic: topic.name); + addTearDown(() async => await subscription.delete()); + + expect( + () => subscription.create(topic: topic.name), + throwsA(isA()), + ); + + await subscription.delete(); + await topic.delete(); + }, + retry: isEmulator ? 3 : 0, + ); + + test('create subscription for non-existent topic throws ' + 'TopicNotFoundException', () async { + final subscriptionName = + 'test-sub-${DateTime.now().millisecondsSinceEpoch}'; + final subscription = client.subscription(subscriptionName); + + expect( + () => + subscription.create(topic: client.topic('non-existent-topic').name), + throwsA(isA()), + ); + }); + }); +} diff --git a/pkgs/google_cloud_pubsub/test/create_topic_test.dart b/pkgs/google_cloud_pubsub/test/create_topic_test.dart new file mode 100644 index 00000000..0a7ddb97 --- /dev/null +++ b/pkgs/google_cloud_pubsub/test/create_topic_test.dart @@ -0,0 +1,62 @@ +// 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. + +@TestOn('vm') +@Tags(['firebase-emulator']) +library; + +import 'package:google_cloud_pubsub/google_cloud_pubsub.dart'; +import 'package:test/test.dart'; + +import 'test_utils.dart'; + +void main() { + group('createTopic', () { + late PubSub client; + + setUp(() async { + client = await createClient(); + }); + + tearDown(() async { + await client.close(); + }); + + test('create topic', () async { + final topicName = 'test-topic-${DateTime.now().millisecondsSinceEpoch}'; + final topic = client.topic(topicName); + + await topic.create(); + addTearDown(() async => await topic.delete()); + + await topic.delete(); + }); + + test( + 'create existing topic throws TopicAlreadyExistsException', + () async { + final topicName = 'test-topic-${DateTime.now().millisecondsSinceEpoch}'; + final topic = client.topic(topicName); + + await topic.create(); + addTearDown(() async => await topic.delete()); + + expect(topic.create(), throwsA(isA())); + + await topic.delete(); + }, + retry: isEmulator ? 3 : 0, + ); + }); +} diff --git a/pkgs/google_cloud_pubsub/test/delete_topic_test.dart b/pkgs/google_cloud_pubsub/test/delete_topic_test.dart new file mode 100644 index 00000000..b94e28ee --- /dev/null +++ b/pkgs/google_cloud_pubsub/test/delete_topic_test.dart @@ -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. + +@TestOn('vm') +@Tags(['firebase-emulator']) +library; + +import 'package:google_cloud_pubsub/google_cloud_pubsub.dart'; +import 'package:test/test.dart'; + +import 'test_utils.dart'; + +void main() { + group('deleteTopic', () { + late PubSub client; + + setUp(() async { + client = await createClient(); + }); + + tearDown(() async { + await client.close(); + }); + + test('Delete non-existent topic throws TopicNotFoundException', () async { + final topic = client.topic( + 'non-existent-${DateTime.now().millisecondsSinceEpoch}', + ); + expect(topic.delete(), throwsA(isA())); + }); + }); +} diff --git a/pkgs/google_cloud_pubsub/test/error_propagation_test.dart b/pkgs/google_cloud_pubsub/test/error_propagation_test.dart new file mode 100644 index 00000000..ee00a0a1 --- /dev/null +++ b/pkgs/google_cloud_pubsub/test/error_propagation_test.dart @@ -0,0 +1,323 @@ +// 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. + +@TestOn('vm') +library; + +import 'dart:async'; + +import 'package:google_cloud_pubsub/google_cloud_pubsub.dart'; +import 'package:google_cloud_pubsub/src/generated/google/pubsub/v1/pubsub.pb.dart' + as pb; +import 'package:google_cloud_pubsub/src/generated/google/pubsub/v1/pubsub.pbgrpc.dart' + as generated; +import 'package:grpc/grpc.dart' as grpc; +import 'package:protobuf/well_known_types/google/protobuf/empty.pb.dart' + as protobuf; +import 'package:protobuf/well_known_types/google/protobuf/timestamp.pb.dart' + as pb_ts; +import 'package:test/fake.dart'; +import 'package:test/test.dart'; + +// A fake ResponseFuture that delegates to a standard Future. +class FakeResponseFuture extends Fake implements grpc.ResponseFuture { + final Future _future; + + FakeResponseFuture(this._future); + + @override + Stream asStream() => _future.asStream(); + + @override + Future catchError(Function onError, {bool Function(Object)? test}) => + _future.catchError(onError, test: test); + + @override + Future then( + FutureOr Function(T value) onValue, { + Function? onError, + }) => _future.then( + onValue, + onError: (Object e, StackTrace s) { + if (onError != null) { + if (onError is FutureOr Function(Object, StackTrace)) { + onError(e, s); + } else if (onError is FutureOr Function(Object)) { + onError(e); + } else { + // ignore: avoid_dynamic_calls + (onError as dynamic)(e, s); + } + } + }, + ); + + @override + Future timeout(Duration timeLimit, {FutureOr Function()? onTimeout}) => + _future.timeout(timeLimit, onTimeout: onTimeout); + + @override + Future whenComplete(FutureOr Function() action) => + _future.whenComplete(action); + + @override + Future cancel() => Future.value(); +} + +// A fake ResponseStream that delegates to a standard Stream. +class FakeResponseStream extends StreamView + implements grpc.ResponseStream { + FakeResponseStream(super.stream); + + @override + grpc.ResponseFuture get single => FakeResponseFuture(super.single); + + @override + Future cancel() => Future.value(); + + @override + Future> get headers => Future.value(const {}); + + @override + Future> get trailers => Future.value(const {}); +} + +class FakeSubscriberClient extends Fake implements generated.SubscriberClient { + final StreamController + streamingPullController = StreamController(); + + bool acknowledgeCalled = false; + List? lastAckIds; + Future Function(List ackIds)? acknowledgeBehavior; + + bool modifyAckDeadlineCalled = false; + List? lastModifyAckDeadlineIds; + int? lastModifyAckDeadlineSeconds; + Future Function(List ackIds, int seconds)? + modifyAckDeadlineBehavior; + + @override + grpc.ResponseStream streamingPull( + Stream request, { + grpc.CallOptions? options, + }) { + // Listen to request stream to prevent sender from hanging on close() + unawaited(request.drain()); + return FakeResponseStream(streamingPullController.stream); + } + + @override + grpc.ResponseFuture acknowledge( + generated.AcknowledgeRequest request, { + grpc.CallOptions? options, + }) { + acknowledgeCalled = true; + lastAckIds = request.ackIds; + + final completer = Completer(); + if (acknowledgeBehavior case final acknowledge?) { + acknowledge(request.ackIds) + .then((_) => completer.complete(protobuf.Empty())) + .catchError(completer.completeError); + } else { + completer.complete(protobuf.Empty()); + } + return FakeResponseFuture(completer.future); + } + + @override + grpc.ResponseFuture modifyAckDeadline( + generated.ModifyAckDeadlineRequest request, { + grpc.CallOptions? options, + }) { + modifyAckDeadlineCalled = true; + lastModifyAckDeadlineIds = request.ackIds; + lastModifyAckDeadlineSeconds = request.ackDeadlineSeconds; + + final completer = Completer(); + if (modifyAckDeadlineBehavior case final modifyAckDeadline?) { + modifyAckDeadline(request.ackIds, request.ackDeadlineSeconds) + .then((_) => completer.complete(protobuf.Empty())) + .catchError(completer.completeError); + } else { + completer.complete(protobuf.Empty()); + } + return FakeResponseFuture(completer.future); + } +} + +class FakeClientChannel extends Fake implements grpc.ClientChannel { + @override + Future shutdown() async { + // No-op for testing + } +} + +void main() { + group('PubSub Unit Tests (Error Propagation)', () { + late FakeSubscriberClient fakeSubscriber; + late PubSub client; + + setUp(() { + fakeSubscriber = FakeSubscriberClient(); + client = PubSub.testing( + projectId: 'test-project', + channel: FakeClientChannel(), + subscriberClient: fakeSubscriber, + ); + }); + + tearDown(() async { + await client.close(); + }); + + test('streamingPull ack error propagates', () async { + final subscription = client.subscription('sub'); + final stream = subscription.streamingPull(); + + // Configure acknowledge to fail + fakeSubscriber.acknowledgeBehavior = (ackIds) async { + throw const grpc.GrpcError.notFound('Subscription not found'); + }; + + // Push a fake message to the stream + Timer(const Duration(milliseconds: 10), () { + final fakeResponse = generated.StreamingPullResponse() + ..receivedMessages.add( + generated.ReceivedMessage() + ..ackId = 'ack-1' + ..message = (pb.PubsubMessage()..messageId = 'msg-1'), + ); + fakeSubscriber.streamingPullController.add(fakeResponse); + }); + + final receivedMessage = await stream.first; + expect(receivedMessage.ackId, equals('ack-1')); + + final ackFuture = subscription.acknowledgeNow([receivedMessage]); + + await expectLater( + ackFuture, + throwsA( + isA().having( + (e) => e.name, + 'name', + 'projects/test-project/subscriptions/sub', + ), + ), + ); + + expect(fakeSubscriber.acknowledgeCalled, isTrue); + expect(fakeSubscriber.lastAckIds, equals(['ack-1'])); + }); + + test('streamingPull modifyAckDeadline error propagates', () async { + final subscription = client.subscription('sub'); + final stream = subscription.streamingPull(); + + // Configure modifyAckDeadline to fail + fakeSubscriber.modifyAckDeadlineBehavior = (ackIds, seconds) async { + throw const grpc.GrpcError.notFound('Subscription not found'); + }; + + // Push a fake message to the stream + Timer(const Duration(milliseconds: 10), () { + final fakeResponse = generated.StreamingPullResponse() + ..receivedMessages.add( + generated.ReceivedMessage() + ..ackId = 'ack-2' + ..message = (pb.PubsubMessage()..messageId = 'msg-2'), + ); + fakeSubscriber.streamingPullController.add(fakeResponse); + }); + + final receivedMessage = await stream.first; + expect(receivedMessage.ackId, equals('ack-2')); + + final modifyDeadlineFuture = subscription.modifyAckDeadlineNow([ + receivedMessage, + ], 10); + + await expectLater( + modifyDeadlineFuture, + throwsA( + isA().having( + (e) => e.name, + 'name', + 'projects/test-project/subscriptions/sub', + ), + ), + ); + + expect(fakeSubscriber.modifyAckDeadlineCalled, isTrue); + expect(fakeSubscriber.lastModifyAckDeadlineIds, equals(['ack-2'])); + expect(fakeSubscriber.lastModifyAckDeadlineSeconds, equals(10)); + }); + + test('message mapping works correctly', () async { + final stream = client.subscription('sub').streamingPull(); + + final publishTime = DateTime.utc(2026, 6, 9, 12, 0, 0); + + Timer(const Duration(milliseconds: 10), () { + final fakeResponse = generated.StreamingPullResponse() + ..receivedMessages.add( + generated.ReceivedMessage() + ..ackId = 'ack-1' + ..message = (pb.PubsubMessage() + ..messageId = 'msg-1' + ..publishTime = pb_ts.Timestamp.fromDateTime(publishTime) + ..data = [1, 2, 3] + ..attributes.addAll({'key': 'value'})), + ); + fakeSubscriber.streamingPullController.add(fakeResponse); + }); + + final receivedMessage = await stream.first; + expect(receivedMessage.ackId, equals('ack-1')); + expect(receivedMessage.messageId, equals('msg-1')); + expect(receivedMessage.publishTime, equals(publishTime)); + + // Test delegation + expect(receivedMessage.data, equals([1, 2, 3])); + expect(receivedMessage.attributes, equals({'key': 'value'})); + + // Test message composition + expect(receivedMessage.message.data, equals([1, 2, 3])); + expect(receivedMessage.message.attributes, equals({'key': 'value'})); + }); + }); + + group('ReceivedMessage composition and delegation', () { + test('properties are correctly mapped and delegated', () { + final publishTime = DateTime.now(); + final message = Message(data: [1, 2, 3], attributes: {'key': 'value'}); + final receivedMessage = ReceivedMessage( + ackId: 'ack-123', + messageId: 'msg-456', + publishTime: publishTime, + message: message, + ); + + expect(receivedMessage.ackId, equals('ack-123')); + expect(receivedMessage.messageId, equals('msg-456')); + expect(receivedMessage.publishTime, equals(publishTime)); + expect(receivedMessage.message, equals(message)); + + // Delegation getters + expect(receivedMessage.data, equals([1, 2, 3])); + expect(receivedMessage.attributes, equals({'key': 'value'})); + }); + }); +} diff --git a/pkgs/google_cloud_pubsub/test/modify_ack_deadline_test.dart b/pkgs/google_cloud_pubsub/test/modify_ack_deadline_test.dart new file mode 100644 index 00000000..64a37c33 --- /dev/null +++ b/pkgs/google_cloud_pubsub/test/modify_ack_deadline_test.dart @@ -0,0 +1,55 @@ +// 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. + +@TestOn('vm') +@Tags(['firebase-emulator']) +library; + +import 'package:google_cloud_pubsub/google_cloud_pubsub.dart'; +import 'package:test/test.dart'; + +import 'test_utils.dart'; + +void main() { + group('modifyAckDeadline', () { + late PubSub client; + + setUp(() async { + client = await createClient(); + }); + + tearDown(() async { + await client.close(); + }); + + test('modifyAckDeadline for non-existent subscription throws ' + 'SubscriptionNotFoundException', () async { + final subscriptionName = + 'non-existent-${DateTime.now().millisecondsSinceEpoch}'; + final subscription = client.subscription(subscriptionName); + + expect( + () => subscription.modifyAckDeadlineNow([ + ReceivedMessage( + ackId: 'ack-id', + messageId: 'msg-id', + publishTime: DateTime.now(), + message: Message(data: []), + ), + ], 10), + throwsA(isA()), + ); + }); + }); +} diff --git a/pkgs/google_cloud_pubsub/test/publish_test.dart b/pkgs/google_cloud_pubsub/test/publish_test.dart new file mode 100644 index 00000000..2ad224bd --- /dev/null +++ b/pkgs/google_cloud_pubsub/test/publish_test.dart @@ -0,0 +1,150 @@ +// 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. + +@TestOn('vm') +@Tags(['firebase-emulator']) +library; + +import 'dart:convert'; + +import 'package:google_cloud_pubsub/google_cloud_pubsub.dart'; +import 'package:test/test.dart'; + +import 'test_utils.dart'; + +void main() { + group('publish', () { + late PubSub client; + + setUp(() async { + client = await createClient(); + }); + + tearDown(() async { + await client.close(); + }); + + test( + 'publish to non-existent topic throws TopicNotFoundException', + () async { + final topicName = + 'non-existent-${DateTime.now().millisecondsSinceEpoch}'; + final topic = client.topic(topicName); + + expect( + () => topic.publish(utf8.encode('Hello')), + throwsA(isA()), + ); + }, + ); + + test('publish and pull message', () async { + final topicName = 'test-topic-${DateTime.now().millisecondsSinceEpoch}'; + final subscriptionName = + 'test-sub-${DateTime.now().millisecondsSinceEpoch}'; + final topic = client.topic(topicName); + final subscription = client.subscription(subscriptionName); + + await topic.create(); + addTearDown(() async => await topic.delete()); + + await subscription.create(topic: topic.name); + addTearDown(() async => await subscription.delete()); + + final data = utf8.encode('Hello PubSub'); + await topic.publish(data); + + final messages = await pullReliably(subscription, count: 1); + expect(messages, hasLength(1)); + expect(messages.first.data, equals(data)); + + await subscription.acknowledgeNow([messages.first]); + }); + + test('publish and streaming pull message', () async { + final topicName = 'test-topic-${DateTime.now().millisecondsSinceEpoch}'; + final subscriptionName = + 'test-sub-${DateTime.now().millisecondsSinceEpoch}'; + final topic = client.topic(topicName); + final subscription = client.subscription(subscriptionName); + + await topic.create(); + addTearDown(() async => await topic.delete()); + + await subscription.create(topic: topic.name); + addTearDown(() async => await subscription.delete()); + + final data = utf8.encode('Hello Streaming PubSub'); + await topic.publish(data); + + final stream = subscription.streamingPull(); + final receivedMessage = await stream.first; + + expect(receivedMessage.data, equals(data)); + await subscription.acknowledgeNow([receivedMessage]); + }); + + test('publish and pull message with attributes', () async { + final topicName = 'test-topic-${DateTime.now().millisecondsSinceEpoch}'; + final subscriptionName = + 'test-sub-${DateTime.now().millisecondsSinceEpoch}'; + final topic = client.topic(topicName); + final subscription = client.subscription(subscriptionName); + + await topic.create(); + addTearDown(() async => await topic.delete()); + + await subscription.create(topic: topic.name); + addTearDown(() async => await subscription.delete()); + + final data = utf8.encode('Hello PubSub with Attributes'); + final attributes = {'key1': 'value1', 'key2': 'value2'}; + await topic.publish(data, attributes: attributes); + + final messages = await pullReliably(subscription, count: 1); + expect(messages, hasLength(1)); + expect(messages.first.data, equals(data)); + expect(messages.first.attributes, equals(attributes)); + + await subscription.acknowledgeNow([messages.first]); + }); + + test('publish multiple messages and pull batch', () async { + final topicName = 'test-topic-${DateTime.now().millisecondsSinceEpoch}'; + final subscriptionName = + 'test-sub-${DateTime.now().millisecondsSinceEpoch}'; + final topic = client.topic(topicName); + final subscription = client.subscription(subscriptionName); + + await topic.create(); + addTearDown(() async => await topic.delete()); + + await subscription.create(topic: topic.name); + addTearDown(() async => await subscription.delete()); + + final data1 = utf8.encode('Message 1'); + final data2 = utf8.encode('Message 2'); + await topic.publish(data1); + await topic.publish(data2); + + final messages = await pullReliably(subscription, count: 2); + expect(messages, hasLength(2)); + + final receivedData = messages.map((m) => m.data).toList(); + expect(receivedData, containsAll([data1, data2])); + + await subscription.acknowledgeNow(messages); + }); + }); +} diff --git a/pkgs/google_cloud_pubsub/test/pull_test.dart b/pkgs/google_cloud_pubsub/test/pull_test.dart new file mode 100644 index 00000000..182aae0d --- /dev/null +++ b/pkgs/google_cloud_pubsub/test/pull_test.dart @@ -0,0 +1,45 @@ +// 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. + +@TestOn('vm') +@Tags(['firebase-emulator']) +library; + +import 'package:google_cloud_pubsub/google_cloud_pubsub.dart'; +import 'package:test/test.dart'; + +import 'test_utils.dart'; + +void main() { + group('pull', () { + late PubSub client; + + setUp(() async { + client = await createClient(); + }); + + tearDown(() async { + await client.close(); + }); + + test('pull from non-existent subscription throws ' + 'SubscriptionNotFoundException', () async { + final subscriptionName = + 'non-existent-${DateTime.now().millisecondsSinceEpoch}'; + final subscription = client.subscription(subscriptionName); + + expect(subscription.pull, throwsA(isA())); + }); + }); +} diff --git a/pkgs/google_cloud_pubsub/test/streaming_pull_test.dart b/pkgs/google_cloud_pubsub/test/streaming_pull_test.dart new file mode 100644 index 00000000..e9cccbd2 --- /dev/null +++ b/pkgs/google_cloud_pubsub/test/streaming_pull_test.dart @@ -0,0 +1,57 @@ +// 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. + +@TestOn('vm') +@Tags(['firebase-emulator']) +library; + +import 'package:google_cloud_pubsub/google_cloud_pubsub.dart'; +import 'package:test/test.dart'; + +import 'test_utils.dart'; + +void main() { + group('streamingPull', () { + late PubSub client; + + setUp(() async { + client = await createClient(); + }); + + tearDown(() async { + await client.close(); + }); + + test('streaming pull throws StreamBrokenException ' + 'when subscription is deleted', () async { + final topicName = 'test-topic-${DateTime.now().millisecondsSinceEpoch}'; + final subscriptionName = + 'test-sub-${DateTime.now().millisecondsSinceEpoch}'; + final topic = client.topic(topicName); + final subscription = client.subscription(subscriptionName); + + await topic.create(); + addTearDown(() async => await topic.delete()); + + await subscription.create(topic: topic.name); + + final stream = subscription.streamingPull(); + + // Delete subscription to break the stream + await subscription.delete(); + + expect(stream.first, throwsA(isA())); + }); + }); +} diff --git a/pkgs/google_cloud_pubsub/test/test_utils.dart b/pkgs/google_cloud_pubsub/test/test_utils.dart new file mode 100644 index 00000000..4c907e9b --- /dev/null +++ b/pkgs/google_cloud_pubsub/test/test_utils.dart @@ -0,0 +1,69 @@ +// 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:io'; + +import 'package:google_cloud_pubsub/google_cloud_pubsub.dart'; +import 'package:test/test.dart'; + +final isEmulator = Platform.environment['PUBSUB_EMULATOR_HOST'] != null; + +/// Creates a [PubSub] client configured for either the emulator or production +/// based on environment variables. +Future createClient() async { + final host = Platform.environment['PUBSUB_EMULATOR_HOST']; + final project = Platform.environment['GOOGLE_CLOUD_PROJECT']; + + if (host != null) { + return PubSub(projectId: 'test-project'); + } else if (project != null) { + return PubSub( + projectId: project, + authenticator: await applicationDefaultCredentialsAuthenticator([ + 'https://www.googleapis.com/auth/pubsub', + ]), + ); + } else { + fail( + 'Neither PUBSUB_EMULATOR_HOST nor GOOGLE_CLOUD_PROJECT ' + 'environment variable set', + ); + } +} + +/// Pulls up to [count] messages from the [subscription], retrying up to 10 +/// times with a 1-second delay between attempts if the expected count is +/// not met. +/// +/// Because message delivery in Google Cloud Pub/Sub is eventually consistent, +/// a published message may not be immediately available in the first pull +/// request. Polling dynamically like this avoids both test flakiness (by +/// waiting up to 10 seconds) and unnecessary test suite delay (by returning +/// instantly as soon as all messages are retrieved). +Future> pullReliably( + Subscription subscription, { + required int count, +}) async { + final messages = []; + for (var i = 0; i < 10 && messages.length < count; i++) { + if (i > 0) { + await Future.delayed(const Duration(seconds: 1)); + } + final pulled = await subscription.pull( + maxMessages: count - messages.length, + ); + messages.addAll(pulled); + } + return messages; +} diff --git a/pkgs/google_cloud_pubsub/tool/compile_protos.dart b/pkgs/google_cloud_pubsub/tool/compile_protos.dart index 1ab28204..65fe73b6 100644 --- a/pkgs/google_cloud_pubsub/tool/compile_protos.dart +++ b/pkgs/google_cloud_pubsub/tool/compile_protos.dart @@ -1,3 +1,17 @@ +// 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:io'; void main() async { diff --git a/pkgs/google_cloud_pubsub/tool/fetch_protos.dart b/pkgs/google_cloud_pubsub/tool/fetch_protos.dart index df037544..a7e3abe8 100644 --- a/pkgs/google_cloud_pubsub/tool/fetch_protos.dart +++ b/pkgs/google_cloud_pubsub/tool/fetch_protos.dart @@ -1,3 +1,17 @@ +// 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:io'; import 'package:http/http.dart' as http;