diff --git a/docs/guide/messaging/transports/gcp-pubsub/index.md b/docs/guide/messaging/transports/gcp-pubsub/index.md index b3dc2d2b2..da1d073fe 100644 --- a/docs/guide/messaging/transports/gcp-pubsub/index.md +++ b/docs/guide/messaging/transports/gcp-pubsub/index.md @@ -64,6 +64,92 @@ opts.UsePubsub("your-project-id") The credential manages its own token refresh lifecycle, so no additional background task is required. For more control over the underlying GCP client builders, see [Customisation](/guide/messaging/transports/gcp-pubsub/customisation). +## Multiple / Named Brokers + +You can connect to more than one GCP Pub/Sub broker (typically a different GCP project) from a single Wolverine application by registering an additional, *named* broker alongside the default one. Endpoints are then pinned to the named broker with the `...OnNamedBroker` overloads: + + + +```cs +var host = await Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + // The default / shared Pub/Sub broker + opts.UsePubsub("your-project-id").AutoProvision(); + + // An additional, independent Pub/Sub broker pointed at a different GCP project. + // The Wolverine Uri scheme for endpoints on this broker becomes the broker name + // ("americas"), e.g. americas://americas-project-id/colors + opts.AddNamedPubsubBroker(new BrokerName("americas"), "americas-project-id") + .AutoProvision(); + + // Pin specific endpoints to the named broker + opts.PublishMessage() + .ToPubsubTopicOnNamedBroker(new BrokerName("americas"), "colors"); + opts.ListenToPubsubTopicOnNamedBroker(new BrokerName("americas"), "colors"); + }).StartAsync(); +``` +snippet source | anchor + + +Note that the `Uri` scheme within Wolverine for any endpoint on a *named* GCP Pub/Sub broker is the broker name you supply, not `pubsub`. So in the example above you would see `Uri` values like `americas://americas-project-id/colors`. + +## Multi-Tenancy with a Broker Per Tenant + +Named brokers (above) are a *static* topology: you pin specific endpoints to a specific broker at configuration time. **Broker-per-tenant** is different — it is *runtime* routing. You declare one shared topic topology, and each tenant is served by its **own dedicated GCP project**. Which project a message goes to (and which project an inbound message came from) is decided at runtime by the message's [tenant id](/guide/handlers/multi-tenancy), typically set through `DeliveryOptions.TenantId`. + +Project-id-per-tenant is the natural isolation axis for Pub/Sub: the topic and subscription names embed the project id, so the *same* logical topic under a *different* project is already a physically distinct Pub/Sub resource — "shared by name, isolated by project". + + + +```cs +var host = await Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + // The "default" / shared Pub/Sub connection on its own GCP project + opts.UsePubsub("shared-project-id") + .AutoProvision() + + // How should Wolverine route a message whose TenantId is null or + // unknown? FallbackToDefault (the default) uses the shared project; + // TenantIdRequired throws; IgnoreUnknownTenants silently drops it. + .TenantIdBehavior(TenantedIdBehavior.FallbackToDefault) + + // Each tenant is served by its OWN dedicated GCP project, but shares + // the topic topology declared below. Project-id-per-tenant is the + // natural isolation axis: the same logical topic under a different + // project is a physically distinct Pub/Sub resource. + .AddTenant("tenant1", "tenant1-project-id") + + // A tenant may also carry its own dedicated credentials by configuring + // its client builders (seeded from the parent transport otherwise): + .AddTenant("tenant2", "tenant2-project-id", tenant => + { + tenant.ConfigurePublisherApiBuilder = + builder => { /* builder.GoogleCredential = ...; */ return ValueTask.CompletedTask; }; + }); + + // One shared topology; messages are routed to the right project at runtime + // by Envelope.TenantId (e.g. new DeliveryOptions { TenantId = "tenant1" }). + opts.PublishMessage().ToPubsubTopic("colors"); + opts.ListenToPubsubTopic("colors"); + }).StartAsync(); +``` +snippet source | anchor + + +To route a specific message to a tenant's project, stamp the tenant id on the send: + +```csharp +await bus.SendAsync(new ColorMessage("blue"), new DeliveryOptions { TenantId = "tenant1" }); +``` + +Wolverine wraps the outbound endpoint in a `TenantedSender` that dispatches on `Envelope.TenantId`, and builds a compound listener that runs one listener per tenant project — each inbound envelope is stamped with the tenant id of the project it was consumed from. When `AutoProvision()` is enabled, Wolverine provisions the shared topology (topics and subscriptions) on **every** tenant project, not just the default one. + +::: tip The emulator caveat +The GCP Pub/Sub emulator ignores credentials and accepts arbitrary project ids with no auth, so per-tenant *projects* are trivially testable on a single emulator (just use distinct project id strings). Per-tenant *credentials*, however, cannot be exercised against the emulator — test that path against real GCP. +::: + ## Request/Reply [Request/reply](https://www.enterpriseintegrationpatterns.com/patterns/messaging/RequestReply.html) mechanics (`IMessageBus.InvokeAsync()`) are possible with the GCP Pub/Sub transport *if* Wolverine has the ability to auto-provision a specific response topic and subscription for each node. That topic and subscription would be named like `wlvrn.response.[application node id]` if you happen to notice that in your GCP Pub/Sub. diff --git a/src/Transports/GCP/Wolverine.Pubsub.Tests/DocumentationSamples.cs b/src/Transports/GCP/Wolverine.Pubsub.Tests/DocumentationSamples.cs index 6fc89c470..d2bf5c091 100644 --- a/src/Transports/GCP/Wolverine.Pubsub.Tests/DocumentationSamples.cs +++ b/src/Transports/GCP/Wolverine.Pubsub.Tests/DocumentationSamples.cs @@ -6,6 +6,7 @@ using Microsoft.Extensions.Hosting; using Wolverine.ComplianceTests.Compliance; using Wolverine.Transports; +using Wolverine.Transports.Sending; using Wolverine.Util; namespace Wolverine.Pubsub.Tests; @@ -61,6 +62,68 @@ public async Task enable_system_endpoints() #endregion } + public async Task named_pubsub_brokers() + { + #region sample_named_pubsub_broker + var host = await Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + // The default / shared Pub/Sub broker + opts.UsePubsub("your-project-id").AutoProvision(); + + // An additional, independent Pub/Sub broker pointed at a different GCP project. + // The Wolverine Uri scheme for endpoints on this broker becomes the broker name + // ("americas"), e.g. americas://americas-project-id/colors + opts.AddNamedPubsubBroker(new BrokerName("americas"), "americas-project-id") + .AutoProvision(); + + // Pin specific endpoints to the named broker + opts.PublishMessage() + .ToPubsubTopicOnNamedBroker(new BrokerName("americas"), "colors"); + opts.ListenToPubsubTopicOnNamedBroker(new BrokerName("americas"), "colors"); + }).StartAsync(); + + #endregion + } + + public async Task broker_per_tenant() + { + #region sample_pubsub_broker_per_tenant + var host = await Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + // The "default" / shared Pub/Sub connection on its own GCP project + opts.UsePubsub("shared-project-id") + .AutoProvision() + + // How should Wolverine route a message whose TenantId is null or + // unknown? FallbackToDefault (the default) uses the shared project; + // TenantIdRequired throws; IgnoreUnknownTenants silently drops it. + .TenantIdBehavior(TenantedIdBehavior.FallbackToDefault) + + // Each tenant is served by its OWN dedicated GCP project, but shares + // the topic topology declared below. Project-id-per-tenant is the + // natural isolation axis: the same logical topic under a different + // project is a physically distinct Pub/Sub resource. + .AddTenant("tenant1", "tenant1-project-id") + + // A tenant may also carry its own dedicated credentials by configuring + // its client builders (seeded from the parent transport otherwise): + .AddTenant("tenant2", "tenant2-project-id", tenant => + { + tenant.ConfigurePublisherApiBuilder = + builder => { /* builder.GoogleCredential = ...; */ return ValueTask.CompletedTask; }; + }); + + // One shared topology; messages are routed to the right project at runtime + // by Envelope.TenantId (e.g. new DeliveryOptions { TenantId = "tenant1" }). + opts.PublishMessage().ToPubsubTopic("colors"); + opts.ListenToPubsubTopic("colors"); + }).StartAsync(); + + #endregion + } + public async Task configuring_listeners() { #region sample_listen_to_pubsub_topic @@ -470,4 +533,5 @@ public async ValueTask GetAsync() => throw new NotImplementedException(); } -#endregion \ No newline at end of file +#endregion +public record ColorMessage(string Color); diff --git a/src/Transports/GCP/Wolverine.Pubsub.Tests/NamedBrokerComplianceTests.cs b/src/Transports/GCP/Wolverine.Pubsub.Tests/NamedBrokerComplianceTests.cs new file mode 100644 index 000000000..699dc5c9a --- /dev/null +++ b/src/Transports/GCP/Wolverine.Pubsub.Tests/NamedBrokerComplianceTests.cs @@ -0,0 +1,120 @@ +using Google.Api.Gax; +using JasperFx.Core; +using Microsoft.Extensions.Hosting; +using Shouldly; +using Wolverine.Runtime; +using Wolverine.Tracking; +using Xunit; + +namespace Wolverine.Pubsub.Tests; + +public class NamedBrokerComplianceTests +{ + // A named broker is a second, independent PubsubTransport whose Protocol (and therefore URI scheme) is the + // broker name, and which carries its own project id. No broker needed. + [Fact] + public void named_broker_is_a_distinct_transport_with_its_own_protocol_and_project() + { + var options = new WolverineOptions(); + + options.UsePubsub("wolverine"); + options.AddNamedPubsubBroker(new BrokerName("americas"), "wolverine2"); + + var defaultTransport = options.Transports.GetOrCreate(); + var named = options.Transports.OfType().Single(x => x.Protocol == "americas"); + + named.ShouldNotBeSameAs(defaultTransport); + defaultTransport.Protocol.ShouldBe(PubsubTransport.ProtocolName); + defaultTransport.ProjectId.ShouldBe("wolverine"); + named.Protocol.ShouldBe("americas"); + named.ProjectId.ShouldBe("wolverine2"); + } + + [Fact] + public void named_broker_endpoint_uri_scheme_is_the_broker_name() + { + var options = new WolverineOptions(); + + options.AddNamedPubsubBroker(new BrokerName("americas"), "wolverine2"); + options.ListenToPubsubTopicOnNamedBroker(new BrokerName("americas"), "colors"); + + var named = options.Transports.OfType().Single(x => x.Protocol == "americas"); + var endpoint = named.Topics["colors"]; + + endpoint.Uri.Scheme.ShouldBe("americas"); + endpoint.Uri.ShouldBe(new Uri("americas://wolverine2/colors")); + named.ResourceUri.ShouldBe(new Uri("americas://wolverine2")); + } +} + +// Integration: round-trip a message over a second project on the same emulator, addressed exclusively through the +// named broker. Skip-guarded when the emulator/Docker is unavailable. +public class NamedBrokerRoundTripTests : IAsyncLifetime +{ + private IHost? _host; + private bool _skip; + + public async Task InitializeAsync() + { + _skip = !await TestingExtensions.IsEmulatorAvailable(); + if (_skip) + { + return; + } + + _host = await Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + // Default/shared broker on project "wolverine"... + opts.UsePubsubTesting().AutoProvision().AutoPurgeOnStartup(); + + // ...plus a named broker on a second project "wolverine2" on the same emulator. + opts.AddNamedPubsubBrokerTesting(new BrokerName("americas"), "wolverine2").AutoProvision() + .AutoPurgeOnStartup(); + + opts.PublishMessage() + .ToPubsubTopicOnNamedBroker(new BrokerName("americas"), "named-broker-colors"); + opts.ListenToPubsubTopicOnNamedBroker(new BrokerName("americas"), "named-broker-colors"); + }).StartAsync(); + } + + public async Task DisposeAsync() + { + if (_host is not null) + { + await _host.StopAsync(); + _host.Dispose(); + } + } + + [Fact] + public async Task round_trips_a_message_through_the_named_broker() + { + if (_skip) + { + return; + } + + var message = new NamedBrokerMessage("blue"); + + var session = await _host!.TrackActivity() + .IncludeExternalTransports() + .Timeout(1.Minutes()) + .SendMessageAndWaitAsync(message); + + var received = session.Received.SingleEnvelope(); + received.Message.ShouldBeOfType().Color.ShouldBe("blue"); + + // The receiving endpoint lives on the named broker, so its Uri scheme is the broker name. + received.Destination!.Scheme.ShouldBe("americas"); + } +} + +public record NamedBrokerMessage(string Color); + +public static class NamedBrokerMessageHandler +{ + public static void Handle(NamedBrokerMessage message) + { + } +} diff --git a/src/Transports/GCP/Wolverine.Pubsub.Tests/PubsubConfigurationTests.cs b/src/Transports/GCP/Wolverine.Pubsub.Tests/PubsubConfigurationTests.cs index c874e7aa9..99f592ace 100644 --- a/src/Transports/GCP/Wolverine.Pubsub.Tests/PubsubConfigurationTests.cs +++ b/src/Transports/GCP/Wolverine.Pubsub.Tests/PubsubConfigurationTests.cs @@ -12,7 +12,7 @@ public class PubsubConfigurationTests [Fact] public async Task configure_publisher_api_client_sets_callback_on_transport() { - var transport = new PubsubTransport("test"); + var transport = new PubsubTransport { ProjectId = "test" }; var config = new PubsubConfiguration(transport, new WolverineOptions()); var called = false; @@ -26,7 +26,7 @@ public async Task configure_publisher_api_client_sets_callback_on_transport() [Fact] public async Task configure_subscriber_api_client_sets_callback_on_transport() { - var transport = new PubsubTransport("test"); + var transport = new PubsubTransport { ProjectId = "test" }; var config = new PubsubConfiguration(transport, new WolverineOptions()); var called = false; @@ -40,7 +40,7 @@ public async Task configure_subscriber_api_client_sets_callback_on_transport() [Fact] public async Task configure_subscriber_client_sets_callback_on_transport() { - var transport = new PubsubTransport("test"); + var transport = new PubsubTransport { ProjectId = "test" }; var config = new PubsubConfiguration(transport, new WolverineOptions()); var called = false; @@ -54,7 +54,7 @@ public async Task configure_subscriber_client_sets_callback_on_transport() [Fact] public async Task multiple_configure_publisher_api_client_calls_compose_in_order() { - var transport = new PubsubTransport("test"); + var transport = new PubsubTransport { ProjectId = "test" }; var config = new PubsubConfiguration(transport, new WolverineOptions()); var order = new List(); @@ -68,7 +68,7 @@ public async Task multiple_configure_publisher_api_client_calls_compose_in_order [Fact] public async Task multiple_configure_subscriber_client_calls_compose_in_order() { - var transport = new PubsubTransport("test"); + var transport = new PubsubTransport { ProjectId = "test" }; var config = new PubsubConfiguration(transport, new WolverineOptions()); var order = new List(); @@ -82,7 +82,7 @@ public async Task multiple_configure_subscriber_client_calls_compose_in_order() [Fact] public async Task async_configure_publisher_api_client_callback_is_awaited() { - var transport = new PubsubTransport("test"); + var transport = new PubsubTransport { ProjectId = "test" }; var config = new PubsubConfiguration(transport, new WolverineOptions()); var called = false; @@ -99,7 +99,7 @@ public async Task async_configure_publisher_api_client_callback_is_awaited() [Fact] public async Task use_credential_sets_credential_on_publisher_api_builder() { - var transport = new PubsubTransport("test"); + var transport = new PubsubTransport { ProjectId = "test" }; var config = new PubsubConfiguration(transport, new WolverineOptions()); var credential = GoogleCredential.FromAccessToken("test-token"); GoogleCredential? observed = null; @@ -114,7 +114,7 @@ public async Task use_credential_sets_credential_on_publisher_api_builder() [Fact] public async Task use_credential_sets_credential_on_subscriber_api_builder() { - var transport = new PubsubTransport("test"); + var transport = new PubsubTransport { ProjectId = "test" }; var config = new PubsubConfiguration(transport, new WolverineOptions()); var credential = GoogleCredential.FromAccessToken("test-token"); GoogleCredential? observed = null; @@ -129,7 +129,7 @@ public async Task use_credential_sets_credential_on_subscriber_api_builder() [Fact] public async Task use_credential_sets_credential_on_subscriber_client_builder() { - var transport = new PubsubTransport("test"); + var transport = new PubsubTransport { ProjectId = "test" }; var config = new PubsubConfiguration(transport, new WolverineOptions()); var credential = GoogleCredential.FromAccessToken("test-token"); GoogleCredential? observed = null; @@ -144,7 +144,7 @@ public async Task use_credential_sets_credential_on_subscriber_client_builder() [Fact] public async Task async_use_credential_factory_sets_credential_on_publisher_api_builder() { - var transport = new PubsubTransport("test"); + var transport = new PubsubTransport { ProjectId = "test" }; var config = new PubsubConfiguration(transport, new WolverineOptions()); var credential = GoogleCredential.FromAccessToken("test-token"); GoogleCredential? observed = null; @@ -164,7 +164,7 @@ public async Task async_use_credential_factory_sets_credential_on_publisher_api_ [Fact] public async Task multiple_configure_subscriber_api_client_calls_compose_in_order() { - var transport = new PubsubTransport("test"); + var transport = new PubsubTransport { ProjectId = "test" }; var config = new PubsubConfiguration(transport, new WolverineOptions()); var order = new List(); @@ -178,7 +178,7 @@ public async Task multiple_configure_subscriber_api_client_calls_compose_in_orde [Fact] public async Task async_configure_subscriber_api_client_callback_is_awaited() { - var transport = new PubsubTransport("test"); + var transport = new PubsubTransport { ProjectId = "test" }; var config = new PubsubConfiguration(transport, new WolverineOptions()); var called = false; @@ -195,7 +195,7 @@ public async Task async_configure_subscriber_api_client_callback_is_awaited() [Fact] public async Task async_configure_subscriber_client_callback_is_awaited() { - var transport = new PubsubTransport("test"); + var transport = new PubsubTransport { ProjectId = "test" }; var config = new PubsubConfiguration(transport, new WolverineOptions()); var called = false; @@ -212,7 +212,7 @@ public async Task async_configure_subscriber_client_callback_is_awaited() [Fact] public async Task async_use_credential_factory_sets_credential_on_subscriber_api_builder() { - var transport = new PubsubTransport("test"); + var transport = new PubsubTransport { ProjectId = "test" }; var config = new PubsubConfiguration(transport, new WolverineOptions()); var credential = GoogleCredential.FromAccessToken("test-token"); GoogleCredential? observed = null; @@ -231,7 +231,7 @@ public async Task async_use_credential_factory_sets_credential_on_subscriber_api [Fact] public async Task async_use_credential_factory_sets_credential_on_subscriber_client_builder() { - var transport = new PubsubTransport("test"); + var transport = new PubsubTransport { ProjectId = "test" }; var config = new PubsubConfiguration(transport, new WolverineOptions()); var credential = GoogleCredential.FromAccessToken("test-token"); GoogleCredential? observed = null; @@ -253,7 +253,7 @@ public async Task subscriber_client_credential_factory_is_invoked_on_each_connec // Each listener (re)connect rebuilds the streaming SubscriberClient and re-applies the // configure callback, so an async credential factory runs again every time. This is what // lets rolling/rotated credentials be picked up without restarting the application. - var transport = new PubsubTransport("test"); + var transport = new PubsubTransport { ProjectId = "test" }; var config = new PubsubConfiguration(transport, new WolverineOptions()); var invocations = 0; @@ -274,7 +274,7 @@ public async Task subscriber_client_credential_factory_is_invoked_on_each_connec [Fact] public async Task subscriber_client_receives_a_fresh_credential_on_each_connect() { - var transport = new PubsubTransport("test"); + var transport = new PubsubTransport { ProjectId = "test" }; var config = new PubsubConfiguration(transport, new WolverineOptions()); var counter = 0; diff --git a/src/Transports/GCP/Wolverine.Pubsub.Tests/PubsubEndpointTests.cs b/src/Transports/GCP/Wolverine.Pubsub.Tests/PubsubEndpointTests.cs index a1e233a35..e0968efe9 100644 --- a/src/Transports/GCP/Wolverine.Pubsub.Tests/PubsubEndpointTests.cs +++ b/src/Transports/GCP/Wolverine.Pubsub.Tests/PubsubEndpointTests.cs @@ -13,8 +13,9 @@ public class PubsubEndpointTests { private PubsubTransport createTransport() { - return new PubsubTransport("wolverine") + return new PubsubTransport { + ProjectId = "wolverine", PublisherApiClient = Substitute.For(), SubscriberApiClient = Substitute.For(), EmulatorDetection = EmulatorDetection.EmulatorOnly diff --git a/src/Transports/GCP/Wolverine.Pubsub.Tests/PubsubPerTenantBrokerTests.cs b/src/Transports/GCP/Wolverine.Pubsub.Tests/PubsubPerTenantBrokerTests.cs new file mode 100644 index 000000000..eb1e211a8 --- /dev/null +++ b/src/Transports/GCP/Wolverine.Pubsub.Tests/PubsubPerTenantBrokerTests.cs @@ -0,0 +1,236 @@ +using Google.Api.Gax; +using Google.Api.Gax.Grpc; +using Google.Cloud.PubSub.V1; +using Grpc.Core; +using JasperFx.Core; +using Microsoft.Extensions.Hosting; +using Shouldly; +using Wolverine.Tracking; +using Xunit; + +namespace Wolverine.Pubsub.Tests; + +/// +/// Integration coverage for broker-per-tenant Google Cloud Platform Pub/Sub (GH-3306). Project-id-per-tenant is the +/// isolation axis: the default connection uses project "wolverine" and the tenant uses project "wolverine2". The +/// Pub/Sub emulator accepts arbitrary project ids at request time on the same endpoint, so the tenant's topic/ +/// subscription are genuinely distinct GCP resources from the shared/default ones — letting us prove real routing +/// (a tenant message lands under the tenant project and NOT the default one, and vice versa) plus inbound +/// stamping. +/// +/// Skip-guarded when the emulator/Docker is unavailable. +/// +public class PubsubPerTenantBrokerTests : IAsyncLifetime +{ + private const string DefaultProject = "wolverine"; + private const string TenantProject = "wolverine2"; + private const string TenantId = "tenant2"; + + private bool _skip; + + public async Task InitializeAsync() + { + _skip = !await TestingExtensions.IsEmulatorAvailable(); + Environment.SetEnvironmentVariable("PUBSUB_EMULATOR_HOST", TestingExtensions.EmulatorHost); + } + + public Task DisposeAsync() => Task.CompletedTask; + + [Fact] + public async Task tenant_message_is_published_to_the_tenant_project_and_not_the_default() + { + if (_skip) return; + + var topic = $"pertenant-{Guid.NewGuid():N}"; + + // Pre-create a topic + verification subscription under BOTH projects. The subscription must exist before we + // publish for Pub/Sub to retain the message. + var defaultSub = await provisionAsync(DefaultProject, topic); + var tenantSub = await provisionAsync(TenantProject, topic); + + using var host = await buildSenderHostAsync(topic); + + await host.MessageBus().SendAsync(new PerTenantMessage("for-tenant"), + new DeliveryOptions { TenantId = TenantId }); + + // Landed under the tenant project... + (await pullOneAsync(tenantSub, 15.Seconds())).ShouldNotBeNull(); + // ...and NOT under the default project. + (await pullOneAsync(defaultSub, 3.Seconds())).ShouldBeNull(); + } + + [Fact] + public async Task default_message_is_published_to_the_default_project() + { + if (_skip) return; + + var topic = $"pertenant-{Guid.NewGuid():N}"; + + var defaultSub = await provisionAsync(DefaultProject, topic); + var tenantSub = await provisionAsync(TenantProject, topic); + + using var host = await buildSenderHostAsync(topic); + + // No tenant id => FallbackToDefault routes to the shared/default connection. + await host.MessageBus().SendAsync(new PerTenantMessage("no-tenant")); + + (await pullOneAsync(defaultSub, 15.Seconds())).ShouldNotBeNull(); + (await pullOneAsync(tenantSub, 3.Seconds())).ShouldBeNull(); + } + + [Fact] + public async Task tenant_message_is_consumed_and_stamped_with_the_tenant_id() + { + if (_skip) return; + + var topic = $"pertenant-{Guid.NewGuid():N}"; + + using var host = await Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + opts.ServiceName = "PerTenantInbound"; + opts.UsePubsubTesting() + .AutoProvision() + .TenantIdBehavior(Transports.Sending.TenantedIdBehavior.FallbackToDefault) + .AddTenant(TenantId, TenantProject, + t => t.EmulatorDetection = EmulatorDetection.EmulatorOnly); + + opts.Policies.DisableConventionalLocalRouting(); + opts.PublishMessage().ToPubsubTopic(topic).SendInline(); + opts.ListenToPubsubTopic(topic); + }) + .StartAsync(); + + // The default listener polls the default project and the tenant listener polls the tenant project; the + // message only exists under the tenant project, so only the tenant listener consumes it and stamps the id. + var session = await host + .TrackActivity() + .IncludeExternalTransports() + .Timeout(60.Seconds()) + .WaitForMessageToBeReceivedAt(host) + .ExecuteAndWaitAsync(c => + c.SendAsync(new PerTenantMessage("for-tenant"), new DeliveryOptions { TenantId = TenantId })); + + var received = session.Received.SingleEnvelope(); + received.TenantId.ShouldBe(TenantId); + received.Message.ShouldBeOfType().Value.ShouldBe("for-tenant"); + } + + [Fact] + public async Task auto_provision_creates_the_topology_under_the_tenant_project() + { + if (_skip) return; + + var topic = $"pertenant-{Guid.NewGuid():N}"; + + using var host = await Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + opts.ServiceName = "PerTenantProvisioning"; + opts.UsePubsubTesting() + .AutoProvision() + .AddTenant(TenantId, TenantProject, + t => t.EmulatorDetection = EmulatorDetection.EmulatorOnly); + + opts.PublishMessage().ToPubsubTopic(topic); + opts.ListenToPubsubTopic(topic); + }) + .StartAsync(); + + // Prove the topic was actually created under the tenant project (GetTopicAsync throws when absent). + var publisher = await new PublisherServiceApiClientBuilder + { + EmulatorDetection = EmulatorDetection.EmulatorOnly + }.BuildAsync(); + + var tenantTopic = await publisher.GetTopicAsync(new TopicName(TenantProject, topic)); + tenantTopic.ShouldNotBeNull(); + } + + private static Task buildSenderHostAsync(string topic) + { + return Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + opts.ServiceName = "PerTenantSender"; + opts.UsePubsubTesting() + .TenantIdBehavior(Transports.Sending.TenantedIdBehavior.FallbackToDefault) + .AddTenant(TenantId, TenantProject, + t => t.EmulatorDetection = EmulatorDetection.EmulatorOnly); + + opts.Policies.DisableConventionalLocalRouting(); + opts.PublishMessage().ToPubsubTopic(topic).SendInline(); + }) + .StartAsync(); + } + + private static async Task provisionAsync(string projectId, string topic) + { + var publisher = await new PublisherServiceApiClientBuilder + { + EmulatorDetection = EmulatorDetection.EmulatorOnly + }.BuildAsync(); + var subscriber = await new SubscriberServiceApiClientBuilder + { + EmulatorDetection = EmulatorDetection.EmulatorOnly + }.BuildAsync(); + + var topicName = new TopicName(projectId, topic); + var subscriptionName = new SubscriptionName(projectId, $"{topic}-verify"); + + try + { + await publisher.CreateTopicAsync(topicName); + } + catch (RpcException ex) when (ex.StatusCode == StatusCode.AlreadyExists) + { + } + + try + { + await subscriber.CreateSubscriptionAsync(subscriptionName, topicName, pushConfig: null, + ackDeadlineSeconds: 60); + } + catch (RpcException ex) when (ex.StatusCode == StatusCode.AlreadyExists) + { + } + + return subscriptionName; + } + + private static async Task pullOneAsync(SubscriptionName subscription, TimeSpan timeout) + { + var subscriber = await new SubscriberServiceApiClientBuilder + { + EmulatorDetection = EmulatorDetection.EmulatorOnly + }.BuildAsync(); + + var deadline = DateTime.UtcNow + timeout; + while (DateTime.UtcNow < deadline) + { + var response = await subscriber.PullAsync(subscription, maxMessages: 1, + CallSettings.FromExpiration(Expiration.FromTimeout(2.Seconds()))); + + var message = response.ReceivedMessages.FirstOrDefault(); + if (message is not null) + { + await subscriber.AcknowledgeAsync(subscription, new[] { message.AckId }); + return message; + } + + await Task.Delay(250.Milliseconds()); + } + + return null; + } +} + +public record PerTenantMessage(string Value); + +public static class PerTenantMessageHandler +{ + public static void Handle(PerTenantMessage message) + { + // no-op; presence lets Wolverine discover a handler so receive tests can track processing + } +} diff --git a/src/Transports/GCP/Wolverine.Pubsub.Tests/PubsubPerTenantConfigurationTests.cs b/src/Transports/GCP/Wolverine.Pubsub.Tests/PubsubPerTenantConfigurationTests.cs new file mode 100644 index 000000000..17f921fed --- /dev/null +++ b/src/Transports/GCP/Wolverine.Pubsub.Tests/PubsubPerTenantConfigurationTests.cs @@ -0,0 +1,82 @@ +using Google.Api.Gax; +using JasperFx.Core; +using Shouldly; +using Wolverine.Configuration; +using Wolverine.Transports.Sending; +using Xunit; + +namespace Wolverine.Pubsub.Tests; + +// Broker-per-tenant (GH-3306) unit coverage that needs NO emulator: it exercises the tenant registration / topology +// wiring, not live send/receive. Live tenant routing behavior is covered by PubsubPerTenantBrokerTests. +public class PubsubPerTenantConfigurationTests +{ + [Fact] + public void add_tenant_registers_a_tenant_with_its_own_project() + { + var options = new WolverineOptions(); + options.UsePubsub("wolverine") + .AddTenant("tenant2", "wolverine2"); + + var transport = options.Transports.GetOrCreate(); + transport.Tenants.Count().ShouldBe(1); + + var tenant = transport.Tenants["tenant2"]; + tenant.TenantId.ShouldBe("tenant2"); + tenant.ProjectId.ShouldBe("wolverine2"); + } + + [Fact] + public void add_tenant_configure_action_can_override_emulator_detection() + { + var options = new WolverineOptions(); + options.UsePubsub("wolverine") + .AddTenant("tenant2", "wolverine2", t => t.EmulatorDetection = EmulatorDetection.EmulatorOnly); + + var transport = options.Transports.GetOrCreate(); + transport.Tenants["tenant2"].EmulatorDetection.ShouldBe(EmulatorDetection.EmulatorOnly); + } + + [Fact] + public void tenant_id_behavior_defaults_to_fallback_to_default() + { + var transport = new PubsubTransport { ProjectId = "wolverine" }; + transport.TenantedIdBehavior.ShouldBe(TenantedIdBehavior.FallbackToDefault); + } + + [Fact] + public void tenant_id_behavior_is_configurable() + { + var options = new WolverineOptions(); + options.UsePubsub("wolverine") + .TenantIdBehavior(TenantedIdBehavior.TenantIdRequired); + + var transport = options.Transports.GetOrCreate(); + transport.TenantedIdBehavior.ShouldBe(TenantedIdBehavior.TenantIdRequired); + } + + [Fact] + public void pubsub_endpoints_are_tenant_aware_by_default() + { + var transport = new PubsubTransport { ProjectId = "wolverine" }; + var topic = transport.Topics["orders"]; + topic.TenancyBehavior.ShouldBe(TenancyBehavior.TenantAware); + } + + [Fact] + public void tenant_topic_and_subscription_names_use_the_tenant_project_but_share_the_id() + { + var transport = new PubsubTransport { ProjectId = "wolverine" }; + var topic = transport.Topics["orders"]; + + // Default project resolves to the endpoint's own names... + topic.TopicNameFor("wolverine").ShouldBe(topic.Server.Topic.Name); + topic.SubscriptionNameFor("wolverine").ShouldBe(topic.Server.Subscription.Name); + + // ...a tenant project yields a physically distinct resource sharing the same logical id. + topic.TopicNameFor("wolverine2").ProjectId.ShouldBe("wolverine2"); + topic.TopicNameFor("wolverine2").TopicId.ShouldBe(topic.Server.Topic.Name.TopicId); + topic.SubscriptionNameFor("wolverine2").ProjectId.ShouldBe("wolverine2"); + topic.SubscriptionNameFor("wolverine2").SubscriptionId.ShouldBe(topic.Server.Subscription.Name.SubscriptionId); + } +} diff --git a/src/Transports/GCP/Wolverine.Pubsub.Tests/PubsubTransportTests.cs b/src/Transports/GCP/Wolverine.Pubsub.Tests/PubsubTransportTests.cs index b6f7fd4e2..9c2b14a73 100644 --- a/src/Transports/GCP/Wolverine.Pubsub.Tests/PubsubTransportTests.cs +++ b/src/Transports/GCP/Wolverine.Pubsub.Tests/PubsubTransportTests.cs @@ -8,7 +8,7 @@ public class PubsubTransportTests [Fact] public void find_topic_by_uri() { - var transport = new PubsubTransport("wolverine"); + var transport = new PubsubTransport { ProjectId = "wolverine" }; var topic = transport.GetOrCreateEndpoint(new Uri($"{PubsubTransport.ProtocolName}://wolverine/one")) .ShouldBeOfType(); @@ -19,7 +19,7 @@ public void find_topic_by_uri() [Fact] public void response_subscriptions_are_disabled_by_default() { - var transport = new PubsubTransport("wolverine"); + var transport = new PubsubTransport { ProjectId = "wolverine" }; transport.SystemEndpointsEnabled.ShouldBeFalse(); } @@ -27,7 +27,7 @@ public void response_subscriptions_are_disabled_by_default() [Fact] public void return_all_endpoints_gets_dead_letter_subscription_too() { - var transport = new PubsubTransport("wolverine"); + var transport = new PubsubTransport { ProjectId = "wolverine" }; transport.DeadLetter.Enabled = true; @@ -56,7 +56,7 @@ public void findEndpointByUri_should_correctly_find_by_queuename() var queueNameInPascalCase = "TestQueue"; var queueNameLowerCase = "testqueue"; - var transport = new PubsubTransport("wolverine"); + var transport = new PubsubTransport { ProjectId = "wolverine" }; var abc = transport.Topics[queueNameInPascalCase]; var xzy = transport.Topics[queueNameLowerCase]; diff --git a/src/Transports/GCP/Wolverine.Pubsub.Tests/TestingExtensions.cs b/src/Transports/GCP/Wolverine.Pubsub.Tests/TestingExtensions.cs index a1d85c129..df0f5b549 100644 --- a/src/Transports/GCP/Wolverine.Pubsub.Tests/TestingExtensions.cs +++ b/src/Transports/GCP/Wolverine.Pubsub.Tests/TestingExtensions.cs @@ -1,19 +1,55 @@ +using System.Net.Sockets; using Google.Api.Gax; namespace Wolverine.Pubsub.Tests; public static class TestingExtensions { + public const string EmulatorHost = "localhost:8085"; + public static PubsubConfiguration UsePubsubTesting(this WolverineOptions options) { // Use localhost (not the IPv6 literal [::1]) so this resolves to the IPv4 address that // Docker publishes the emulator on. CI Linux runners publish the container port on IPv4 // only, where [::1]:8085 is unreachable. See #3191. - Environment.SetEnvironmentVariable("PUBSUB_EMULATOR_HOST", "localhost:8085"); + Environment.SetEnvironmentVariable("PUBSUB_EMULATOR_HOST", EmulatorHost); Environment.SetEnvironmentVariable("PUBSUB_PROJECT_ID", "wolverine"); return options .UsePubsub("wolverine") .UseEmulatorDetection(EmulatorDetection.EmulatorOnly); } -} \ No newline at end of file + + /// + /// Register an additional, named Pub/Sub broker pointed at a second project on the same emulator. The emulator + /// accepts arbitrary project ids at request time, so no compose change is needed — just a distinct project id. + /// + public static PubsubConfiguration AddNamedPubsubBrokerTesting(this WolverineOptions options, BrokerName name, + string projectId) + { + Environment.SetEnvironmentVariable("PUBSUB_EMULATOR_HOST", EmulatorHost); + + return options + .AddNamedPubsubBroker(name, projectId) + .UseEmulatorDetection(EmulatorDetection.EmulatorOnly); + } + + /// + /// True when the Pub/Sub emulator TCP port is reachable. Integration tests skip-guard on this so the suite is + /// green when Docker/the emulator is not running. + /// + public static async Task IsEmulatorAvailable() + { + try + { + using var client = new TcpClient(); + var connect = client.ConnectAsync("localhost", 8085); + var completed = await Task.WhenAny(connect, Task.Delay(TimeSpan.FromSeconds(2))); + return completed == connect && client.Connected; + } + catch + { + return false; + } + } +} diff --git a/src/Transports/GCP/Wolverine.Pubsub.Tests/broker_connection_summary_3269.cs b/src/Transports/GCP/Wolverine.Pubsub.Tests/broker_connection_summary_3269.cs index f5f6e5235..d7f9be054 100644 --- a/src/Transports/GCP/Wolverine.Pubsub.Tests/broker_connection_summary_3269.cs +++ b/src/Transports/GCP/Wolverine.Pubsub.Tests/broker_connection_summary_3269.cs @@ -12,21 +12,21 @@ public class broker_connection_summary_3269 [Fact] public void describe_endpoint_reports_the_project_id() { - var transport = new PubsubTransport("my-project"); + var transport = new PubsubTransport { ProjectId = "my-project" }; transport.DescribeEndpoint().ShouldBe("project: my-project"); } [Fact] public void describe_endpoint_marks_the_emulator() { - var transport = new PubsubTransport("my-project") { EmulatorDetection = EmulatorDetection.EmulatorOnly }; + var transport = new PubsubTransport { ProjectId = "my-project", EmulatorDetection = EmulatorDetection.EmulatorOnly }; transport.DescribeEndpoint().ShouldBe("project: my-project (emulator)"); } [Fact] public void broker_description_endpoint_is_populated() { - var transport = new PubsubTransport("my-project"); + var transport = new PubsubTransport { ProjectId = "my-project" }; new BrokerDescription(transport).Endpoint.ShouldBe("project: my-project"); } } diff --git a/src/Transports/GCP/Wolverine.Pubsub.Tests/broker_role_tests.cs b/src/Transports/GCP/Wolverine.Pubsub.Tests/broker_role_tests.cs index b359f32a1..9c78c4a4e 100644 --- a/src/Transports/GCP/Wolverine.Pubsub.Tests/broker_role_tests.cs +++ b/src/Transports/GCP/Wolverine.Pubsub.Tests/broker_role_tests.cs @@ -11,8 +11,9 @@ public class broker_role_tests [Fact] public void pubsub_endpoint_broker_role_is_pubsub() { - var transport = new PubsubTransport("wolverine") + var transport = new PubsubTransport { + ProjectId = "wolverine", PublisherApiClient = Substitute.For(), SubscriberApiClient = Substitute.For(), EmulatorDetection = EmulatorDetection.EmulatorOnly diff --git a/src/Transports/GCP/Wolverine.Pubsub/Internal/BatchedPubsubListener.cs b/src/Transports/GCP/Wolverine.Pubsub/Internal/BatchedPubsubListener.cs index 17d4b23ca..82614ccb6 100644 --- a/src/Transports/GCP/Wolverine.Pubsub/Internal/BatchedPubsubListener.cs +++ b/src/Transports/GCP/Wolverine.Pubsub/Internal/BatchedPubsubListener.cs @@ -10,8 +10,9 @@ public BatchedPubsubListener( PubsubEndpoint endpoint, PubsubTransport transport, IReceiver receiver, - IWolverineRuntime runtime - ) : base(endpoint, transport, receiver, runtime) + IWolverineRuntime runtime, + PubsubClientSet clients + ) : base(endpoint, transport, receiver, runtime, clients) { } @@ -19,11 +20,11 @@ public override async Task StartAsync() { await listenForMessagesAsync(async () => { - var subscriptionName = _endpoint.Server.Subscription.Name; + var subscriptionName = ListeningSubscriptionName; var subscriberBuilder = new SubscriberClientBuilder { SubscriptionName = subscriptionName, - EmulatorDetection = _transport.EmulatorDetection, + EmulatorDetection = _clients.EmulatorDetection, Settings = new() { // https://cloud.google.com/dotnet/docs/reference/Google.Cloud.PubSub.V1/latest/Google.Cloud.PubSub.V1.SubscriberClient.Settings#Google_Cloud_PubSub_V1_SubscriberClient_Settings_FlowControlSettings @@ -32,8 +33,8 @@ await listenForMessagesAsync(async () => FlowControlSettings = new(_endpoint.Client.MaxOutstandingMessages, _endpoint.Client.MaxOutstandingByteCount), } }; - if (_transport.ConfigureSubscriberClientBuilder != null) - await _transport.ConfigureSubscriberClientBuilder(subscriberBuilder); + if (_clients.ConfigureSubscriberClientBuilder != null) + await _clients.ConfigureSubscriberClientBuilder(subscriberBuilder); await using SubscriberClient subscriber = await subscriberBuilder.BuildAsync(); var ctRegistration = _cancellation.Token.Register(() => subscriber.StopAsync(CancellationToken.None)); try diff --git a/src/Transports/GCP/Wolverine.Pubsub/Internal/InlinePubsubListener.cs b/src/Transports/GCP/Wolverine.Pubsub/Internal/InlinePubsubListener.cs index 55f03ed02..0fa600488 100644 --- a/src/Transports/GCP/Wolverine.Pubsub/Internal/InlinePubsubListener.cs +++ b/src/Transports/GCP/Wolverine.Pubsub/Internal/InlinePubsubListener.cs @@ -10,8 +10,9 @@ public InlinePubsubListener( PubsubEndpoint endpoint, PubsubTransport transport, IReceiver receiver, - IWolverineRuntime runtime - ) : base(endpoint, transport, receiver, runtime) + IWolverineRuntime runtime, + PubsubClientSet clients + ) : base(endpoint, transport, receiver, runtime, clients) { } @@ -20,14 +21,14 @@ public override async Task StartAsync() { await listenForMessagesAsync(async () => { - var subscriptionName = _endpoint.Server.Subscription.Name; + var subscriptionName = ListeningSubscriptionName; var subscriberBuilder = new SubscriberClientBuilder { SubscriptionName = subscriptionName, - EmulatorDetection = _transport.EmulatorDetection, + EmulatorDetection = _clients.EmulatorDetection, }; - if (_transport.ConfigureSubscriberClientBuilder != null) - await _transport.ConfigureSubscriberClientBuilder(subscriberBuilder); + if (_clients.ConfigureSubscriberClientBuilder != null) + await _clients.ConfigureSubscriberClientBuilder(subscriberBuilder); await using SubscriberClient subscriber = await subscriberBuilder.BuildAsync(); var ctRegistration = _cancellation.Token.Register(() => subscriber.StopAsync(CancellationToken.None)); try diff --git a/src/Transports/GCP/Wolverine.Pubsub/Internal/InlinePubsubSender.cs b/src/Transports/GCP/Wolverine.Pubsub/Internal/InlinePubsubSender.cs index 8d8a0a69b..91fc17a50 100644 --- a/src/Transports/GCP/Wolverine.Pubsub/Internal/InlinePubsubSender.cs +++ b/src/Transports/GCP/Wolverine.Pubsub/Internal/InlinePubsubSender.cs @@ -8,14 +8,17 @@ public class InlinePubsubSender : ISender { private readonly PubsubEndpoint _endpoint; private readonly ILogger _logger; + private readonly PubsubClientSet _clients; public InlinePubsubSender( PubsubEndpoint endpoint, - IWolverineRuntime runtime + IWolverineRuntime runtime, + PubsubClientSet? clients = null ) { _endpoint = endpoint; _logger = runtime.LoggerFactory.CreateLogger(); + _clients = clients ?? endpoint.Transport.DefaultClients; } public bool SupportsNativeScheduledSend => false; @@ -39,6 +42,6 @@ public async Task PingAsync() public async ValueTask SendAsync(Envelope envelope) { - await _endpoint.SendMessageAsync(envelope, _logger); + await _endpoint.SendMessageAsync(envelope, _logger, _clients); } } \ No newline at end of file diff --git a/src/Transports/GCP/Wolverine.Pubsub/Internal/PubsubClientSet.cs b/src/Transports/GCP/Wolverine.Pubsub/Internal/PubsubClientSet.cs new file mode 100644 index 000000000..2791c838e --- /dev/null +++ b/src/Transports/GCP/Wolverine.Pubsub/Internal/PubsubClientSet.cs @@ -0,0 +1,25 @@ +using Google.Api.Gax; +using Google.Cloud.PubSub.V1; + +namespace Wolverine.Pubsub.Internal; + +/// +/// A resolved set of Google Cloud Platform Pub/Sub API clients (plus the project id and emulator/streaming +/// configuration they were built with) for a single connection target. The default transport connection and each +/// per-tenant connection (broker-per-tenant, GH-3306) produce one of these; senders and listeners are parameterized +/// with a so the same endpoint topology can be published/consumed over a different +/// GCP project at runtime, keyed by . +/// +public class PubsubClientSet +{ + public required string ProjectId { get; init; } + public required EmulatorDetection EmulatorDetection { get; init; } + public PublisherServiceApiClient? PublisherApiClient { get; init; } + public SubscriberServiceApiClient? SubscriberApiClient { get; init; } + + /// + /// Optional async hook applied to the streaming each listener builds + /// for this connection (e.g. per-tenant credentials). + /// + public Func? ConfigureSubscriberClientBuilder { get; init; } +} diff --git a/src/Transports/GCP/Wolverine.Pubsub/Internal/PubsubListener.cs b/src/Transports/GCP/Wolverine.Pubsub/Internal/PubsubListener.cs index 7e847096f..3bd44ff5a 100644 --- a/src/Transports/GCP/Wolverine.Pubsub/Internal/PubsubListener.cs +++ b/src/Transports/GCP/Wolverine.Pubsub/Internal/PubsubListener.cs @@ -23,16 +23,22 @@ public abstract class PubsubListener : IListener, ISupportDeadLetterQueue protected readonly PubsubTransport _transport; protected Task _task; + + /// + /// The connection (default or per-tenant) this listener consumes and, for requeue/dead-letter, re-publishes over. + /// + protected readonly PubsubClientSet _clients; private readonly IPubsubEnvelopeMapper _mapper; public PubsubListener( PubsubEndpoint endpoint, PubsubTransport transport, IReceiver receiver, - IWolverineRuntime runtime + IWolverineRuntime runtime, + PubsubClientSet clients ) { - if (transport.SubscriberApiClient is null) + if (clients.SubscriberApiClient is null) { throw new WolverinePubsubTransportNotConnectedException(); } @@ -43,6 +49,7 @@ IWolverineRuntime runtime _transport = transport; _receiver = receiver; _runtime = runtime; + _clients = clients; _logger = runtime.LoggerFactory.CreateLogger(); if (_endpoint.DeadLetterName.IsNotEmpty()) @@ -58,17 +65,22 @@ IWolverineRuntime runtime { return; } - await _deadLetterTopic.SendMessageAsync(e, _logger); + await _deadLetterTopic.SendMessageAsync(e, _logger, _clients); }, _logger, runtime.Cancellation); _requeue = new RetryBlock(async (e, _) => { - await _endpoint.SendMessageAsync(e, _logger); + await _endpoint.SendMessageAsync(e, _logger, _clients); }, _logger, runtime.Cancellation); _task = StartAsync(); } + /// + /// The subscription this listener pulls from, resolved for its connection's project (default or tenant). + /// + protected SubscriptionName ListeningSubscriptionName => _endpoint.SubscriptionNameFor(_clients.ProjectId); + public Uri Address => _endpoint.Uri; public IHandlerPipeline? Pipeline => _receiver.Pipeline; diff --git a/src/Transports/GCP/Wolverine.Pubsub/Internal/PubsubTenant.cs b/src/Transports/GCP/Wolverine.Pubsub/Internal/PubsubTenant.cs new file mode 100644 index 000000000..d9bd5cc73 --- /dev/null +++ b/src/Transports/GCP/Wolverine.Pubsub/Internal/PubsubTenant.cs @@ -0,0 +1,88 @@ +using Google.Api.Gax; +using Google.Cloud.PubSub.V1; + +namespace Wolverine.Pubsub.Internal; + +/// +/// A single tenant in the broker-per-tenant Google Cloud Platform Pub/Sub model (GH-3306). Project-id-per-tenant is +/// the natural isolation axis: / embed the project id, so a +/// tenant pointed at its own already yields physically distinct GCP resources for the same +/// logical topology. Each tenant owns its own / +/// pair (built in ), and optionally its own credential hooks (seeded from the parent +/// transport). Modeled after the NATS per-tenant connection support. +/// +public class PubsubTenant +{ + public PubsubTenant(string tenantId, string projectId) + { + TenantId = tenantId ?? throw new ArgumentNullException(nameof(tenantId)); + ProjectId = projectId ?? throw new ArgumentNullException(nameof(projectId)); + } + + public string TenantId { get; } + + /// + /// The GCP project id that isolates this tenant's topology. Distinct from the parent/default project id. + /// + public string ProjectId { get; } + + /// + /// Emulator detection for this tenant's clients. Seeded from the parent transport so tenant tests run against + /// the same emulator; may be overridden per tenant. + /// + public EmulatorDetection EmulatorDetection { get; set; } = EmulatorDetection.None; + + /// + /// Optional async hook to configure this tenant's (e.g. dedicated + /// credentials). Seeded from the parent transport unless overridden. + /// + public Func? ConfigurePublisherApiBuilder { get; set; } + + /// + /// Optional async hook to configure this tenant's . Seeded from + /// the parent transport unless overridden. + /// + public Func? ConfigureSubscriberApiBuilder { get; set; } + + /// + /// Optional async hook to configure the streaming each of this tenant's + /// listeners builds. Seeded from the parent transport unless overridden. + /// + public Func? ConfigureSubscriberClientBuilder { get; set; } + + /// + /// The resolved client set for this tenant, built once during the transport's ConnectAsync and owned for the + /// lifetime of the transport. + /// + internal PubsubClientSet? Clients { get; private set; } + + internal async Task ConnectAsync() + { + var pubBuilder = new PublisherServiceApiClientBuilder + { + EmulatorDetection = EmulatorDetection + }; + if (ConfigurePublisherApiBuilder != null) + { + await ConfigurePublisherApiBuilder(pubBuilder); + } + + var subBuilder = new SubscriberServiceApiClientBuilder + { + EmulatorDetection = EmulatorDetection + }; + if (ConfigureSubscriberApiBuilder != null) + { + await ConfigureSubscriberApiBuilder(subBuilder); + } + + Clients = new PubsubClientSet + { + ProjectId = ProjectId, + EmulatorDetection = EmulatorDetection, + PublisherApiClient = await pubBuilder.BuildAsync(), + SubscriberApiClient = await subBuilder.BuildAsync(), + ConfigureSubscriberClientBuilder = ConfigureSubscriberClientBuilder + }; + } +} diff --git a/src/Transports/GCP/Wolverine.Pubsub/PubsubConfiguration.cs b/src/Transports/GCP/Wolverine.Pubsub/PubsubConfiguration.cs index 184c131d6..9a83c7151 100644 --- a/src/Transports/GCP/Wolverine.Pubsub/PubsubConfiguration.cs +++ b/src/Transports/GCP/Wolverine.Pubsub/PubsubConfiguration.cs @@ -1,7 +1,9 @@ using Google.Api.Gax; using Google.Apis.Auth.OAuth2; using Google.Cloud.PubSub.V1; +using Wolverine.Pubsub.Internal; using Wolverine.Transports; +using Wolverine.Transports.Sending; namespace Wolverine.Pubsub; @@ -200,6 +202,50 @@ public PubsubConfiguration UseCredential(Func> crede return this; } + /// + /// Control how Wolverine routes outbound messages whose is null or does not + /// match a registered tenant (broker-per-tenant). Defaults to + /// , which uses the default/shared connection. + /// + public PubsubConfiguration TenantIdBehavior(TenantedIdBehavior behavior) + { + Transport.TenantedIdBehavior = behavior; + + return this; + } + + /// + /// Register a tenant served by its own dedicated Google Cloud Platform Pub/Sub connection using a different + /// GCP project (broker-per-tenant). The tenant shares the endpoint topology declared on this transport, but + /// messages are routed to its project at runtime by . Credential / emulator + /// settings are seeded from the parent transport. + /// + /// The tenant id used to route messages (e.g. via DeliveryOptions.TenantId). + /// The GCP project id that isolates this tenant's topology. + public PubsubConfiguration AddTenant(string tenantId, string projectId) + { + return AddTenant(tenantId, projectId, null); + } + + /// + /// Register a tenant served by its own dedicated Google Cloud Platform Pub/Sub connection using a different + /// GCP project (broker-per-tenant), with an opportunity to override the tenant's credential / emulator hooks. + /// Any hook the tenant does not set is seeded from the parent transport. + /// + /// The tenant id used to route messages (e.g. via DeliveryOptions.TenantId). + /// The GCP project id that isolates this tenant's topology. + /// Optional per-tenant configuration (dedicated credentials, emulator detection). + public PubsubConfiguration AddTenant(string tenantId, string projectId, Action? configure) + { + var tenant = new PubsubTenant(tenantId, projectId); + + configure?.Invoke(tenant); + + Transport.Tenants[tenantId] = tenant; + + return this; + } + protected override PubsubTopicListenerConfiguration createListenerExpression(PubsubEndpoint listenerEndpoint) { return new PubsubTopicListenerConfiguration(listenerEndpoint); diff --git a/src/Transports/GCP/Wolverine.Pubsub/PubsubEndpoint.cs b/src/Transports/GCP/Wolverine.Pubsub/PubsubEndpoint.cs index 46e6235ff..4e8020ad5 100644 --- a/src/Transports/GCP/Wolverine.Pubsub/PubsubEndpoint.cs +++ b/src/Transports/GCP/Wolverine.Pubsub/PubsubEndpoint.cs @@ -34,6 +34,31 @@ protected override PubsubEnvelopeMapper buildMapper(IWolverineRuntime runtime) public PubsubServerOptions Server = new(); + internal PubsubTransport Transport => _transport; + + /// + /// The topic name for this endpoint under the given project id. For the default project this is the endpoint's + /// own name; for a tenant project (broker-per-tenant) it is the same + /// topic id under the tenant's project, yielding a physically distinct GCP topic. + /// + internal TopicName TopicNameFor(string projectId) + { + return projectId == _transport.ProjectId + ? Server.Topic.Name + : new TopicName(projectId, Server.Topic.Name.TopicId); + } + + /// + /// The subscription name for this endpoint under the given project id. Preserves the (possibly per-node) + /// subscription id computed during , just under the tenant's project. + /// + internal SubscriptionName SubscriptionNameFor(string projectId) + { + return projectId == _transport.ProjectId + ? Server.Subscription.Name + : new SubscriptionName(projectId, Server.Subscription.Name.SubscriptionId); + } + public PubsubEndpoint( string topicName, PubsubTransport transport, @@ -97,23 +122,60 @@ public async ValueTask SetupAsync(ILogger logger) return; } + // Only competing-consumer listeners get a per-node subscription so that each node + // load balances a distinct copy of the stream. A leader-pinned (or otherwise + // single-node) listener must read from one shared, cluster-stable subscription; + // otherwise every node creates its own subscription and Pub/Sub fans a copy of every + // message to each, breaking the single-consumer (leader-only) guarantee. + // + // Applied once here (before per-connection provisioning) so the mutated subscription id is shared by the + // default project and every tenant project. + if ((IsListener || IsDeadLetter) && !IsDeadLetter && ListenerScope == ListenerScope.CompetingConsumers) + { + Server.Subscription.Name = + Server.Subscription.Name.WithAssignedNodeNumber(_transport.AssignedNodeNumber); + } + + // Provision on the default/shared connection... + await provisionAsync(logger, _transport.DefaultClients); + + // ...and on each tenant's own project (broker-per-tenant, GH-3306). Each tenant is an independent GCP + // project, so the same logical topic/subscription must be created there too using the tenant's client. + if (isTenantAware()) + { + foreach (var tenant in _transport.Tenants) + { + await provisionAsync(logger, _transport.GetTenantClients(tenant)); + } + } + } + + private async Task provisionAsync(ILogger logger, PubsubClientSet clients) + { + if (clients.PublisherApiClient is null) + { + throw new WolverinePubsubTransportNotConnectedException(); + } + + var topicName = TopicNameFor(clients.ProjectId); + try { - await _transport.PublisherApiClient.CreateTopicAsync(new Topic + await clients.PublisherApiClient.CreateTopicAsync(new Topic { - TopicName = Server.Topic.Name, + TopicName = topicName, MessageRetentionDuration = Server.Topic.Options.MessageRetentionDuration }); } catch (RpcException ex) when (ex.StatusCode == StatusCode.AlreadyExists) { logger.LogInformation("{Uri}: Google Cloud Platform Pub/Sub topic \"{Topic}\" already exists", Uri, - Server.Topic.Name); + topicName); } catch (Exception ex) { logger.LogError(ex, "{Uri}: Error trying to initialize Google Cloud Platform Pub/Sub topic \"{Topic}\"", - Uri, Server.Topic.Name); + Uri, topicName); throw; } @@ -123,28 +185,19 @@ await _transport.PublisherApiClient.CreateTopicAsync(new Topic return; } - if (_transport.SubscriberApiClient is null) + if (clients.SubscriberApiClient is null) { throw new WolverinePubsubTransportNotConnectedException(); } + var subscriptionName = SubscriptionNameFor(clients.ProjectId); + try { - // Only competing-consumer listeners get a per-node subscription so that each node - // load balances a distinct copy of the stream. A leader-pinned (or otherwise - // single-node) listener must read from one shared, cluster-stable subscription; - // otherwise every node creates its own subscription and Pub/Sub fans a copy of every - // message to each, breaking the single-consumer (leader-only) guarantee. - if (!IsDeadLetter && ListenerScope == ListenerScope.CompetingConsumers) - { - Server.Subscription.Name = - Server.Subscription.Name.WithAssignedNodeNumber(_transport.AssignedNodeNumber); - } - var request = new Subscription { - SubscriptionName = Server.Subscription.Name, - TopicAsTopicName = Server.Topic.Name, + SubscriptionName = subscriptionName, + TopicAsTopicName = topicName, AckDeadlineSeconds = Server.Subscription.Options.AckDeadlineSeconds, EnableExactlyOnceDelivery = Server.Subscription.Options.EnableExactlyOnceDelivery, EnableMessageOrdering = Server.Subscription.Options.EnableMessageOrdering, @@ -168,23 +221,28 @@ await _transport.PublisherApiClient.CreateTopicAsync(new Topic request.Filter = Server.Subscription.Options.Filter; } - await _transport.SubscriberApiClient.CreateSubscriptionAsync(request); + await clients.SubscriberApiClient.CreateSubscriptionAsync(request); } catch (RpcException ex) when (ex.StatusCode == StatusCode.AlreadyExists) { logger.LogInformation("{Uri}: Google Cloud Platform Pub/Sub subscription \"{Subscription}\" already exists", - Uri, Server.Subscription.Name); + Uri, subscriptionName); } catch (Exception ex) { logger.LogError(ex, "{Uri}: Error trying to initialize Google Cloud Platform Pub/Sub subscription \"{Subscription}\" to topic \"{Topic}\"", - Uri, Server.Subscription.Name, Server.Topic.Name); + Uri, subscriptionName, topicName); throw; } } + private bool isTenantAware() + { + return _transport.Tenants.Any() && TenancyBehavior == TenancyBehavior.TenantAware; + } + public async ValueTask CheckAsync() { if ( @@ -307,22 +365,35 @@ public override ValueTask BuildListenerAsync(IWolverineRuntime runtim { EnvelopeMapper ??= BuildMapper(runtime); + if (!isTenantAware()) + { + return ValueTask.FromResult(buildListener(runtime, receiver, _transport.DefaultClients)); + } + + // Broker-per-tenant (GH-3306): the default listener consumes the shared project, and each tenant is + // consumed over its own project via a dedicated listener that stamps the tenant id onto inbound envelopes + // (mirrors the NATS / RabbitMQ / Azure Service Bus CompoundListener pattern). Pub/Sub acknowledges inline + // in the streaming callback, so per-envelope completion never routes back through CompoundListener. + var compound = new CompoundListener(Uri); + compound.Inner.Add(buildListener(runtime, receiver, _transport.DefaultClients)); + + foreach (var tenant in _transport.Tenants) + { + var tenantReceiver = new ReceiverWithRules(receiver, [new TenantIdRule(tenant.TenantId)]); + compound.Inner.Add(buildListener(runtime, tenantReceiver, _transport.GetTenantClients(tenant))); + } + + return ValueTask.FromResult(compound); + } + + private IListener buildListener(IWolverineRuntime runtime, IReceiver receiver, PubsubClientSet clients) + { if (Mode == EndpointMode.Inline) { - return ValueTask.FromResult(new InlinePubsubListener( - this, - _transport, - receiver, - runtime - )); + return new InlinePubsubListener(this, _transport, receiver, runtime, clients); } - return ValueTask.FromResult(new BatchedPubsubListener( - this, - _transport, - receiver, - runtime - )); + return new BatchedPubsubListener(this, _transport, receiver, runtime, clients); } public override DeadLetterStorageMode DeadLetterStorage => DeadLetterName.IsNotEmpty() @@ -358,9 +429,11 @@ public override bool TryBuildDeadLetterSender(IWolverineRuntime runtime, out ISe return false; } - internal async Task SendMessageAsync(Envelope envelope, ILogger logger) + internal async Task SendMessageAsync(Envelope envelope, ILogger logger, PubsubClientSet? clients = null) { - if (_transport.PublisherApiClient is null) + clients ??= _transport.DefaultClients; + + if (clients.PublisherApiClient is null) { throw new WolverinePubsubTransportNotConnectedException(); } @@ -378,9 +451,9 @@ internal async Task SendMessageAsync(Envelope envelope, ILogger logger) message.OrderingKey = envelope.GroupId ?? orderBy ?? message.OrderingKey; - await _transport.PublisherApiClient.PublishAsync(new PublishRequest + await clients.PublisherApiClient.PublishAsync(new PublishRequest { - TopicAsTopicName = Server.Topic.Name, + TopicAsTopicName = TopicNameFor(clients.ProjectId), Messages = { message } }); } @@ -417,6 +490,25 @@ protected override ISender CreateSender(IWolverineRuntime runtime) throw new WolverinePubsubTransportNotConnectedException(); } + if (isTenantAware()) + { + // Broker-per-tenant (GH-3306): route outbound sends by Envelope.TenantId. Both the default-fallback + // sender and every per-tenant sender MUST be the fire-and-forget InlinePubsubSender (not + // BatchedSender + PubsubSenderProtocol): TenantedSender deliberately does not implement + // ISenderRequiresCallback, so a BatchedSender underneath it would never have its outbox entries + // deleted. See GH-2361. + var defaultSender = new InlinePubsubSender(this, runtime, _transport.DefaultClients); + var tenantedSender = new TenantedSender(Uri, _transport.TenantedIdBehavior, defaultSender); + + foreach (var tenant in _transport.Tenants) + { + var tenantSender = new InlinePubsubSender(this, runtime, _transport.GetTenantClients(tenant)); + tenantedSender.RegisterSender(tenant.TenantId, tenantSender); + } + + return tenantedSender; + } + if (Mode == EndpointMode.Inline) { return new InlinePubsubSender(this, runtime); diff --git a/src/Transports/GCP/Wolverine.Pubsub/PubsubTransport.cs b/src/Transports/GCP/Wolverine.Pubsub/PubsubTransport.cs index 84f763d01..023217a91 100644 --- a/src/Transports/GCP/Wolverine.Pubsub/PubsubTransport.cs +++ b/src/Transports/GCP/Wolverine.Pubsub/PubsubTransport.cs @@ -4,6 +4,7 @@ using JasperFx.Core; using JasperFx.Descriptors; using Wolverine.Configuration; +using Wolverine.Pubsub.Internal; using Wolverine.Runtime; using Wolverine.Transports; @@ -26,6 +27,13 @@ public class PubsubTransport : BrokerTransport, IAsyncDisposable internal PublisherServiceApiClient? PublisherApiClient; internal SubscriberServiceApiClient? SubscriberApiClient; + /// + /// Registered tenants for the broker-per-tenant model (GH-3306). Each tenant is served by its own dedicated + /// GCP project/connection while sharing the endpoint topology. Keyed by tenant id. + /// + [IgnoreDescription] + public LightweightCache Tenants { get; } = new(); + /// /// Is this transport connection allowed to build and use response topic and subscription /// for just this node? @@ -57,18 +65,45 @@ public class PubsubTransport : BrokerTransport, IAsyncDisposable [IgnoreDescription] public Func? ConfigureSubscriberClientBuilder { get; set; } - public PubsubTransport() : base(ProtocolName, "Google Cloud Platform Pub/Sub", ["gcp", ProtocolName]) + /// + /// Build the Google Cloud Platform Pub/Sub transport. The single-string constructor sets the transport + /// protocol (broker name / URI scheme), NOT the project id — this is required so that + /// can spin up + /// named brokers via Activator.CreateInstance(typeof(PubsubTransport), name.Name). Set the + /// separately via UsePubsub/AddNamedPubsubBroker. + /// + public PubsubTransport(string protocol) : base(protocol, "Google Cloud Platform Pub/Sub", ["gcp", protocol]) { IdentifierDelimiter = "."; Topics = new LightweightCache(name => new PubsubEndpoint(name, this)); } - public PubsubTransport(string projectId) : this() + public PubsubTransport() : this(ProtocolName) { - ProjectId = projectId; } - public override Uri ResourceUri => new Uri("pubsub://" + ProjectId); + public override Uri ResourceUri => new Uri($"{Protocol}://" + ProjectId); + + /// + /// The resolved API client set for the default/shared connection. Senders and listeners that are not bound to a + /// specific tenant use this. + /// + internal PubsubClientSet DefaultClients => new() + { + ProjectId = ProjectId, + EmulatorDetection = EmulatorDetection, + PublisherApiClient = PublisherApiClient, + SubscriberApiClient = SubscriberApiClient, + ConfigureSubscriberClientBuilder = ConfigureSubscriberClientBuilder + }; + + /// + /// Resolve the API client set for a tenant (broker-per-tenant). Built during . + /// + internal PubsubClientSet GetTenantClients(PubsubTenant tenant) + { + return tenant.Clients ?? throw new WolverinePubsubTransportNotConnectedException(); + } public override string? DescribeEndpoint() { @@ -108,6 +143,21 @@ public override async ValueTask ConnectAsync(IWolverineRuntime runtime) AssignedNodeNumber = runtime.DurabilitySettings.AssignedNodeNumber; PublisherApiClient = await pubBuilder.BuildAsync(); SubscriberApiClient = await subApiBuilder.BuildAsync(); + + // Broker-per-tenant (GH-3306): every registered tenant builds its own dedicated client pair against its own + // GCP project. Credential/emulator hooks are seeded from the parent transport when the tenant did not set + // its own, so a tenant only re-points the axes it actually overrides. + foreach (var tenant in Tenants) + { + tenant.EmulatorDetection = tenant.EmulatorDetection == EmulatorDetection.None + ? EmulatorDetection + : tenant.EmulatorDetection; + tenant.ConfigurePublisherApiBuilder ??= ConfigurePublisherApiBuilder; + tenant.ConfigureSubscriberApiBuilder ??= ConfigureSubscriberApiBuilder; + tenant.ConfigureSubscriberClientBuilder ??= ConfigureSubscriberClientBuilder; + + await tenant.ConnectAsync(); + } } public override Endpoint? ReplyEndpoint() diff --git a/src/Transports/GCP/Wolverine.Pubsub/PubsubTransportExtensions.cs b/src/Transports/GCP/Wolverine.Pubsub/PubsubTransportExtensions.cs index b4386c66c..1f7a78088 100644 --- a/src/Transports/GCP/Wolverine.Pubsub/PubsubTransportExtensions.cs +++ b/src/Transports/GCP/Wolverine.Pubsub/PubsubTransportExtensions.cs @@ -13,11 +13,11 @@ public static class PubsubTransportExtensions /// /// /// - internal static PubsubTransport PubsubTransport(this WolverineOptions endpoints) + internal static PubsubTransport PubsubTransport(this WolverineOptions endpoints, BrokerName? name = null) { var transports = endpoints.As().Transports; - return transports.GetOrCreate(); + return transports.GetOrCreate(name); } /// @@ -50,6 +50,88 @@ public static PubsubConfiguration UsePubsub(this WolverineOptions endpoints, str return new PubsubConfiguration(transport, endpoints); } + /// + /// Add an additional, named broker connection to Google Cloud Platform Pub/Sub alongside the default one. + /// Endpoints created through the ...OnNamedBroker overloads are pinned to this broker. Note that the + /// Wolverine scheme for endpoints on a named broker is the broker name itself + /// (e.g. {name}://{projectId}/{topic}), not pubsub. + /// + /// + /// The unique name of the additional broker. + /// The GCP project id for this named broker. + /// + /// + public static PubsubConfiguration AddNamedPubsubBroker(this WolverineOptions endpoints, BrokerName name, + string projectId, Action? configure = null) + { + var transport = endpoints.PubsubTransport(name); + + transport.ProjectId = projectId ?? throw new ArgumentNullException(nameof(projectId)); + + configure?.Invoke(transport); + + return new PubsubConfiguration(transport, endpoints); + } + + /// + /// Listen for incoming messages at the designated Google Cloud Platform Pub/Sub topic on a named broker. + /// + /// + /// The name of the additional broker registered via . + /// The name of the Google Cloud Platform Pub/Sub topic + /// + /// Optional configuration for this Google Cloud Platform Pub/Sub endpoint. + /// + /// + public static PubsubTopicListenerConfiguration ListenToPubsubTopicOnNamedBroker( + this WolverineOptions endpoints, + BrokerName name, + string topicName, + Action? configure = null + ) + { + var transport = endpoints.PubsubTransport(name); + var topic = transport.Topics[transport.MaybeCorrectName(topicName)]; + + topic.EndpointName = topicName; + topic.IsListener = true; + + configure?.Invoke(topic); + + return new PubsubTopicListenerConfiguration(topic); + } + + /// + /// Publish the designated messages to a Google Cloud Platform Pub/Sub topic on a named broker. + /// + /// + /// The name of the additional broker registered via . + /// + /// + /// Optional configuration for this Google Cloud Platform Pub/Sub endpoint. + /// + /// + public static PubsubTopicSubscriberConfiguration ToPubsubTopicOnNamedBroker( + this IPublishToExpression publishing, + BrokerName name, + string topicName, + Action? configure = null + ) + { + var transports = publishing.As().Parent.Transports; + var transport = transports.GetOrCreate(name); + var topic = transport.Topics[transport.MaybeCorrectName(topicName)]; + + topic.EndpointName = topicName; + + configure?.Invoke(topic); + + // This is necessary unfortunately to hook up the subscription rules + publishing.To(topic.Uri); + + return new PubsubTopicSubscriberConfiguration(topic); + } + /// /// Listen for incoming messages at the designated Google Cloud Platform Pub/Sub topic by name. ///