feat: Pub/sub

[sigurdm]

Initial implementation. Only basic functionality implemented.
Streaming pull is implemented.

No built-in retry yet.

Tools to fetch and compile protos.

[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.

[gemini-code-assist]

There are several issues with the current authentication logic:

1. Race Condition: Multiple concurrent RPCs could trigger multiple calls to clientViaApplicationDefaultCredentials before _authClient is set.
2. Token Refresh: AutoRefreshingAuthClient only refreshes tokens when used via its HTTP interface. Since this code manually extracts the token string for gRPC, the token will eventually expire and never be refreshed. You should call refreshCredentials() to ensure a valid token.
3. Resource Leak: The _authClient is never closed. It should be closed in the close() method of the PubSub class.

[gemini-code-assist]

The current logic for splitting the host and port fails for IPv6 addresses (e.g., [::1]:8085). Using lastIndexOf(':') and checking for the closing bracket ] is a more robust approach for parsing the port.

Suggested change

	if (host.contains(':')) {
	final parts = host.split(':');
	cleanHost = parts[0];
	port = int.parse(parts[1]);
	}
	final lastColon = host.lastIndexOf(':');
	if (lastColon != -1 && !host.endsWith(']')) {
	cleanHost = host.substring(0, lastColon);
	port = int.parse(host.substring(lastColon + 1));
	}

[gemini-code-assist]

The _authClient (an AutoRefreshingAuthClient) should be closed when the PubSub client is closed to release underlying resources.

Suggested change

	Future<void> close() async {
	await _channel.shutdown();
	}
	Future<void> close() async {
	await _channel.shutdown();
	await _authClient?.close();
	}

[gemini-code-assist]

The logic for mapping a grpc.ReceivedMessage to the domain ReceivedMessage is duplicated here and in streamingPull. Consider extracting this into a private helper method for better maintainability.