Skip to content

Commit 086673b

Browse files
committed
Add SourceFlow.Cloud.GCP provider (Google Cloud Pub/Sub) with tests
New Google Cloud provider mirroring the AWS and Azure implementations, using the same cloud-agnostic abstractions (ICommandDispatcher/IEventDispatcher, ICommand/IEventRoutingConfiguration, IBusBootstrapConfiguration, the fluent BusConfigurationBuilder, IMessageEncryption, IIdempotencyService). Pub/Sub mapping: a command "queue" = a topic + pull subscription; an event "topic" = a topic + a pull subscription per subscriber. Unlike AWS/Azure, Pub/Sub has only topics and subscriptions, so no SNS->SQS forward trick is needed and the bootstrapper provisions resources at runtime (no static config). src/SourceFlow.Cloud.GCP: - PubSubCommandDispatcher / PubSubEventDispatcher (publish to topics) - PubSubCommandListener / PubSubEventListener (pull from subscriptions, ack) - GcpBusBootstrapper (IHostedService; creates topics + subscriptions, resolves short names to resource names, tolerant of AlreadyExists) - GcpHealthCheck, GcpKmsMessageEncryption (Cloud KMS envelope encryption), GcpTelemetryExtensions, PubSubClientFactory (honours PUBSUB_EMULATOR_HOST), UseSourceFlowGcp IoC extension (singleton in-memory idempotency + cleanup) - net8.0;net9.0;net10.0, full package metadata mirroring AWS tests/SourceFlow.Cloud.GCP.Tests: - 10 unit tests (options, IoC registration, dispatcher routing) — passing - 4 emulator integration tests (bootstrap provisioning, command/event publish->pull round trip, idempotent re-bootstrap) — passing against the Pub/Sub emulator; gated on PUBSUB_EMULATOR_HOST via SkippableFact CI: .github/workflows/GCP-Build.yml (build + unit + pack always; integration against the Pub/Sub emulator) and .github/gcp-emulator/docker-compose.yml.
1 parent 223c37a commit 086673b

24 files changed

Lines changed: 1993 additions & 0 deletions
Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,14 @@
1+
# Google Cloud Pub/Sub emulator for local development and CI.
2+
# The SourceFlow.Cloud.GCP clients auto-detect the emulator via PUBSUB_EMULATOR_HOST
3+
# (EmulatorDetection.EmulatorOrProduction). Unlike the Azure Service Bus emulator there is
4+
# no SQL backing store and no entity pre-declaration — the bootstrapper creates topics and
5+
# subscriptions at runtime.
6+
services:
7+
pubsub-emulator:
8+
image: gcr.io/google.com/cloudsdktool/google-cloud-cli:emulators
9+
command: >-
10+
gcloud beta emulators pubsub start
11+
--host-port=0.0.0.0:8085
12+
--project=sourceflow-emulator
13+
ports:
14+
- "8085:8085"

.github/workflows/GCP-Build.yml

Lines changed: 95 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,95 @@
1+
# Builds and tests the Google Cloud (Pub/Sub) cloud extension.
2+
# Unit tests have no dependencies; integration tests run against the Pub/Sub emulator
3+
# (the clients auto-detect PUBSUB_EMULATOR_HOST). Unlike the Azure Service Bus emulator the
4+
# Pub/Sub emulator needs no SQL backing store and no entity pre-declaration.
5+
6+
name: gcp-build
7+
8+
on:
9+
push:
10+
branches: [ "gcp_cloud" ]
11+
paths-ignore:
12+
- "**/*.md"
13+
- "**/*.gitignore"
14+
- "**/*.gitattributes"
15+
pull_request:
16+
branches: [ "gcp_cloud", "master" ]
17+
paths-ignore:
18+
- "**/*.md"
19+
- "**/*.gitignore"
20+
- "**/*.gitattributes"
21+
22+
jobs:
23+
build-and-unit-test:
24+
runs-on: ubuntu-latest
25+
steps:
26+
- uses: actions/checkout@v4
27+
- name: Setup .NET
28+
uses: actions/setup-dotnet@v4
29+
with:
30+
dotnet-version: |
31+
8.0.x
32+
9.0.x
33+
- name: Restore
34+
run: dotnet restore SourceFlow.Net.sln
35+
- name: Build
36+
run: dotnet build SourceFlow.Net.sln --configuration Release --no-restore
37+
- name: Run GCP unit tests
38+
run: >-
39+
dotnet test tests/SourceFlow.Cloud.GCP.Tests/SourceFlow.Cloud.GCP.Tests.csproj
40+
--configuration Release --no-build --verbosity normal
41+
--filter "Category=Unit"
42+
- name: Pack SourceFlow.Cloud.GCP
43+
run: dotnet pack src/SourceFlow.Cloud.GCP/SourceFlow.Cloud.GCP.csproj --configuration Release --no-build --output ./packages
44+
- name: Upload package artifact
45+
uses: actions/upload-artifact@v4
46+
with:
47+
name: gcp-nupkg
48+
path: ./packages/*.nupkg
49+
retention-days: 7
50+
51+
integration-test:
52+
runs-on: ubuntu-latest
53+
env:
54+
PUBSUB_EMULATOR_HOST: localhost:8085
55+
steps:
56+
- uses: actions/checkout@v4
57+
- name: Setup .NET
58+
uses: actions/setup-dotnet@v4
59+
with:
60+
dotnet-version: |
61+
8.0.x
62+
9.0.x
63+
64+
# GitHub `services:` cannot pass a multi-word command, so run the emulator via compose.
65+
- name: Start Pub/Sub emulator
66+
working-directory: .github/gcp-emulator
67+
run: docker compose up -d
68+
69+
- name: Wait for emulator
70+
working-directory: .github/gcp-emulator
71+
run: |
72+
echo "Waiting for the Pub/Sub emulator..."
73+
for i in $(seq 1 30); do
74+
if docker compose logs pubsub-emulator 2>&1 | grep -qi "Server started, listening"; then
75+
echo "Emulator is ready."; exit 0
76+
fi
77+
echo "Attempt $i/30 - not ready yet..."; sleep 3
78+
done
79+
echo "ERROR: emulator did not become ready"; docker compose logs; exit 1
80+
81+
- name: Restore & build
82+
run: |
83+
dotnet restore SourceFlow.Net.sln
84+
dotnet build SourceFlow.Net.sln --configuration Release --no-restore
85+
86+
- name: Run GCP integration tests (emulator)
87+
run: >-
88+
dotnet test tests/SourceFlow.Cloud.GCP.Tests/SourceFlow.Cloud.GCP.Tests.csproj
89+
--configuration Release --no-build --verbosity normal
90+
--filter "Category=Integration"
91+
92+
- name: Dump emulator logs on failure
93+
if: failure()
94+
working-directory: .github/gcp-emulator
95+
run: docker compose logs

SourceFlow.Net.sln

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,10 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "SourceFlow.Cloud.AWS.Tests"
3737
EndProject
3838
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "SourceFlow.Stores.EntityFramework.Tests", "tests\SourceFlow.Stores.EntityFramework.Tests\SourceFlow.Stores.EntityFramework.Tests.csproj", "{C56C4BC2-6BDC-EB3D-FC92-F9633530A501}"
3939
EndProject
40+
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "SourceFlow.Cloud.GCP", "src\SourceFlow.Cloud.GCP\SourceFlow.Cloud.GCP.csproj", "{8864B03A-260C-4CD4-B6F3-77797CF2EF06}"
41+
EndProject
42+
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "SourceFlow.Cloud.GCP.Tests", "tests\SourceFlow.Cloud.GCP.Tests\SourceFlow.Cloud.GCP.Tests.csproj", "{110350B5-0DDC-413F-852B-AA1B8022EEDC}"
43+
EndProject
4044
Global
4145
GlobalSection(SolutionConfigurationPlatforms) = preSolution
4246
Debug|Any CPU = Debug|Any CPU
@@ -119,6 +123,30 @@ Global
119123
{C56C4BC2-6BDC-EB3D-FC92-F9633530A501}.Release|x64.Build.0 = Release|Any CPU
120124
{C56C4BC2-6BDC-EB3D-FC92-F9633530A501}.Release|x86.ActiveCfg = Release|Any CPU
121125
{C56C4BC2-6BDC-EB3D-FC92-F9633530A501}.Release|x86.Build.0 = Release|Any CPU
126+
{8864B03A-260C-4CD4-B6F3-77797CF2EF06}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
127+
{8864B03A-260C-4CD4-B6F3-77797CF2EF06}.Debug|Any CPU.Build.0 = Debug|Any CPU
128+
{8864B03A-260C-4CD4-B6F3-77797CF2EF06}.Debug|x64.ActiveCfg = Debug|Any CPU
129+
{8864B03A-260C-4CD4-B6F3-77797CF2EF06}.Debug|x64.Build.0 = Debug|Any CPU
130+
{8864B03A-260C-4CD4-B6F3-77797CF2EF06}.Debug|x86.ActiveCfg = Debug|Any CPU
131+
{8864B03A-260C-4CD4-B6F3-77797CF2EF06}.Debug|x86.Build.0 = Debug|Any CPU
132+
{8864B03A-260C-4CD4-B6F3-77797CF2EF06}.Release|Any CPU.ActiveCfg = Release|Any CPU
133+
{8864B03A-260C-4CD4-B6F3-77797CF2EF06}.Release|Any CPU.Build.0 = Release|Any CPU
134+
{8864B03A-260C-4CD4-B6F3-77797CF2EF06}.Release|x64.ActiveCfg = Release|Any CPU
135+
{8864B03A-260C-4CD4-B6F3-77797CF2EF06}.Release|x64.Build.0 = Release|Any CPU
136+
{8864B03A-260C-4CD4-B6F3-77797CF2EF06}.Release|x86.ActiveCfg = Release|Any CPU
137+
{8864B03A-260C-4CD4-B6F3-77797CF2EF06}.Release|x86.Build.0 = Release|Any CPU
138+
{110350B5-0DDC-413F-852B-AA1B8022EEDC}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
139+
{110350B5-0DDC-413F-852B-AA1B8022EEDC}.Debug|Any CPU.Build.0 = Debug|Any CPU
140+
{110350B5-0DDC-413F-852B-AA1B8022EEDC}.Debug|x64.ActiveCfg = Debug|Any CPU
141+
{110350B5-0DDC-413F-852B-AA1B8022EEDC}.Debug|x64.Build.0 = Debug|Any CPU
142+
{110350B5-0DDC-413F-852B-AA1B8022EEDC}.Debug|x86.ActiveCfg = Debug|Any CPU
143+
{110350B5-0DDC-413F-852B-AA1B8022EEDC}.Debug|x86.Build.0 = Debug|Any CPU
144+
{110350B5-0DDC-413F-852B-AA1B8022EEDC}.Release|Any CPU.ActiveCfg = Release|Any CPU
145+
{110350B5-0DDC-413F-852B-AA1B8022EEDC}.Release|Any CPU.Build.0 = Release|Any CPU
146+
{110350B5-0DDC-413F-852B-AA1B8022EEDC}.Release|x64.ActiveCfg = Release|Any CPU
147+
{110350B5-0DDC-413F-852B-AA1B8022EEDC}.Release|x64.Build.0 = Release|Any CPU
148+
{110350B5-0DDC-413F-852B-AA1B8022EEDC}.Release|x86.ActiveCfg = Release|Any CPU
149+
{110350B5-0DDC-413F-852B-AA1B8022EEDC}.Release|x86.Build.0 = Release|Any CPU
122150
EndGlobalSection
123151
GlobalSection(SolutionProperties) = preSolution
124152
HideSolutionNode = FALSE
@@ -130,6 +158,8 @@ Global
130158
{C8765CB0-C453-0848-D98B-B0CF4E5D986F} = {02EA681E-C7D8-13C7-8484-4AC65E1B71E8}
131159
{0A833B33-8C55-4364-8D70-9A31994A6F61} = {653DCB25-EC82-421B-86F7-1DD8879B3926}
132160
{C56C4BC2-6BDC-EB3D-FC92-F9633530A501} = {653DCB25-EC82-421B-86F7-1DD8879B3926}
161+
{8864B03A-260C-4CD4-B6F3-77797CF2EF06} = {02EA681E-C7D8-13C7-8484-4AC65E1B71E8}
162+
{110350B5-0DDC-413F-852B-AA1B8022EEDC} = {653DCB25-EC82-421B-86F7-1DD8879B3926}
133163
EndGlobalSection
134164
GlobalSection(ExtensibilityGlobals) = postSolution
135165
SolutionGuid = {D02B8992-CC81-4194-BBF7-5EC40A96C698}
Lines changed: 132 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,132 @@
1+
# SourceFlow.Cloud.GCP
2+
3+
**Google Cloud integration for distributed command and event processing**
4+
5+
[![NuGet](https://img.shields.io/nuget/v/SourceFlow.Cloud.GCP.svg)](https://www.nuget.org/packages/SourceFlow.Cloud.GCP/)
6+
[![License](https://img.shields.io/badge/license-MIT-blue.svg)](LICENSE)
7+
8+
## Overview
9+
10+
SourceFlow.Cloud.GCP extends the SourceFlow.Net framework with Google Cloud integration, enabling distributed command and event processing using Google Cloud Pub/Sub and Cloud KMS. The fluent bus API is identical to the AWS and Azure providers — only the backing services change.
11+
12+
Google Cloud Pub/Sub has only **topics** and **subscriptions** (no separate queues). A command "queue" is modelled as a topic plus a pull subscription; an event "topic" is a topic plus a pull subscription per subscriber.
13+
14+
**Key Features:**
15+
- 🚀 Pub/Sub command dispatching (publish to topics, pull from subscriptions)
16+
- 📢 Pub/Sub event publishing with per-subscriber pull subscriptions
17+
- 🔐 Cloud KMS envelope encryption for sensitive data
18+
- ⚙️ Fluent bus configuration API
19+
- 🔄 Automatic resource provisioning (topics + subscriptions)
20+
- 📊 Built-in health checks and OpenTelemetry metrics
21+
- 🧪 Pub/Sub emulator support for local development
22+
23+
---
24+
25+
## Installation
26+
27+
```bash
28+
dotnet add package SourceFlow.Cloud.GCP
29+
```
30+
31+
**Prerequisites:** SourceFlow ≥ 2.0.0, Google Cloud SDK / Application Default Credentials, .NET 8.0 / 9.0 / 10.0.
32+
33+
---
34+
35+
## Quick Start
36+
37+
```csharp
38+
using SourceFlow.Cloud.GCP;
39+
40+
// Register SourceFlow core
41+
services.UseSourceFlow(typeof(Program).Assembly);
42+
43+
// Configure Google Cloud Pub/Sub messaging
44+
services.UseSourceFlowGcp(
45+
options => { options.ProjectId = "my-project"; },
46+
bus => bus
47+
.Send
48+
.Command<CreateOrderCommand>(q => q.Queue("orders"))
49+
.Command<ProcessPaymentCommand>(q => q.Queue("payments"))
50+
.Raise
51+
.Event<OrderCreatedEvent>(t => t.Topic("order-events"))
52+
.Event<PaymentProcessedEvent>(t => t.Topic("payment-events"))
53+
.Listen.To
54+
.CommandQueue("orders")
55+
.CommandQueue("payments")
56+
.Subscribe.To
57+
.Topic("order-events")
58+
.Topic("payment-events"));
59+
```
60+
61+
This registers GCP dispatchers, configures routing, starts the Pub/Sub pull listeners, and automatically provisions topics and subscriptions at startup.
62+
63+
---
64+
65+
## Configuration Options
66+
67+
| Option | Type | Default | Description |
68+
| --- | --- | --- | --- |
69+
| `ProjectId` | string | (required) | Google Cloud project that owns the topics/subscriptions |
70+
| `EnableCommandRouting` | bool | true | Enable command dispatching to topics |
71+
| `EnableEventRouting` | bool | true | Enable event publishing to topics |
72+
| `EnableCommandListener` | bool | true | Enable the command pull listener |
73+
| `EnableEventListener` | bool | true | Enable the event pull listener |
74+
| `MaxMessagesPerPull` | int | 10 | Messages requested per pull |
75+
| `AckDeadlineSeconds` | int | 60 | Ack deadline applied to subscriptions at bootstrap |
76+
| `SubscriptionSuffix` | string | `-sub` | Suffix used to derive a subscription id from a name |
77+
78+
---
79+
80+
## Resource Provisioning
81+
82+
The `GcpBusBootstrapper` runs as an `IHostedService` at startup and idempotently creates:
83+
84+
- **Topics** — one per command queue name and per event topic name.
85+
- **Pull subscriptions**`{name}-sub` for each listening command queue and each subscribed event topic.
86+
87+
All operations tolerate `AlreadyExists`, so it is safe to run on every startup.
88+
89+
---
90+
91+
## Message Encryption (Cloud KMS)
92+
93+
```csharp
94+
services.AddSingleton(new GcpKmsOptions
95+
{
96+
KeyName = "projects/my-project/locations/global/keyRings/my-ring/cryptoKeys/my-key"
97+
});
98+
services.AddSingleton<IMessageEncryption, GcpKmsMessageEncryption>();
99+
```
100+
101+
Envelope encryption: a random 256-bit data key encrypts the payload with AES-256-GCM, and Cloud KMS wraps (encrypts) the data key. Cloud KMS has no `GenerateDataKey` operation, so the data key is generated locally and wrapped via the KMS `Encrypt` call.
102+
103+
---
104+
105+
## Local Development (Pub/Sub emulator)
106+
107+
```bash
108+
gcloud beta emulators pubsub start --host-port=localhost:8085
109+
export PUBSUB_EMULATOR_HOST=localhost:8085
110+
```
111+
112+
The client libraries auto-detect `PUBSUB_EMULATOR_HOST` (via `EmulatorDetection.EmulatorOrProduction`). The bootstrapper creates topics/subscriptions in the emulator at startup — no manual setup required.
113+
114+
---
115+
116+
## Idempotency
117+
118+
- **In-memory (single instance)** — registered by default as a singleton with a background cleanup service.
119+
- **SQL-based (multi-instance / production)** — install `SourceFlow.Stores.EntityFramework` and call `services.AddSourceFlowIdempotency(connectionString)` before `UseSourceFlowGcp(...)`.
120+
121+
---
122+
123+
## Monitoring
124+
125+
- **Activity/Meter source:** `SourceFlow.Cloud.GCP` (`gcp.pubsub.commands.dispatched`, `gcp.pubsub.events.published`).
126+
- **Health check:** registered automatically; verifies Pub/Sub connectivity by listing topics in the project.
127+
128+
---
129+
130+
## License
131+
132+
MIT — see [LICENSE](LICENSE).
Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,48 @@
1+
using System;
2+
3+
namespace SourceFlow.Cloud.GCP.Configuration;
4+
5+
/// <summary>
6+
/// Configuration options for the SourceFlow Google Cloud Pub/Sub provider.
7+
/// </summary>
8+
public class GcpOptions
9+
{
10+
/// <summary>
11+
/// Google Cloud project id that owns the Pub/Sub topics and subscriptions.
12+
/// Required. When using the Pub/Sub emulator any non-empty value works.
13+
/// </summary>
14+
public string ProjectId { get; set; } = string.Empty;
15+
16+
/// <summary>Enable command dispatching to Pub/Sub topics.</summary>
17+
public bool EnableCommandRouting { get; set; } = true;
18+
19+
/// <summary>Enable event publishing to Pub/Sub topics.</summary>
20+
public bool EnableEventRouting { get; set; } = true;
21+
22+
/// <summary>Enable the background command-subscription pull listener.</summary>
23+
public bool EnableCommandListener { get; set; } = true;
24+
25+
/// <summary>Enable the background event-subscription pull listener.</summary>
26+
public bool EnableEventListener { get; set; } = true;
27+
28+
/// <summary>Maximum number of messages to request per pull.</summary>
29+
public int MaxMessagesPerPull { get; set; } = 10;
30+
31+
/// <summary>Ack deadline (seconds) applied to subscriptions created at bootstrap.</summary>
32+
public int AckDeadlineSeconds { get; set; } = 60;
33+
34+
/// <summary>Delay applied between pulls that return no messages, to avoid a tight loop.</summary>
35+
public TimeSpan EmptyPullDelay { get; set; } = TimeSpan.FromSeconds(1);
36+
37+
/// <summary>Maximum retry attempts for transient listener failures.</summary>
38+
public int MaxRetries { get; set; } = 3;
39+
40+
/// <summary>Base delay for exponential backoff on listener failures.</summary>
41+
public TimeSpan RetryDelay { get; set; } = TimeSpan.FromSeconds(1);
42+
43+
/// <summary>
44+
/// Suffix appended to a queue/topic name to derive its pull subscription id
45+
/// (e.g. queue <c>orders</c> → subscription <c>orders-sub</c>).
46+
/// </summary>
47+
public string SubscriptionSuffix { get; set; } = "-sub";
48+
}

0 commit comments

Comments
 (0)