diff --git a/README.md b/README.md index 3b4f5ab..093fdec 100644 --- a/README.md +++ b/README.md @@ -45,6 +45,8 @@ Repo features in more detail: manager from being continually polled. * Configuration is based on environment variables. * File upload and tracking examples. +* Server-sent events over MQTT. + * For more information, see [`docs/patterns/server-sent-events.md`](docs/patterns/server-sent-events.md) * Documentation on common data handling practices: * [Event Sourcing](docs/patterns/event-sourcing.md) @@ -100,7 +102,21 @@ Below are the recommended steps for using this as a template: 2. Write database accessors and/or service connectors based on your application's needs. 3. Prune database accessors and/or service connectors that your application does not need. -4. Consider revising/pruning the Markdown files such as this README or those located in the `docs/` directory. They +4. The example of Server-Sent Events needed a bespoke handler, the standard NSwag clients did not know how to support + event streams. This handler can be found in the `RedShirt.Example.Api.Interop` project, defined in + `EventStreamListener.cs`. + * The `EventStreamListener.cs` contains the abstract `EventStreamListener` and the applied + `MessagesEventStreamListener`, which sets some defaults. + * If you are implementing Server-Sent Events but not as `/messages` as this example does, then I would recommend + adjusting the comments described in `MessagesEventStreamListener`. + * If you are not implementing Server-Side Events in your applied application, then I would recommend deleting + `EventStreamListener.cs` outright. + * The pattern of emulating the generated swagger clients by throwing a `SwaggerException` required the manual + declaration of `SwaggerException` and `ProblemDetails` to avoid a chicken-and-egg problem on build where a + cold build without generated code failed because the manual code was expecting it. If you are removing + `EventStreamListener`, then you might also want to consider removing these manual classes and adjusting + `nswag.json` to generate them again. +5. Consider revising/pruning the Markdown files such as this README or those located in the `docs/` directory. They heavily assume that they are speaking for a general template and not for an applied application. ## Secret Manager diff --git a/RedShirt.Example.Api.slnx b/RedShirt.Example.Api.slnx index 7e4e0f0..07d7cbf 100644 --- a/RedShirt.Example.Api.slnx +++ b/RedShirt.Example.Api.slnx @@ -7,6 +7,7 @@ + @@ -76,8 +77,21 @@ Path="src\RedShirt.Example.Api.DataStores.Common.DapperMySql\RedShirt.Example.Api.DataStores.Common.DapperMySql.csproj"/> + + + + + + + + + @@ -143,4 +157,10 @@ + + + + diff --git a/docs/patterns/server-sent-events.md b/docs/patterns/server-sent-events.md new file mode 100644 index 0000000..eb1de4d --- /dev/null +++ b/docs/patterns/server-sent-events.md @@ -0,0 +1,368 @@ +# Client Events / Server-Sent Events + +This document talks a bit about this template's usage of Server-Sent Events to send live updates to connected clients. + +# Overview + +Server-Sent Events (SSE) send a stream of event messages to a client as opposed to having them poll for new information. + +This template uses the MQTT to coordinate messages between clients. Using an external service is necessary for a couple +of reasons: + +* There could be more than one instance of an API running in an environment. An in-process coordination system would not + have awareness of other instances. +* There could be other sources writing events directly to the external service (though this is not recommended for + single-responsibility reasons). + +## Template Example + +This template has a simple messaging example: + +* The POST endpoint at `/messages` can be used by a client to send events that will be read by any instance of the + client's user connected to the event-stream. +* The GET endpoint at `/messages/event-stream` can be used by a client to receive events related to their user. Users + will not receive messages intended for other users. + +# Configuration Options + +STUB + +# Special Cases + +## AWS IoT Endpoint Resolution + +During local testing, it was discovered that pointing the Broker URL didn't cut it for connecting to MiniStack in the +local Docker Compose stack. + +This issue is in part a product of my choice to use MiniStack's IoT emulator for local testing. However, I am choosing +to keep this feature in the template because it's possible (though not required) to resolve a broker URL using the +AWSSDK.IoT client if one is working in an AWS environment. It will return an IoT data-plane hostname that is stable for +a given account and region. That said, I'm also acknowledging that AWS deployments are not everyone's end goal. In the +same vein as the rest of this template, the AWSSDK.IoT operations have been quarantined in a separate +`RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws` project in order to limit the complexity of the removal. + +If you want to remove the use of the `AWSSDK.IoT` altogether, then the following steps are advised: + +1. Delete the `RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws` project. Off a cliff, it goes. +2. Remove the dependency injection setup issue that pops up in the root `RedShirt.Example.Api` project. +3. Do one of the following in the `RedShirt.Example.Api.ClientEvents.Library.Mqtt` project to address the need for an + implementation of `IMqttBrokerUrlResolver` in `MqttClientFactory`: + * Delete the `IMqttBrokerUrlResolver` altogether, and resolve any build errors that come up from it. I would + recommend this path, as apparently there's no other system outside of AWS that uses a resolution endpoint (except + perhaps if you insisted on also storing the broker URL in a secret manager?). + * If you decide to take this path, then in `` you could probably just fold `ResolveBrokerTargetInnerAsync` into + ResolveBrokerTargetAsync. + * Add in a quick stub implementation of the `IMqttBrokerUrlResolver` interface. + * The implementation can just return an object with a null URL. + * Be sure to set configuration so that this path is never accessed. +4. The `AWSSDK.IoT` library was required to locally test MQTT using MiniStack, so you will need to adjust your local + compose stack to use a different server. The following are some potential options (these haven't been explored in + detail, MiniStack was deemed good enough for the time being): + * Mosquitto: `eclipse-mosquitto` image + * According to Google, this has the lowest resource footprint of the immediate options. + * EMQX: `emqx/emqx` image + * HiveMQ (Community Edition): `hivemq/hivemq-ce` image + * RabbitMQ (not enabled by default, needs to be explicitly enabled) +5. Update configuration in local `test/local/docker-compose.yaml` file to point to the new server solution. + * You may need to also set credentials in secret managers. + +# Other Technologies + +Server-Sent Events at the API level are a new feature in .NET 10. However, it was not the first iteration of the overall +pattern of streaming events. + +## SignalR + +SignalR is a historically-significant library for pushing event data to connected clients in real time. + +I chose not to use SignalR in this template for the following reasons: + +* SignalR supports bidirectional communication, which seemed like more complexity than I needed for my basic goal of + reading events. +* SignalR apparently has more of a resource footprint. + +With these reasons in mind, I would only pivot to SignalR if I had a specific case that required it. + +# Client Examples + +Reading from an event stream is a different challenge then a flat HTTP request, so here are several generated code +samples for interacting with them. + +## C# + +### Interop + +A sample listener has been written for the `RedShirt.Example.Api.Interop` project, called `EventStreamListener`. + +Example usage: + +```csharp +using var httpClient = new HttpClient(); +httpClient.DefaultRequestHeaders.Authorization = + new AuthenticationHeaderValue("Bearer", accessToken); + +var listener = new MessagesEventStreamListener("https://api.example.com", httpClient); + +await listener.StreamEventsAsync( + // Can use an Action or Func + message => Console.WriteLine($"[message] {message}"), + cancellationToken); +``` + +### Standalone + +Use this approach when you do not want a dependency on `RedShirt.Example.Api.Interop`. It uses `HttpClient` to open +`GET /messages/event-stream`, reads the response as a stream, and parses SSE blocks manually (same event name and plain +text `data` payload as the JavaScript example). + +```csharp +using System.Net.Http.Headers; +using System.Text; + +static async Task ListenToMessageStreamAsync( + string apiBaseUrl, + string accessToken, + Action? onEvent = null, + CancellationToken cancellationToken = default) +{ + onEvent ??= message => Console.WriteLine($"[message] {message}"); + + var baseUrl = apiBaseUrl.TrimEnd('/'); + using var request = new HttpRequestMessage(HttpMethod.Get, $"{baseUrl}/messages/event-stream"); + request.Headers.Authorization = new AuthenticationHeaderValue("Bearer", accessToken); + request.Headers.Accept.Add(new MediaTypeWithQualityHeaderValue("text/event-stream")); + + using var httpClient = new HttpClient(); + using var response = await httpClient.SendAsync( + request, + HttpCompletionOption.ResponseHeadersRead, + cancellationToken); + + if (!response.IsSuccessStatusCode) + { + var errorBody = response.Content == null + ? string.Empty + : await response.Content.ReadAsStringAsync(cancellationToken); + throw new HttpRequestException( + $"HTTP {(int)response.StatusCode}: {errorBody}", + null, + response.StatusCode); + } + + if (response.Content == null) + { + return; + } + + await using var stream = await response.Content.ReadAsStreamAsync(cancellationToken); + using var reader = new StreamReader(stream, Encoding.UTF8); + + string? eventName = null; + var dataLines = new List(); + + while (!cancellationToken.IsCancellationRequested) + { + var line = await reader.ReadLineAsync(cancellationToken); + if (line == null) + { + break; + } + + if (line.Length == 0) + { + if (dataLines.Count > 0) + { + var resolvedEventName = eventName ?? "message"; + if (string.Equals(resolvedEventName, "message", StringComparison.Ordinal)) + { + onEvent(string.Join('\n', dataLines)); + } + } + + eventName = null; + dataLines.Clear(); + continue; + } + + if (line.StartsWith(':')) + { + continue; + } + + if (line.StartsWith("event:", StringComparison.Ordinal)) + { + eventName = line["event:".Length..].Trim(); + continue; + } + + if (line.StartsWith("data:", StringComparison.Ordinal)) + { + dataLines.Add(line["data:".Length..].TrimStart()); + } + } + + cancellationToken.ThrowIfCancellationRequested(); +} + +// Example: +// await ListenToMessageStreamAsync("http://localhost:8080", accessToken); +``` + +## JavaScript + +The example stream lives at `GET /messages/event-stream`. Each event uses the SSE event name `message`, and the +`data` field is the plain message text (not JSON). The endpoint requires a bearer token with the `api:read` scope. + +Browsers cannot set custom headers on the built-in `EventSource` API, so the example below uses `fetch` and reads the +response body as a stream. This works in modern browsers and in Node.js 18+. + +```javascript +async function listenToMessageStream( + apiBaseUrl, + accessToken, + onEvent = (eventName, data) => { + console.log(`[${eventName}] ${data}`); + }, +) { + const url = `${apiBaseUrl.replace(/\/$/, "")}/messages/event-stream`; + + const response = await fetch(url, { + headers: { + Authorization: `Bearer ${accessToken}`, + Accept: "text/event-stream", + }, + }); + + if (!response.ok) { + throw new Error(`HTTP ${response.status}: ${await response.text()}`); + } + + const reader = response.body.getReader(); + const decoder = new TextDecoder(); + let buffer = ""; + + console.log(`Listening on ${url} (Ctrl+C to stop)...`); + + while (true) { + const { done, value } = await reader.read(); + if (done) { + break; + } + + buffer += decoder.decode(value, { stream: true }); + + let eventBoundary; + while ((eventBoundary = buffer.indexOf("\n\n")) >= 0) { + const block = buffer.slice(0, eventBoundary); + buffer = buffer.slice(eventBoundary + 2); + + let eventName = "message"; + const dataLines = []; + + for (const line of block.split("\n")) { + if (line.startsWith("event:")) { + eventName = line.slice("event:".length).trim(); + } else if (line.startsWith("data:")) { + dataLines.push(line.slice("data:".length).trimStart()); + } + } + + if (dataLines.length > 0) { + onEvent(eventName, dataLines.join("\n")); + } + } + } +} + +// Example: +// listenToMessageStream("http://localhost:8080", process.env.API_JWT_TOKEN); +``` + +To exercise the stream locally, open a terminal running the listener above, then publish a message with +`POST /messages` (requires `api:write`). The listener should print lines like `[message] hello from mqtt`. + +## Python + +The local-testing scripts in the `test/local/scripts/messages/` directory has a Python-based example of listening for +events. + +## TypeScript + +The TypeScript example uses the same `fetch`-and-stream parsing approach as the JavaScript example above, with typed +parameters and a stricter null check on `response.body`. It runs in Node.js 18+ (with DOM lib types) or in a browser +build that provides `fetch`. + +```typescript +async function listenToMessageStream( + apiBaseUrl: string, + accessToken: string, + onEvent: (eventName: string, data: string) => void = (eventName, data) => { + console.log(`[${eventName}] ${data}`); + }, +): Promise { + const url = `${apiBaseUrl.replace(/\/$/, "")}/messages/event-stream`; + + const response = await fetch(url, { + headers: { + Authorization: `Bearer ${accessToken}`, + Accept: "text/event-stream", + }, + }); + + if (!response.ok) { + throw new Error(`HTTP ${response.status}: ${await response.text()}`); + } + + if (!response.body) { + throw new Error("Response body is null."); + } + + const reader = response.body.getReader(); + const decoder = new TextDecoder(); + let buffer = ""; + + console.log(`Listening on ${url} (Ctrl+C to stop)...`); + + while (true) { + const { done, value } = await reader.read(); + if (done) { + break; + } + + buffer += decoder.decode(value, { stream: true }); + + let eventBoundary: number; + while ((eventBoundary = buffer.indexOf("\n\n")) >= 0) { + const block = buffer.slice(0, eventBoundary); + buffer = buffer.slice(eventBoundary + 2); + + let eventName = "message"; + const dataLines: string[] = []; + + for (const line of block.split("\n")) { + if (line.startsWith("event:")) { + eventName = line.slice("event:".length).trim(); + } else if (line.startsWith("data:")) { + dataLines.push(line.slice("data:".length).trimStart()); + } + } + + if (dataLines.length > 0) { + onEvent(eventName, dataLines.join("\n")); + } + } + } +} + +// Example: +// listenToMessageStream("http://localhost:8080", process.env.API_JWT_TOKEN!); +``` + +# Other Resources + +The following links have useful reading material relating to Server-Sent Events and MQTT: + +* https://milanjovanovic.tech/blog/server-sent-events-in-aspnetcore-and-dotnet-10 +* https://docs.aws.amazon.com/whitepapers/latest/designing-mqtt-topics-aws-iot-core/mqtt-design-best-practices.html +* https://github.com/dotnet/MQTTnet/wiki/Tips-and-Tricks +* https://www.hivemq.com/blog/mqtt-essentials-part-5-mqtt-topics-best-practices/ \ No newline at end of file diff --git a/src/RedShirt.Example.Api.ClientEvents.Domains.Example/Extensions/ServiceCollectionExtensions.cs b/src/RedShirt.Example.Api.ClientEvents.Domains.Example/Extensions/ServiceCollectionExtensions.cs new file mode 100644 index 0000000..507447f --- /dev/null +++ b/src/RedShirt.Example.Api.ClientEvents.Domains.Example/Extensions/ServiceCollectionExtensions.cs @@ -0,0 +1,14 @@ +using Microsoft.Extensions.DependencyInjection; +using RedShirt.Example.Api.ClientEvents.Domains.Example.Services; + +namespace RedShirt.Example.Api.ClientEvents.Domains.Example.Extensions; + +public static class ServiceCollectionExtensions +{ + public static IServiceCollection AddExampleClientEventsDomain(this IServiceCollection services) + { + return services + .AddSingleton() + .AddSingleton(); + } +} \ No newline at end of file diff --git a/src/RedShirt.Example.Api.ClientEvents.Domains.Example/Models/ExampleMessageModel.cs b/src/RedShirt.Example.Api.ClientEvents.Domains.Example/Models/ExampleMessageModel.cs new file mode 100644 index 0000000..1fb04c3 --- /dev/null +++ b/src/RedShirt.Example.Api.ClientEvents.Domains.Example/Models/ExampleMessageModel.cs @@ -0,0 +1,6 @@ +namespace RedShirt.Example.Api.ClientEvents.Domains.Example.Models; + +public sealed class ExampleMessageModel +{ + public required string Message { get; init; } +} \ No newline at end of file diff --git a/src/RedShirt.Example.Api.ClientEvents.Domains.Example/RedShirt.Example.Api.ClientEvents.Domains.Example.csproj b/src/RedShirt.Example.Api.ClientEvents.Domains.Example/RedShirt.Example.Api.ClientEvents.Domains.Example.csproj new file mode 100644 index 0000000..8bc2da7 --- /dev/null +++ b/src/RedShirt.Example.Api.ClientEvents.Domains.Example/RedShirt.Example.Api.ClientEvents.Domains.Example.csproj @@ -0,0 +1,17 @@ + + + + net10.0 + enable + enable + + + + + + + + + + + diff --git a/src/RedShirt.Example.Api.ClientEvents.Domains.Example/Services/ExampleMessageReceiveService.cs b/src/RedShirt.Example.Api.ClientEvents.Domains.Example/Services/ExampleMessageReceiveService.cs new file mode 100644 index 0000000..15ded81 --- /dev/null +++ b/src/RedShirt.Example.Api.ClientEvents.Domains.Example/Services/ExampleMessageReceiveService.cs @@ -0,0 +1,27 @@ +using RedShirt.Example.Api.ClientEvents.Domains.Example.Models; +using RedShirt.Example.Api.ClientEvents.Domains.Example.Utilities; +using RedShirt.Example.Api.ClientEvents.Library.Core.Services; +using System.Runtime.CompilerServices; + +namespace RedShirt.Example.Api.ClientEvents.Domains.Example.Services; + +public interface IExampleMessageReceiveService +{ + IAsyncEnumerable ReceiveForUserAsync(string userId, + CancellationToken cancellationToken = default); +} + +internal sealed class ExampleMessageReceiveService(IApiClientEventReceiver eventReceiver) + : IExampleMessageReceiveService +{ + public async IAsyncEnumerable ReceiveForUserAsync(string userId, + [EnumeratorCancellation] + CancellationToken cancellationToken = default) + { + var topic = ExampleMessageTopicNames.ForUser(userId); + await foreach (var received in eventReceiver.ReceiveAsync([topic], cancellationToken)) + { + yield return received.Payload; + } + } +} \ No newline at end of file diff --git a/src/RedShirt.Example.Api.ClientEvents.Domains.Example/Services/ExampleMessageSendService.cs b/src/RedShirt.Example.Api.ClientEvents.Domains.Example/Services/ExampleMessageSendService.cs new file mode 100644 index 0000000..52cd6b1 --- /dev/null +++ b/src/RedShirt.Example.Api.ClientEvents.Domains.Example/Services/ExampleMessageSendService.cs @@ -0,0 +1,25 @@ +using RedShirt.Example.Api.ClientEvents.Domains.Example.Models; +using RedShirt.Example.Api.ClientEvents.Domains.Example.Utilities; +using RedShirt.Example.Api.ClientEvents.Library.Core.Models; +using RedShirt.Example.Api.ClientEvents.Library.Core.Services; + +namespace RedShirt.Example.Api.ClientEvents.Domains.Example.Services; + +public interface IExampleMessageSendService +{ + Task SendToUserAsync(string userId, ExampleMessageModel message, CancellationToken cancellationToken = default); +} + +internal sealed class ExampleMessageSendService(IApiClientEventSender eventSender) + : IExampleMessageSendService +{ + public Task SendToUserAsync(string userId, ExampleMessageModel message, + CancellationToken cancellationToken = default) + { + return eventSender.SendAsync(new ApiClientEventSendRequest + { + Payload = message, + Topic = ExampleMessageTopicNames.ForUser(userId) + }, cancellationToken); + } +} \ No newline at end of file diff --git a/src/RedShirt.Example.Api.ClientEvents.Domains.Example/Utilities/ExampleMessageTopicNames.cs b/src/RedShirt.Example.Api.ClientEvents.Domains.Example/Utilities/ExampleMessageTopicNames.cs new file mode 100644 index 0000000..32711bd --- /dev/null +++ b/src/RedShirt.Example.Api.ClientEvents.Domains.Example/Utilities/ExampleMessageTopicNames.cs @@ -0,0 +1,9 @@ +namespace RedShirt.Example.Api.ClientEvents.Domains.Example.Utilities; + +internal static class ExampleMessageTopicNames +{ + public static string ForUser(string userId) + { + return $"example-message/user/{userId}"; + } +} \ No newline at end of file diff --git a/src/RedShirt.Example.Api.ClientEvents.Library.Core/Exceptions/ApiClientEventsException.cs b/src/RedShirt.Example.Api.ClientEvents.Library.Core/Exceptions/ApiClientEventsException.cs new file mode 100644 index 0000000..70ff03c --- /dev/null +++ b/src/RedShirt.Example.Api.ClientEvents.Library.Core/Exceptions/ApiClientEventsException.cs @@ -0,0 +1,27 @@ +namespace RedShirt.Example.Api.ClientEvents.Library.Core.Exceptions; + +/// +/// Classified failure from a client-events operation. +/// +public sealed class ApiClientEventsException : Exception +{ + /// + /// When true, a retry wrapper inside the client-events layer has already exhausted retries + /// for the underlying cause; outer retry layers should not retry again. + /// + public required bool IsHandled { get; init; } + + /// + /// When true, suggests a possible transient or environmental cause could be resolved outside the application + /// process (with an infrastructure change, for example) without restarting the application. + /// + public required bool CouldBeTransient { get; init; } + + public ApiClientEventsException(Exception innerException) : base(innerException.Message, innerException) + { + } + + public ApiClientEventsException(string message) : base(message) + { + } +} \ No newline at end of file diff --git a/src/RedShirt.Example.Api.ClientEvents.Library.Core/Models/ApiClientEventReceived.cs b/src/RedShirt.Example.Api.ClientEvents.Library.Core/Models/ApiClientEventReceived.cs new file mode 100644 index 0000000..a2d3f1e --- /dev/null +++ b/src/RedShirt.Example.Api.ClientEvents.Library.Core/Models/ApiClientEventReceived.cs @@ -0,0 +1,12 @@ +namespace RedShirt.Example.Api.ClientEvents.Library.Core.Models; + +/// +/// A client event received from the transport layer. +/// +/// Event body type. +public sealed class ApiClientEventReceived +{ + public required string Topic { get; init; } + + public required TPayload Payload { get; init; } +} \ No newline at end of file diff --git a/src/RedShirt.Example.Api.ClientEvents.Library.Core/Models/ApiClientEventSendRequest.cs b/src/RedShirt.Example.Api.ClientEvents.Library.Core/Models/ApiClientEventSendRequest.cs new file mode 100644 index 0000000..56e26b8 --- /dev/null +++ b/src/RedShirt.Example.Api.ClientEvents.Library.Core/Models/ApiClientEventSendRequest.cs @@ -0,0 +1,15 @@ +namespace RedShirt.Example.Api.ClientEvents.Library.Core.Models; + +/// +/// Payload and optional routing metadata for a client event publish operation. +/// +/// Event body type. +public sealed class ApiClientEventSendRequest +{ + public required TPayload Payload { get; init; } + + /// + /// Optional MQTT topic. When omitted, the implementation applies its default routing. + /// + public string? Topic { get; init; } +} \ No newline at end of file diff --git a/src/RedShirt.Example.Api.ClientEvents.Library.Core/RedShirt.Example.Api.ClientEvents.Library.Core.csproj b/src/RedShirt.Example.Api.ClientEvents.Library.Core/RedShirt.Example.Api.ClientEvents.Library.Core.csproj new file mode 100644 index 0000000..e0ad4b5 --- /dev/null +++ b/src/RedShirt.Example.Api.ClientEvents.Library.Core/RedShirt.Example.Api.ClientEvents.Library.Core.csproj @@ -0,0 +1,9 @@ + + + + net10.0 + enable + enable + + + diff --git a/src/RedShirt.Example.Api.ClientEvents.Library.Core/Services/IApiClientEventReceiver.cs b/src/RedShirt.Example.Api.ClientEvents.Library.Core/Services/IApiClientEventReceiver.cs new file mode 100644 index 0000000..9b15d35 --- /dev/null +++ b/src/RedShirt.Example.Api.ClientEvents.Library.Core/Services/IApiClientEventReceiver.cs @@ -0,0 +1,21 @@ +using RedShirt.Example.Api.ClientEvents.Library.Core.Models; + +namespace RedShirt.Example.Api.ClientEvents.Library.Core.Services; + +/// +/// Streams typed client events. Implementations are expected to use MQTT as the transport; +/// topic names should be MQTT-shaped (for example domain/entity/id). +/// +/// Event body type. +public interface IApiClientEventReceiver +{ + /// + /// Streams client events from the transport. When is null or empty, + /// no topic filter is applied. Otherwise only events published on the listed MQTT-shaped topics are yielded. + /// + /// Optional MQTT topic filter list. + /// Token used to cancel the stream. + IAsyncEnumerable> ReceiveAsync( + IReadOnlyList? topics = null, + CancellationToken cancellationToken = default); +} \ No newline at end of file diff --git a/src/RedShirt.Example.Api.ClientEvents.Library.Core/Services/IApiClientEventSender.cs b/src/RedShirt.Example.Api.ClientEvents.Library.Core/Services/IApiClientEventSender.cs new file mode 100644 index 0000000..ba444fd --- /dev/null +++ b/src/RedShirt.Example.Api.ClientEvents.Library.Core/Services/IApiClientEventSender.cs @@ -0,0 +1,17 @@ +using RedShirt.Example.Api.ClientEvents.Library.Core.Models; + +namespace RedShirt.Example.Api.ClientEvents.Library.Core.Services; + +/// +/// Publishes typed client events. Implementations are expected to use MQTT as the transport; +/// topic names should be MQTT-shaped (for example domain/entity/id). +/// +/// Event body type. +public interface IApiClientEventSender +{ + /// + /// Publishes a client event. When is set, + /// it should be an MQTT-shaped topic understood by the implementation. + /// + Task SendAsync(ApiClientEventSendRequest request, CancellationToken cancellationToken = default); +} \ No newline at end of file diff --git a/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws/Extensions/ServiceCollectionExtensions.cs b/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws/Extensions/ServiceCollectionExtensions.cs new file mode 100644 index 0000000..23608a9 --- /dev/null +++ b/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws/Extensions/ServiceCollectionExtensions.cs @@ -0,0 +1,28 @@ +using Amazon.IoT; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; +using RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws.Services; +using RedShirt.Example.Api.ClientEvents.Library.Mqtt.Services; +using RedShirt.Example.Api.Common.Aws.Extensions; + +namespace RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws.Extensions; + +public static class ServiceCollectionExtensions +{ + private const string ConfigurationSectionName = + $"{Mqtt.Extensions.ServiceCollectionExtensions.ConfigurationSectionName}:AWS"; + + public static IServiceCollection AddAwsMqttClientEvents(this IServiceCollection services, + IConfiguration configuration) + { + services + .AddAwsServiceWithLocalSupport() + .Configure( + configuration.GetSection(ConfigurationSectionName)); + + services.TryAddSingleton(); + + return services; + } +} \ No newline at end of file diff --git a/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws/Globals.cs b/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws/Globals.cs new file mode 100644 index 0000000..005a7da --- /dev/null +++ b/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws/Globals.cs @@ -0,0 +1,4 @@ +using System.Runtime.CompilerServices; + +[assembly: InternalsVisibleTo("DynamicProxyGenAssembly2")] +[assembly: InternalsVisibleTo("RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws.UnitTests")] \ No newline at end of file diff --git a/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws/RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws.csproj b/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws/RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws.csproj new file mode 100644 index 0000000..13b6433 --- /dev/null +++ b/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws/RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws.csproj @@ -0,0 +1,20 @@ + + + + net10.0 + enable + enable + + + + + + + + + + + + + + diff --git a/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws/Services/AwsIotMqttBrokerUrlResolver.cs b/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws/Services/AwsIotMqttBrokerUrlResolver.cs new file mode 100644 index 0000000..064c589 --- /dev/null +++ b/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws/Services/AwsIotMqttBrokerUrlResolver.cs @@ -0,0 +1,115 @@ +using Amazon.IoT; +using Amazon.IoT.Model; +using Microsoft.Extensions.Options; +using RedShirt.Example.Api.ClientEvents.Library.Core.Exceptions; +using RedShirt.Example.Api.ClientEvents.Library.Mqtt.Models; +using RedShirt.Example.Api.ClientEvents.Library.Mqtt.Services; + +namespace RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws.Services; + +/// +/// AWS IoT implementation of that discovers the broker connection target +/// from IoT Core DescribeEndpoint at runtime. +/// +/// +/// +/// Runtime resolution is optional. The MQTT client factory uses this resolver only when +/// ClientEvents:Mqtt:ResolveBrokerAddressExternally is true. In a deployed AWS +/// environment you can set that flag to false and supply the IoT data-plane endpoint as +/// ClientEvents:Mqtt:BrokerUrl instead (for example from IaC output, SSM, or an environment variable). +/// The hostname returned by DescribeEndpoint (iot:Data-ATS) is stable for a given account and +/// region, so a configured BrokerUrl is equivalent to what this resolver would discover on startup. +/// Runtime resolution is a convenience that avoids plumbing the endpoint through configuration; it is not +/// required for production connectivity when DNS resolves to IoT Core directly. +/// +/// +/// When resolution is enabled in real AWS: calling IoT DescribeEndpoint obtains the +/// account-specific data-plane hostname and builds a WebSocket broker URL from it. With +/// unset, the resolver connects directly to that hostname, +/// which matches the normal production path described above. +/// +/// +/// Local Docker / MiniStack only: and the optional +/// HTTP Host header override. Inside Docker Compose the DescribeEndpoint hostname often does not resolve to +/// the LocalStack/MiniStack container, so a static BrokerUrl alone is insufficient. TCP must connect to a +/// configured host (for example ministack) while the upgrade still advertises the IoT endpoint hostname. +/// This split connect/host behaviour is what makes runtime resolution (or equivalent manual wiring) necessary +/// locally but not in AWS. +/// +/// +/// Production AWS WebSocket clients also require wss:// and SigV4 signing on the upgrade; this resolver +/// covers endpoint discovery and URL assembly only. +/// +/// +internal sealed class AwsIotMqttBrokerUrlResolver( + IAmazonIoT amazonIoT, + IOptions configuration) : IMqttBrokerUrlResolver +{ + private const string DefaultWebSocketPath = "/mqtt"; + + private static Uri BuildUriFromHostPort(string hostPort, string path) + { + if (Uri.TryCreate($"ws://{hostPort}{path}", UriKind.Absolute, out var uri)) + { + return uri; + } + + throw new ApiClientEventsException($"IoT endpoint address '{hostPort}' is not a valid host:port value.") + { + CouldBeTransient = false, + IsHandled = false + }; + } + + public async Task ResolveBrokerUrlAsync(CancellationToken cancellationToken) + { + var describeResponse = await amazonIoT.DescribeEndpointAsync(new DescribeEndpointRequest + { + EndpointType = "iot:Data-ATS" + }, cancellationToken); + + if (string.IsNullOrWhiteSpace(describeResponse.EndpointAddress)) + { + throw new ApiClientEventsException("IoT DescribeEndpoint did not return an endpoint address.") + { + CouldBeTransient = true, + IsHandled = false + }; + } + + var endpointAddress = describeResponse.EndpointAddress; + var endpointUri = BuildUriFromHostPort(endpointAddress, DefaultWebSocketPath); + var brokerConnectHost = configuration.Value.BrokerConnectHost; + var connectHost = string.IsNullOrWhiteSpace(brokerConnectHost) + ? endpointUri.Host + : brokerConnectHost; + + var brokerUrl = new UriBuilder(endpointUri.Scheme, connectHost, endpointUri.Port, endpointUri.AbsolutePath) + .Uri + .ToString(); + + var needsHostHeader = !connectHost.Equals(endpointUri.Host, StringComparison.OrdinalIgnoreCase); + return new MqttBrokerTarget + { + BrokerUrl = brokerUrl, + WebSocketHostHeader = needsHostHeader ? endpointAddress : null + }; + } + + internal sealed class ConfigurationModel + { + /// + /// Optional TCP/WebSocket connect host when resolving via IoT DescribeEndpoint. + /// + /// + /// Intended for local Docker / MiniStack testing only, when + /// ResolveBrokerAddressExternally is true. Leave unset in real AWS, where the + /// DescribeEndpoint hostname is both the advertised and connectable target. When set, the resolver + /// TCP-connects to this host and sends the DescribeEndpoint address as the HTTP Host header (for + /// example when *.localhost or *.ministack does not resolve to the MiniStack container from + /// the API container). Has no effect when runtime resolution is disabled and + /// ClientEvents:Mqtt:BrokerUrl is used instead. + /// + public string? BrokerConnectHost { get; init; } + } +} \ No newline at end of file diff --git a/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt/Extensions/ServiceCollectionExtensions.cs b/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt/Extensions/ServiceCollectionExtensions.cs new file mode 100644 index 0000000..f3f3063 --- /dev/null +++ b/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt/Extensions/ServiceCollectionExtensions.cs @@ -0,0 +1,34 @@ +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; +using RedShirt.Example.Api.ClientEvents.Library.Core.Services; +using RedShirt.Example.Api.ClientEvents.Library.Mqtt.Factories; +using RedShirt.Example.Api.ClientEvents.Library.Mqtt.Services; +using RedShirt.Example.Api.ClientEvents.Library.Mqtt.Services.Resilience; +using RedShirt.Example.Api.Common.Extensions; + +namespace RedShirt.Example.Api.ClientEvents.Library.Mqtt.Extensions; + +public static class ServiceCollectionExtensions +{ + public const string ConfigurationSectionName = "ClientEvents:Mqtt"; + + public static IServiceCollection AddMqttClientEvents(this IServiceCollection services, + IConfiguration configuration) + { + services + .AddCommonServices() + .Configure( + configuration.GetSection(ConfigurationSectionName)) + .Configure( + configuration.GetSection(ConfigurationSectionName)); + + services.TryAddSingleton(); + services.TryAddSingleton(); + services.TryAddSingleton(); + services.TryAddSingleton(typeof(IApiClientEventSender<>), typeof(MqttApiClientEventSender<>)); + services.TryAddSingleton(typeof(IApiClientEventReceiver<>), typeof(MqttApiClientEventReceiver<>)); + + return services; + } +} \ No newline at end of file diff --git a/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt/Factories/MqttClientFactory.cs b/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt/Factories/MqttClientFactory.cs new file mode 100644 index 0000000..42a7d5f --- /dev/null +++ b/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt/Factories/MqttClientFactory.cs @@ -0,0 +1,303 @@ +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; +using MQTTnet; +using MQTTnet.Formatter; +using RedShirt.Example.Api.ClientEvents.Library.Core.Exceptions; +using RedShirt.Example.Api.ClientEvents.Library.Mqtt.Models; +using RedShirt.Example.Api.ClientEvents.Library.Mqtt.Services; +using RedShirt.Example.Api.Common.SecretManagers.Core.Services; +using System.Net.Sockets; + +namespace RedShirt.Example.Api.ClientEvents.Library.Mqtt.Factories; + +internal interface IMqttClientFactory +{ + Task CreateConnectedClientAsync(CancellationToken cancellationToken); +} + +internal sealed class ApiMqttClientFactory( + IOptions configuration, + ISecretManagerService secretManagerService, + IMqttBrokerUrlResolver mqttBrokerUrlResolver, + ILogger logger) : IMqttClientFactory +{ + private const string DefaultClientIdPrefix = "RedShirt.Example.Api"; + private const string DefaultWebSocketPath = "/mqtt"; + private const string MqttWebSocketSubProtocol = "mqtt"; + private const int DefaultMqttTcpPort = 1883; + + private static readonly string ServerHostname = + string.IsNullOrWhiteSpace(Environment.MachineName) ? "unknown" : Environment.MachineName; + + /// + /// Try to resolve broker URI and then validate. + /// + /// + /// Farms out the actual resolution to , the main thing that makes this + /// method special is centralized validation. + /// + /// + /// + /// + /// + private async Task ResolveBrokerTargetAsync(ConfigurationModel config, + CancellationToken cancellationToken) + { + var brokerTarget = await ResolveBrokerTargetInnerAsync(config, cancellationToken); + if (string.IsNullOrWhiteSpace(brokerTarget.BrokerUrl)) + { + throw new ApiClientEventsException("ClientEvents MQTT broker URL is not configured.") + { + CouldBeTransient = false, + IsHandled = false + }; + } + + return brokerTarget; + } + + /// + /// Resolves the broker connection target from configuration or an external resolver. + /// + /// + /// When is false, returns + /// as-is. When true, delegates to + /// (for example the AWS IoT implementation in + /// RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws). + /// + /// + /// + /// + /// + private async Task ResolveBrokerTargetInnerAsync(ConfigurationModel config, + CancellationToken cancellationToken) + { + if (!config.ResolveBrokerAddressExternally) + { + return new MqttBrokerTarget + { + BrokerUrl = config.BrokerUrl + }; + } + + return await mqttBrokerUrlResolver.ResolveBrokerUrlAsync(cancellationToken); + } + + /// + /// Builds a fully configured for connecting to the MQTT broker. + /// + /// + /// MQTTnet does not connect from a single opaque URL string. Each transport must be configured explicitly + /// ( for native MQTT, + /// + /// for MQTT-over-WebSocket). Deployment environments also use different schemes for the same broker + /// (for example ws:// against MiniStack locally and mqtts:// in production), so broker URLs + /// are normalized into the builder call the transport requires. + /// MQTTnet WebSocket transport expects a full ws:// or wss:// URI (including the + /// /mqtt path and mqtt subprotocol). MiniStack (and AWS IoT) route upgrades by the IoT + /// endpoint hostname returned from DescribeEndpoint, not a bare gateway host name. + /// MiniStack and AWS IoT data-plane brokers speak MQTT 3.1.1 only. + /// + private async Task GetOptionsBuilderAsync(ConfigurationModel config, + CancellationToken cancellationToken) + { + var brokerTarget = await ResolveBrokerTargetAsync(config, cancellationToken); + + var optionsBuilder = new MqttClientOptionsBuilder() + .WithClientId($"{config.EffectiveClientIdPrefix}/{ServerHostname}/{Guid.NewGuid():N}") + .WithCleanSession(); + + if (!string.IsNullOrWhiteSpace(config.ProtocolVersion)) + { + if (Enum.TryParse(config.ProtocolVersion, out var protocolVersion)) + { + optionsBuilder.WithProtocolVersion(protocolVersion); + } + else + { + logger.LogWarning("Failed to parse MQTT protocol version {ProtocolVersion} to {EnumName}", + config.ProtocolVersion, typeof(MqttProtocolVersion).FullName); + } + } + + if (!Uri.TryCreate(brokerTarget.BrokerUrl!, UriKind.Absolute, out var uri)) + { + optionsBuilder.WithTcpServer(brokerTarget.BrokerUrl!); + } + else if (uri.Scheme.Equals("ws", StringComparison.OrdinalIgnoreCase) + || uri.Scheme.Equals("wss", StringComparison.OrdinalIgnoreCase)) + { + var webSocketUri = BuildWebSocketUri(uri); + optionsBuilder.WithWebSocketServer(webSocketBuilder => + { + webSocketBuilder + .WithUri(webSocketUri) + .WithSubProtocols([MqttWebSocketSubProtocol]); + + if (!string.IsNullOrWhiteSpace(brokerTarget.WebSocketHostHeader)) + { + webSocketBuilder.WithRequestHeaders(new Dictionary + { + ["Host"] = brokerTarget.WebSocketHostHeader + }); + } + }); + } + else if (uri.Scheme.Equals("tcp", StringComparison.OrdinalIgnoreCase) + || uri.Scheme.Equals("mqtt", StringComparison.OrdinalIgnoreCase) + || uri.Scheme.Equals("mqtts", StringComparison.OrdinalIgnoreCase)) + { + optionsBuilder.WithTcpServer(uri.Host, uri.IsDefaultPort ? DefaultMqttTcpPort : uri.Port); + } + else + { + throw new ApiClientEventsException($"Unsupported MQTT broker URL scheme '{uri.Scheme}'.") + { + CouldBeTransient = false, + IsHandled = false + }; + } + + if (await TryResolveCredentialsAsync(config, cancellationToken) is { } credentials) + { + optionsBuilder.WithCredentials(credentials.Username, credentials.Password); + } + + return new MqttClientOptionsBuildResult + { + OptionsBuilder = optionsBuilder, + BrokerUrl = brokerTarget.BrokerUrl! + }; + } + + private static string BuildWebSocketUri(Uri uri) + { + var path = string.IsNullOrEmpty(uri.AbsolutePath) || uri.AbsolutePath == "/" + ? DefaultWebSocketPath + : uri.AbsolutePath; + + return new UriBuilder(uri.Scheme, uri.Host, uri.Port, path).Uri.ToString(); + } + + private async Task TryResolveCredentialsAsync(ConfigurationModel config, + CancellationToken cancellationToken) + { + if (string.IsNullOrWhiteSpace(config.UsernameSecretPath) + || string.IsNullOrWhiteSpace(config.PasswordSecretPath)) + { + return null; + } + + var secrets = await secretManagerService.GetSecretsAsync( + [config.UsernameSecretPath, config.PasswordSecretPath], + cancellationToken); + + if (!secrets.TryGetValue(config.UsernameSecretPath, out var username) + || !secrets.TryGetValue(config.PasswordSecretPath, out var password)) + { + throw new ApiClientEventsException("MQTT username or password secret could not be resolved.") + { + CouldBeTransient = false, + IsHandled = false + }; + } + + return new MqttBrokerCredentials + { + Username = username, + Password = password + }; + } + + public async Task CreateConnectedClientAsync(CancellationToken cancellationToken) + { + var optionsBuild = await GetOptionsBuilderAsync(configuration.Value, cancellationToken); + + var mqttFactory = new MqttClientFactory(); + var client = mqttFactory.CreateMqttClient(); + + try + { + var connectResult = await client.ConnectAsync(optionsBuild.OptionsBuilder.Build(), cancellationToken); + if (connectResult.ResultCode != MqttClientConnectResultCode.Success) + { + throw new ApiClientEventsException( + $"MQTT broker connection failed with result code {connectResult.ResultCode}.") + { + CouldBeTransient = true, + IsHandled = false + }; + } + } + catch (ApiClientEventsException) + { + throw; + } + catch (Exception exception) + { + logger.LogError(exception, "Failed to connect to MQTT broker at {BrokerUrl}", optionsBuild.BrokerUrl); + throw new ApiClientEventsException(exception) + { + CouldBeTransient = true, + IsHandled = false + }; + } + + return client; + } + + private sealed class MqttClientOptionsBuildResult + { + /// + /// Configured options object builder. + /// + public required MqttClientOptionsBuilder OptionsBuilder { get; init; } + + /// + /// Also return BrokerUrl for logging purposes. + /// + public required string BrokerUrl { get; init; } + } + + private sealed class MqttBrokerCredentials + { + public required string Username { get; init; } + + public required string Password { get; init; } + } + + internal sealed class ConfigurationModel + { + /// + /// Prefix for MQTT client ids ({prefix}/{hostname}/{guid}). When null or whitespace, + /// is used. + /// + public string? ClientIdPrefix { get; init; } + + public string? BrokerUrl { get; init; } + + /// + /// When true, resolves the broker address through instead of using + /// verbatim. + /// + /// + /// Use this when the broker URL cannot be known up front or when the registered resolver supplies + /// transport-specific connection metadata (for example a WebSocket Host header). When false, + /// must be set to a complete broker address. + /// + public bool ResolveBrokerAddressExternally { get; init; } + + /// + /// MQTT protocol version to use when connecting. Expected to resolve to a + /// value (for example V311 or V500). + /// + public string? ProtocolVersion { get; init; } + + public string? UsernameSecretPath { get; init; } + + public string? PasswordSecretPath { get; init; } + + public string EffectiveClientIdPrefix => + string.IsNullOrWhiteSpace(ClientIdPrefix) ? DefaultClientIdPrefix : ClientIdPrefix; + } +} \ No newline at end of file diff --git a/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt/Globals.cs b/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt/Globals.cs new file mode 100644 index 0000000..4682326 --- /dev/null +++ b/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt/Globals.cs @@ -0,0 +1,6 @@ +using System.Runtime.CompilerServices; + +[assembly: InternalsVisibleTo("DynamicProxyGenAssembly2")] +[assembly: InternalsVisibleTo("RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws")] +[assembly: InternalsVisibleTo("RedShirt.Example.Api.ClientEvents.Library.Mqtt.UnitTests")] +[assembly: InternalsVisibleTo("RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws.UnitTests")] \ No newline at end of file diff --git a/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt/Models/MqttBrokerTarget.cs b/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt/Models/MqttBrokerTarget.cs new file mode 100644 index 0000000..6425524 --- /dev/null +++ b/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt/Models/MqttBrokerTarget.cs @@ -0,0 +1,8 @@ +namespace RedShirt.Example.Api.ClientEvents.Library.Mqtt.Models; + +internal sealed class MqttBrokerTarget +{ + public string? BrokerUrl { get; init; } + + public string? WebSocketHostHeader { get; init; } +} \ No newline at end of file diff --git a/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt/Models/MqttClientEventsExceptionArbiterReport.cs b/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt/Models/MqttClientEventsExceptionArbiterReport.cs new file mode 100644 index 0000000..607d03c --- /dev/null +++ b/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt/Models/MqttClientEventsExceptionArbiterReport.cs @@ -0,0 +1,10 @@ +namespace RedShirt.Example.Api.ClientEvents.Library.Mqtt.Models; + +internal sealed class MqttClientEventsExceptionArbiterReport +{ + public required bool AlreadyHandled { get; init; } + + public required bool IsExpected { get; init; } + + public required bool CouldBeTransient { get; init; } +} \ No newline at end of file diff --git a/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt/RedShirt.Example.Api.ClientEvents.Library.Mqtt.csproj b/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt/RedShirt.Example.Api.ClientEvents.Library.Mqtt.csproj new file mode 100644 index 0000000..1409a56 --- /dev/null +++ b/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt/RedShirt.Example.Api.ClientEvents.Library.Mqtt.csproj @@ -0,0 +1,22 @@ + + + + net10.0 + enable + enable + + + + + + + + + + + + + + + + diff --git a/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt/Services/IMqttBrokerUrlResolver.cs b/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt/Services/IMqttBrokerUrlResolver.cs new file mode 100644 index 0000000..4bb7c88 --- /dev/null +++ b/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt/Services/IMqttBrokerUrlResolver.cs @@ -0,0 +1,21 @@ +using RedShirt.Example.Api.ClientEvents.Library.Mqtt.Models; + +namespace RedShirt.Example.Api.ClientEvents.Library.Mqtt.Services; + +/// +/// Resolves an MQTT broker connection target when a configured BrokerUrl is not sufficient on its own. +/// +/// +/// Most environments can connect using a static broker URL from configuration. This resolver exists for the special +/// case where the broker address must be discovered at runtime — for example via AWS IoT +/// DescribeEndpoint in RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws. +/// +internal interface IMqttBrokerUrlResolver +{ + /// + /// Resolves the broker URL and any transport-specific connection hints needed to reach it. + /// + /// Cancellation token. + /// A broker target suitable for passing to the MQTT client options builder. + Task ResolveBrokerUrlAsync(CancellationToken cancellationToken); +} \ No newline at end of file diff --git a/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt/Services/MqttApiClientEventReceiver.cs b/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt/Services/MqttApiClientEventReceiver.cs new file mode 100644 index 0000000..7519fff --- /dev/null +++ b/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt/Services/MqttApiClientEventReceiver.cs @@ -0,0 +1,133 @@ +using Microsoft.Extensions.Logging; +using MQTTnet; +using MQTTnet.Protocol; +using RedShirt.Example.Api.ClientEvents.Library.Core.Exceptions; +using RedShirt.Example.Api.ClientEvents.Library.Core.Models; +using RedShirt.Example.Api.ClientEvents.Library.Core.Services; +using RedShirt.Example.Api.ClientEvents.Library.Mqtt.Factories; +using RedShirt.Example.Api.ClientEvents.Library.Mqtt.Services.Resilience; +using System.Runtime.CompilerServices; +using System.Text.Json; +using System.Threading.Channels; + +namespace RedShirt.Example.Api.ClientEvents.Library.Mqtt.Services; + +internal sealed class MqttApiClientEventReceiver( + IMqttClientFactory mqttClientFactory, + IMqttClientEventsRetryWrapperService retryWrapperService, + ILogger> logger) : IApiClientEventReceiver +{ + private static async Task DisconnectSafelyAsync(IMqttClient client, CancellationToken cancellationToken) + { + if (!client.IsConnected) + { + return; + } + + try + { + await client.DisconnectAsync(cancellationToken: cancellationToken); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + throw; + } + catch + { + // Best-effort disconnect during cleanup. + } + } + + public async IAsyncEnumerable> ReceiveAsync( + IReadOnlyList? topics = null, + [EnumeratorCancellation] + CancellationToken cancellationToken = default) + { + IMqttClient? client = null; + Channel>? channel = null; + + try + { + client = await retryWrapperService.RunAsync( + mqttClientFactory.CreateConnectedClientAsync, + cancellationToken); + + channel = Channel.CreateUnbounded>( + new UnboundedChannelOptions {SingleReader = true, SingleWriter = false}); + + // MQTT pushes messages on this callback thread; ReceiveAsync exposes an IAsyncEnumerable to callers. + // The channel bridges those models: WriteAsync enqueues from the handler, and then the ReadAllAsync invocation + // below dequeues messages into the async iterator that SSE (and any other consumers) pull from. + client.ApplicationMessageReceivedAsync += async args => + { + try + { + var payloadText = args.ApplicationMessage.ConvertPayloadToString(); + var payload = JsonSerializer.Deserialize(payloadText); + if (payload is null) + { + return; + } + + // Must forward into the channel: the handler cannot yield return, and the reader is blocked + // on ReadAllAsync until a message arrives from the broker. WriteAsync is in-process only + // (an unbounded Channel queue). It does not send MQTT traffic. The broker already delivered + // the payload before this callback ran. + await channel.Writer.WriteAsync(new ApiClientEventReceived + { + Topic = args.ApplicationMessage.Topic, + Payload = payload + }, cancellationToken); + } + catch (Exception exception) + { + logger.LogWarning(exception, "Failed to deserialize MQTT client event payload"); + } + }; + + var subscribeBuilder = new MqttClientSubscribeOptionsBuilder(); + if (topics is {Count: > 0}) + { + foreach (var topic in topics.Where(topic => !string.IsNullOrWhiteSpace(topic))) + { + subscribeBuilder.WithTopicFilter(topic, MqttQualityOfServiceLevel.AtLeastOnce); + } + } + else + { + subscribeBuilder.WithTopicFilter("#", MqttQualityOfServiceLevel.AtLeastOnce); + } + + await retryWrapperService.RunAsync(async token => + { + var subscribeResult = await client.SubscribeAsync(subscribeBuilder.Build(), token); + if (subscribeResult.Items.Any(item => + item.ResultCode is MqttClientSubscribeResultCode.UnspecifiedError + or MqttClientSubscribeResultCode.NotAuthorized + or MqttClientSubscribeResultCode.TopicFilterInvalid)) + { + throw new ApiClientEventsException("MQTT subscription was not fully granted.") + { + CouldBeTransient = true, + IsHandled = false + }; + } + }, cancellationToken); + + await foreach (var received in channel.Reader.ReadAllAsync(cancellationToken)) + { + yield return received; + } + } + finally + { + channel?.Writer.TryComplete(); + + if (client is not null) + { + await DisconnectSafelyAsync(client, cancellationToken); + client.Dispose(); + } + } + } +} \ No newline at end of file diff --git a/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt/Services/MqttApiClientEventSender.cs b/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt/Services/MqttApiClientEventSender.cs new file mode 100644 index 0000000..1d5fdf7 --- /dev/null +++ b/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt/Services/MqttApiClientEventSender.cs @@ -0,0 +1,98 @@ +using Microsoft.Extensions.Logging; +using MQTTnet; +using MQTTnet.Protocol; +using RedShirt.Example.Api.ClientEvents.Library.Core.Exceptions; +using RedShirt.Example.Api.ClientEvents.Library.Core.Models; +using RedShirt.Example.Api.ClientEvents.Library.Core.Services; +using RedShirt.Example.Api.ClientEvents.Library.Mqtt.Factories; +using RedShirt.Example.Api.ClientEvents.Library.Mqtt.Services.Resilience; +using System.Text.Json; + +namespace RedShirt.Example.Api.ClientEvents.Library.Mqtt.Services; + +internal sealed class MqttApiClientEventSender( + IMqttClientFactory mqttClientFactory, + IMqttClientEventsRetryWrapperService retryWrapperService, + ILogger> logger) : IApiClientEventSender +{ + private static async Task DisconnectSafelyAsync(IMqttClient client, CancellationToken cancellationToken) + { + if (!client.IsConnected) + { + return; + } + + try + { + await client.DisconnectAsync(cancellationToken: cancellationToken); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + throw; + } + catch + { + // Best-effort disconnect during cleanup. + } + } + + public Task SendAsync(ApiClientEventSendRequest request, CancellationToken cancellationToken) + { + if (string.IsNullOrWhiteSpace(request.Topic)) + { + throw new ApiClientEventsException("An MQTT-shaped topic is required to publish a client event.") + { + CouldBeTransient = false, + IsHandled = false + }; + } + + return retryWrapperService.RunAsync(async token => + { + var payload = JsonSerializer.Serialize(request.Payload); + IMqttClient? client = null; + + try + { + client = await mqttClientFactory.CreateConnectedClientAsync(token); + var publishResult = await client.PublishAsync(new MqttApplicationMessageBuilder() + .WithTopic(request.Topic) + .WithPayload(payload) + .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce) + .Build(), token); + + if (publishResult.ReasonCode is MqttClientPublishReasonCode.UnspecifiedError + or MqttClientPublishReasonCode.ImplementationSpecificError) + { + throw new ApiClientEventsException( + $"MQTT publish failed with reason code {publishResult.ReasonCode}.") + { + CouldBeTransient = true, + IsHandled = false + }; + } + } + catch (ApiClientEventsException) + { + throw; + } + catch (Exception exception) + { + logger.LogError(exception, "Failed to publish MQTT client event to topic {Topic}", request.Topic); + throw new ApiClientEventsException(exception) + { + CouldBeTransient = true, + IsHandled = false + }; + } + finally + { + if (client is not null) + { + await DisconnectSafelyAsync(client, token); + client.Dispose(); + } + } + }, cancellationToken); + } +} \ No newline at end of file diff --git a/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt/Services/Resilience/MqttClientEventsExceptionArbiterService.cs b/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt/Services/Resilience/MqttClientEventsExceptionArbiterService.cs new file mode 100644 index 0000000..169338c --- /dev/null +++ b/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt/Services/Resilience/MqttClientEventsExceptionArbiterService.cs @@ -0,0 +1,82 @@ +using RedShirt.Example.Api.ClientEvents.Library.Core.Exceptions; +using RedShirt.Example.Api.ClientEvents.Library.Mqtt.Models; +using RedShirt.Example.Api.Common.SecretManagers.Core.Exceptions; +using System.Net.Sockets; +using System.Text.Json; + +namespace RedShirt.Example.Api.ClientEvents.Library.Mqtt.Services.Resilience; + +/// +/// Classifies MQTT client-events exceptions for retry decisions. +/// +internal interface IMqttClientEventsExceptionArbiterService +{ + MqttClientEventsExceptionArbiterReport GetReport(Exception exception); +} + +/// +/// MQTT-oriented exception arbiter modelled after the connector / distributed arbiters: +/// known infrastructure failures may be transient; caller cancel and configuration errors are not. +/// +internal sealed class MqttClientEventsExceptionArbiterService : IMqttClientEventsExceptionArbiterService +{ + private static MqttClientEventsExceptionArbiterReport Fresh(bool isExpected, bool couldBeTransient) + { + return new MqttClientEventsExceptionArbiterReport + { + AlreadyHandled = false, + IsExpected = isExpected, + CouldBeTransient = couldBeTransient + }; + } + + private static MqttClientEventsExceptionArbiterReport Handled(bool isExpected, bool couldBeTransient) + { + return new MqttClientEventsExceptionArbiterReport + { + AlreadyHandled = true, + IsExpected = isExpected, + CouldBeTransient = couldBeTransient + }; + } + + private static MqttClientEventsExceptionArbiterReport ClassifyApiClientEventsException( + ApiClientEventsException exception) + { + if (exception.IsHandled) + { + return Handled(true, exception.CouldBeTransient); + } + + if (!exception.CouldBeTransient) + { + return Fresh(true, false); + } + + return Fresh(true, true); + } + + public MqttClientEventsExceptionArbiterReport GetReport(Exception exception) + { + ArgumentNullException.ThrowIfNull(exception); + + while (exception is AggregateException {InnerExceptions.Count: 1, InnerException: not null} aggregate) + { + exception = aggregate.InnerException!; + } + + return exception switch + { + ApiClientEventsException clientEvents => ClassifyApiClientEventsException(clientEvents), + ApiSecretManagerException secretManager => + Handled(true, secretManager is {IsHandled: false, CouldBeTransient: true}), + SocketException + or TimeoutException => Fresh(true, true), + JsonException => Fresh(true, false), + TaskCanceledException => Fresh(true, true), + OperationCanceledException => Fresh(true, false), + ArgumentException => Fresh(true, false), + _ => Fresh(false, false) + }; + } +} \ No newline at end of file diff --git a/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt/Services/Resilience/MqttClientEventsRetryWrapperService.cs b/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt/Services/Resilience/MqttClientEventsRetryWrapperService.cs new file mode 100644 index 0000000..bcb8cf0 --- /dev/null +++ b/src/RedShirt.Example.Api.ClientEvents.Library.Mqtt/Services/Resilience/MqttClientEventsRetryWrapperService.cs @@ -0,0 +1,165 @@ +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; +using Polly; +using Polly.Retry; +using RedShirt.Example.Api.ClientEvents.Library.Core.Exceptions; +using RedShirt.Example.Api.Common.Services.Utility; + +namespace RedShirt.Example.Api.ClientEvents.Library.Mqtt.Services.Resilience; + +/// +/// Retries MQTT client-events operations that fail with expected transient exceptions, +/// then surfaces remaining failures as . +/// +internal interface IMqttClientEventsRetryWrapperService +{ + /// + /// Executes with retry for expected transient MQTT client-events failures. + /// + Task RunAsync(Func> func, CancellationToken cancellationToken = default); + + /// + /// Executes with retry for expected transient MQTT client-events failures. + /// + Task RunAsync(Func func, CancellationToken cancellationToken = default); +} + +/// +/// Polly v8-based retry wrapper for MQTT client-events calls. +/// Retries when reports an expected transient failure, +/// using exponential backoff via . +/// +internal sealed class MqttClientEventsRetryWrapperService( + IMqttClientEventsExceptionArbiterService exceptionArbiterService, + ILogger logger, + ISleepService sleepService, + IOptions options) + : IMqttClientEventsRetryWrapperService +{ + private const int DefaultRetryCount = 3; + + private ResiliencePipeline? _retryPipeline; + + private ResiliencePipeline GetRetryPipeline() + { + return _retryPipeline ??= new ResiliencePipelineBuilder() + .AddRetry(new RetryStrategyOptions + { + MaxRetryAttempts = options.Value.EffectiveRetryCount, + ShouldHandle = args => + { + // ReSharper disable once DuplicatedSequentialIfBodies + if (args.Outcome.Exception is not { } exception) + { + return PredicateResult.False(); + } + + if (args.Context.CancellationToken.IsCancellationRequested) + { + return PredicateResult.False(); + } + + var report = exceptionArbiterService.GetReport(exception); + return report is {IsExpected: true, CouldBeTransient: true} + ? PredicateResult.True() + : PredicateResult.False(); + }, + DelayGenerator = static _ => new ValueTask(TimeSpan.Zero), + OnRetry = async args => + { + logger.LogWarning(args.Outcome.Exception, + "Retrying MQTT client-events operation after attempt {AttemptNumber}", + args.AttemptNumber); + await sleepService.DelayAsync(TimeSpan.FromSeconds(Math.Pow(2, args.AttemptNumber)), + args.Context.CancellationToken); + } + }) + .Build(); + } + + private bool TryGetWrappedException(Exception exception, out Exception? wrappedException) + { + wrappedException = null; + + var report = exceptionArbiterService.GetReport(exception); + + if (report.AlreadyHandled && exception is ApiClientEventsException) + { + return false; + } + + if (!report.IsExpected) + { + return false; + } + + wrappedException = new ApiClientEventsException(exception) + { + CouldBeTransient = report.CouldBeTransient, + IsHandled = true + }; + + return true; + } + + public async Task RunAsync(Func> func, + CancellationToken cancellationToken = default) + { + try + { + return await GetRetryPipeline().ExecuteAsync( + async token => await func(token), + cancellationToken); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + throw; + } + catch (Exception exception) + { + if (TryGetWrappedException(exception, out var wrappedException) && wrappedException is not null) + { + throw wrappedException; + } + + throw; + } + } + + public async Task RunAsync(Func func, CancellationToken cancellationToken = default) + { + try + { + await GetRetryPipeline().ExecuteAsync( + async token => await func(token), + cancellationToken); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + throw; + } + catch (Exception exception) + { + if (TryGetWrappedException(exception, out var wrappedException) && wrappedException is not null) + { + throw wrappedException; + } + + throw; + } + } + + internal sealed class ConfigurationModel + { + /// + /// Maximum number of retry attempts for expected transient MQTT client-events failures. + /// When null, is used. + /// + public required int? RetryCount { get; init; } + + /// + /// Effective retry attempt count (floored at zero). + /// + public int EffectiveRetryCount => Math.Max(0, RetryCount ?? DefaultRetryCount); + } +} \ No newline at end of file diff --git a/src/RedShirt.Example.Api.Core/RedShirt.Example.Api.Core.csproj b/src/RedShirt.Example.Api.Core/RedShirt.Example.Api.Core.csproj index d41754d..0719031 100644 --- a/src/RedShirt.Example.Api.Core/RedShirt.Example.Api.Core.csproj +++ b/src/RedShirt.Example.Api.Core/RedShirt.Example.Api.Core.csproj @@ -24,6 +24,7 @@ + diff --git a/src/RedShirt.Example.Api.Core/UseCases/Messages/Commands/Send/SendExampleMessageCommand.cs b/src/RedShirt.Example.Api.Core/UseCases/Messages/Commands/Send/SendExampleMessageCommand.cs new file mode 100644 index 0000000..e71c9ae --- /dev/null +++ b/src/RedShirt.Example.Api.Core/UseCases/Messages/Commands/Send/SendExampleMessageCommand.cs @@ -0,0 +1,3 @@ +namespace RedShirt.Example.Api.Core.UseCases.Messages.Commands.Send; + +public sealed record SendExampleMessageCommand(string UserId, string Message); \ No newline at end of file diff --git a/src/RedShirt.Example.Api.Core/UseCases/Messages/Commands/Send/SendExampleMessageCommandHandler.cs b/src/RedShirt.Example.Api.Core/UseCases/Messages/Commands/Send/SendExampleMessageCommandHandler.cs new file mode 100644 index 0000000..4307645 --- /dev/null +++ b/src/RedShirt.Example.Api.Core/UseCases/Messages/Commands/Send/SendExampleMessageCommandHandler.cs @@ -0,0 +1,21 @@ +using RedShirt.Example.Api.ClientEvents.Domains.Example.Models; +using RedShirt.Example.Api.ClientEvents.Domains.Example.Services; +using RedShirt.Example.Api.Core.Cqrs; + +namespace RedShirt.Example.Api.Core.UseCases.Messages.Commands.Send; + +public interface ISendExampleMessageCommandHandler : ICqrsHandler; + +internal sealed class SendExampleMessageCommandHandler( + IExampleMessageSendService exampleMessageSendService, + ICoreRequestValidator coreRequestValidator) : ISendExampleMessageCommandHandler +{ + public async Task Handle(SendExampleMessageCommand command, CancellationToken cancellationToken = default) + { + await coreRequestValidator.ValidateAsync(command, cancellationToken); + await exampleMessageSendService.SendToUserAsync(command.UserId, new ExampleMessageModel + { + Message = command.Message + }, cancellationToken); + } +} \ No newline at end of file diff --git a/src/RedShirt.Example.Api.Core/UseCases/Messages/Commands/Send/SendExampleMessageCommandValidator.cs b/src/RedShirt.Example.Api.Core/UseCases/Messages/Commands/Send/SendExampleMessageCommandValidator.cs new file mode 100644 index 0000000..6f71cc0 --- /dev/null +++ b/src/RedShirt.Example.Api.Core/UseCases/Messages/Commands/Send/SendExampleMessageCommandValidator.cs @@ -0,0 +1,17 @@ +using FluentValidation; + +namespace RedShirt.Example.Api.Core.UseCases.Messages.Commands.Send; + +public sealed class SendExampleMessageCommandValidator : AbstractValidator +{ + public SendExampleMessageCommandValidator() + { + RuleFor(command => command.UserId) + .Must(userId => !string.IsNullOrWhiteSpace(userId)) + .WithMessage("UserId is required"); + + RuleFor(command => command.Message) + .Must(message => !string.IsNullOrWhiteSpace(message)) + .WithMessage("Message is required"); + } +} \ No newline at end of file diff --git a/src/RedShirt.Example.Api.Core/UseCases/Messages/Queries/Stream/StreamExampleMessagesQuery.cs b/src/RedShirt.Example.Api.Core/UseCases/Messages/Queries/Stream/StreamExampleMessagesQuery.cs new file mode 100644 index 0000000..046390c --- /dev/null +++ b/src/RedShirt.Example.Api.Core/UseCases/Messages/Queries/Stream/StreamExampleMessagesQuery.cs @@ -0,0 +1,3 @@ +namespace RedShirt.Example.Api.Core.UseCases.Messages.Queries.Stream; + +public sealed record StreamExampleMessagesQuery(string UserId); \ No newline at end of file diff --git a/src/RedShirt.Example.Api.Core/UseCases/Messages/Queries/Stream/StreamExampleMessagesQueryHandler.cs b/src/RedShirt.Example.Api.Core/UseCases/Messages/Queries/Stream/StreamExampleMessagesQueryHandler.cs new file mode 100644 index 0000000..50d641e --- /dev/null +++ b/src/RedShirt.Example.Api.Core/UseCases/Messages/Queries/Stream/StreamExampleMessagesQueryHandler.cs @@ -0,0 +1,21 @@ +using RedShirt.Example.Api.ClientEvents.Domains.Example.Models; +using RedShirt.Example.Api.ClientEvents.Domains.Example.Services; +using RedShirt.Example.Api.Core.Cqrs; + +namespace RedShirt.Example.Api.Core.UseCases.Messages.Queries.Stream; + +public interface + IStreamExampleMessagesQueryHandler : ICqrsHandler>; + +internal sealed class StreamExampleMessagesQueryHandler( + IExampleMessageReceiveService exampleMessageReceiveService, + ICoreRequestValidator coreRequestValidator) : IStreamExampleMessagesQueryHandler +{ + public async Task> Handle(StreamExampleMessagesQuery query, + CancellationToken cancellationToken = default) + { + await coreRequestValidator.ValidateAsync(query, cancellationToken); + return exampleMessageReceiveService.ReceiveForUserAsync(query.UserId, cancellationToken); + } +} \ No newline at end of file diff --git a/src/RedShirt.Example.Api.Core/UseCases/Messages/Queries/Stream/StreamExampleMessagesQueryValidator.cs b/src/RedShirt.Example.Api.Core/UseCases/Messages/Queries/Stream/StreamExampleMessagesQueryValidator.cs new file mode 100644 index 0000000..1cfe655 --- /dev/null +++ b/src/RedShirt.Example.Api.Core/UseCases/Messages/Queries/Stream/StreamExampleMessagesQueryValidator.cs @@ -0,0 +1,13 @@ +using FluentValidation; + +namespace RedShirt.Example.Api.Core.UseCases.Messages.Queries.Stream; + +public sealed class StreamExampleMessagesQueryValidator : AbstractValidator +{ + public StreamExampleMessagesQueryValidator() + { + RuleFor(query => query.UserId) + .Must(userId => !string.IsNullOrWhiteSpace(userId)) + .WithMessage("UserId is required"); + } +} \ No newline at end of file diff --git a/src/RedShirt.Example.Api/Controllers/MessageController.cs b/src/RedShirt.Example.Api/Controllers/MessageController.cs new file mode 100644 index 0000000..2aee967 --- /dev/null +++ b/src/RedShirt.Example.Api/Controllers/MessageController.cs @@ -0,0 +1,81 @@ +using Microsoft.AspNetCore.Mvc; +using Microsoft.AspNetCore.RateLimiting; +using RedShirt.Example.Api.Attributes.Authorization; +using RedShirt.Example.Api.Authorization.Extensions; +using RedShirt.Example.Api.ClientEvents.Domains.Example.Models; +using RedShirt.Example.Api.Constants; +using RedShirt.Example.Api.Core.UseCases.Messages.Commands.Send; +using RedShirt.Example.Api.Core.UseCases.Messages.Queries.Stream; +using RedShirt.Example.Api.Models.Messages; +using System.Runtime.CompilerServices; + +namespace RedShirt.Example.Api.Controllers; + +[ApiController] +[EnableRateLimiting(RateLimitingConstants.PolicyHeaderDefault)] +[Route("messages")] +public class MessageController : ControllerBase +{ + private static async IAsyncEnumerable ToServerSentEvents( + IAsyncEnumerable messages, + [EnumeratorCancellation] + CancellationToken cancellationToken) + { + await foreach (var message in messages.WithCancellation(cancellationToken)) + { + yield return message.Message; + } + } + + /// Event stream + /// + /// This GET /messages/event-stream endpoint is not directly supported by OpenAPI. + /// The generated NSwag MessageClient.GetEventStreamAsync method is intended for connectivity checks only. + /// Use MessagesEventStreamListener in RedShirt.Example.Api.Interop project to receive message events. + /// + [HttpGet("event-stream")] + [ApproveReadOnly] + [Produces("text/event-stream")] + [ProducesResponseType(StatusCodes.Status200OK)] + [ProducesResponseType(StatusCodes.Status400BadRequest)] + [ProducesResponseType(StatusCodes.Status502BadGateway)] + public async Task GetEventStream( + [FromServices] + IStreamExampleMessagesQueryHandler streamExampleMessagesQueryHandler, + CancellationToken cancellationToken) + { + if (!User.TryGetUserId(out var userId)) + { + return TypedResults.BadRequest("Authenticated user id is required."); + } + + var messageStream = + await streamExampleMessagesQueryHandler.Handle(new StreamExampleMessagesQuery(userId), cancellationToken); + + return TypedResults.ServerSentEvents( + ToServerSentEvents(messageStream, cancellationToken), + "message"); + } + + [HttpPost] + [ProducesResponseType(StatusCodes.Status202Accepted)] + [ProducesResponseType(StatusCodes.Status400BadRequest)] + [ProducesResponseType(StatusCodes.Status502BadGateway)] + public async Task Post( + [FromBody] + ExampleMessagePostRequest request, + [FromServices] + ISendExampleMessageCommandHandler sendExampleMessageCommandHandler, + CancellationToken cancellationToken) + { + if (!User.TryGetUserId(out var userId)) + { + return BadRequest("Authenticated user id is required."); + } + + await sendExampleMessageCommandHandler.Handle(new SendExampleMessageCommand(userId, request.Message), + cancellationToken); + + return Accepted(); + } +} \ No newline at end of file diff --git a/src/RedShirt.Example.Api/ExceptionHandlers/ApiExceptionHandler.cs b/src/RedShirt.Example.Api/ExceptionHandlers/ApiExceptionHandler.cs index 67d10e3..035b15e 100644 --- a/src/RedShirt.Example.Api/ExceptionHandlers/ApiExceptionHandler.cs +++ b/src/RedShirt.Example.Api/ExceptionHandlers/ApiExceptionHandler.cs @@ -1,5 +1,6 @@ using Microsoft.AspNetCore.Diagnostics; using Microsoft.AspNetCore.Mvc; +using RedShirt.Example.Api.ClientEvents.Library.Core.Exceptions; using RedShirt.Example.Api.Common.Exceptions.Responses; using RedShirt.Example.Api.Connectors.Bar.Core.Exceptions; using RedShirt.Example.Api.Connectors.Foo.Core.Exceptions; @@ -42,6 +43,7 @@ private static bool TryMapException(Exception exception, out int statusCode, out return true; case FooConnectorException: case BarConnectorException: + case ApiClientEventsException: statusCode = StatusCodes.Status502BadGateway; title = "Bad Gateway"; return true; diff --git a/src/RedShirt.Example.Api/Extensions/ServiceCollectionExtensions.cs b/src/RedShirt.Example.Api/Extensions/ServiceCollectionExtensions.cs index 98124b1..31388d5 100644 --- a/src/RedShirt.Example.Api/Extensions/ServiceCollectionExtensions.cs +++ b/src/RedShirt.Example.Api/Extensions/ServiceCollectionExtensions.cs @@ -1,3 +1,6 @@ +using RedShirt.Example.Api.ClientEvents.Domains.Example.Extensions; +using RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws.Extensions; +using RedShirt.Example.Api.ClientEvents.Library.Mqtt.Extensions; using RedShirt.Example.Api.Common.Aws.SsmSecretManager.Extensions; using RedShirt.Example.Api.Common.Distributed.Extensions; using RedShirt.Example.Api.Common.Extensions; @@ -58,6 +61,9 @@ internal static IServiceCollection ConfigureApiServices(this IServiceCollection .AddCustomers(configuration) // Customer store via Entity Framework Core (MariaDB / MySQL). .AddOrders() .AddProducts(configuration) - .AddUploads(configuration); + .AddUploads(configuration) + .AddMqttClientEvents(configuration) + .AddAwsMqttClientEvents(configuration) + .AddExampleClientEventsDomain(); } } \ No newline at end of file diff --git a/src/RedShirt.Example.Api/Models/Messages/ExampleMessagePostRequest.cs b/src/RedShirt.Example.Api/Models/Messages/ExampleMessagePostRequest.cs new file mode 100644 index 0000000..1b68397 --- /dev/null +++ b/src/RedShirt.Example.Api/Models/Messages/ExampleMessagePostRequest.cs @@ -0,0 +1,6 @@ +namespace RedShirt.Example.Api.Models.Messages; + +public sealed class ExampleMessagePostRequest +{ + public required string Message { get; init; } +} \ No newline at end of file diff --git a/src/RedShirt.Example.Api/RedShirt.Example.Api.csproj b/src/RedShirt.Example.Api/RedShirt.Example.Api.csproj index 7a37681..e696445 100644 --- a/src/RedShirt.Example.Api/RedShirt.Example.Api.csproj +++ b/src/RedShirt.Example.Api/RedShirt.Example.Api.csproj @@ -4,6 +4,8 @@ net10.0 enable enable + true + $(NoWarn);1591 @@ -33,6 +35,10 @@ + + + + diff --git a/swagger/RedShirt.Example.Api.Interop/EventStreamListener.cs b/swagger/RedShirt.Example.Api.Interop/EventStreamListener.cs new file mode 100644 index 0000000..f3ef014 --- /dev/null +++ b/swagger/RedShirt.Example.Api.Interop/EventStreamListener.cs @@ -0,0 +1,418 @@ +using System; +using System.Collections.Generic; +using System.IO; +using System.Net.Http; +using System.Net.Http.Headers; +using System.Text; +using System.Text.Json; +using System.Threading; +using System.Threading.Tasks; + +namespace RedShirt.Example.Api.Interop; + +public abstract class EventStreamListener +{ + private static readonly Lazy DefaultJsonSerializerSettings = + new(CreateJsonSerializerSettings, true); + + private readonly HttpClient _httpClient; + + protected EventStreamListener(string baseUrl, HttpClient httpClient, string eventEndpointPath) + { + if (eventEndpointPath == null) + { + throw new ArgumentNullException(nameof(eventEndpointPath)); + } + + BaseUrl = NormalizeBaseUrl(baseUrl); + _httpClient = httpClient ?? throw new ArgumentNullException(nameof(httpClient)); + EventEndpointPath = NormalizeEventEndpointPath(eventEndpointPath); + } + + // ReSharper disable once MemberCanBePrivate.Global + protected string BaseUrl { get; } + + // ReSharper disable once MemberCanBePrivate.Global + protected string EventEndpointPath { get; } + + protected virtual string DefaultEventName => "event"; + + protected virtual JsonSerializerOptions JsonSerializerSettings => DefaultJsonSerializerSettings.Value; + + protected virtual void PrepareRequest(HttpClient client, HttpRequestMessage request, string url) + { + } + + protected virtual void ProcessResponse(HttpClient client, HttpResponseMessage response) + { + } + + protected Task StreamEventsAsync( + Action onEvent, + CancellationToken cancellationToken = default) + { + if (onEvent == null) + { + throw new ArgumentNullException(nameof(onEvent)); + } + + return StreamEventsAsync( + (data, _) => + { + onEvent(data); + return Task.CompletedTask; + }, + cancellationToken); + } + + protected Task StreamEventsAsync( + Func onEvent, + CancellationToken cancellationToken = default) + { + if (onEvent == null) + { + throw new ArgumentNullException(nameof(onEvent)); + } + + return StreamEventsCoreAsync(onEvent, cancellationToken); + } + + protected abstract Task StreamEventsCoreAsync( + Func onEvent, + CancellationToken cancellationToken); + + protected async Task OpenEventStreamAsync( + Func onEvent, + string? expectedEventName, + CancellationToken cancellationToken) + { + var client = _httpClient; + var disposeClient = false; + + try + { + using var request = new HttpRequestMessage(HttpMethod.Get, BuildRequestUri()); + request.Headers.Accept.Add(MediaTypeWithQualityHeaderValue.Parse("text/event-stream")); + + PrepareRequest(client, request, request.RequestUri!.ToString()); + + using var response = await client + .SendAsync(request, HttpCompletionOption.ResponseHeadersRead, cancellationToken) + .ConfigureAwait(false); + + var headers = CopyResponseHeaders(response); + ProcessResponse(client, response); + + var status = (int) response.StatusCode; + + /* + * The below is organized as a series of if statements and not a switch statement + * for consistency with the generated NSwag client code. + */ + + // ReSharper disable once ConvertIfStatementToSwitchStatement + if (status == 200) + { + if (response.Content == null) + { + return; + } + + await using var stream = await ReadAsStreamAsync(response.Content, cancellationToken) + .ConfigureAwait(false); + await ReadServerSentEventsAsync( + stream, + onEvent, + expectedEventName, + DefaultEventName, + cancellationToken) + .ConfigureAwait(false); + return; + } + + if (status == 400) + { + var objectResponse = await ReadObjectResponseAsync(response, headers, cancellationToken) + .ConfigureAwait(false); + if (objectResponse.Object == null) + { + throw new SwaggerException( + "Response was null which was not expected.", + status, + objectResponse.Text, + headers, + null); + } + + throw new SwaggerException( + "A server side error occurred.", + status, + objectResponse.Text, + headers, + objectResponse.Object, + null); + } + + if (status == 502) + { + var responseText = response.Content == null + ? string.Empty + : await ReadAsStringAsync(response.Content, cancellationToken).ConfigureAwait(false); + throw new SwaggerException("A server side error occurred.", status, responseText, headers, null); + } + + var responseData = response.Content == null + ? null + : await ReadAsStringAsync(response.Content, cancellationToken).ConfigureAwait(false); + throw new SwaggerException( + "The HTTP status code of the response was not expected (" + status + ").", + status, + responseData, + headers, + null); + } + finally + { + if (disposeClient) + { + client.Dispose(); + } + } + } + + private Uri BuildRequestUri() + { + var urlBuilder = new StringBuilder(); + if (!string.IsNullOrEmpty(BaseUrl)) + { + urlBuilder.Append(BaseUrl); + } + + urlBuilder.Append(EventEndpointPath); + return new Uri(urlBuilder.ToString(), UriKind.RelativeOrAbsolute); + } + + private static string NormalizeBaseUrl(string baseUrl) + { + if (string.IsNullOrEmpty(baseUrl)) + { + return baseUrl; + } + + return baseUrl.EndsWith("/", StringComparison.Ordinal) ? baseUrl : baseUrl + "/"; + } + + private static string NormalizeEventEndpointPath(string eventEndpointPath) + { + return eventEndpointPath.TrimStart('/'); + } + + private static Dictionary> CopyResponseHeaders(HttpResponseMessage response) + { + var headers = new Dictionary>(); + foreach (var item in response.Headers) + { + headers[item.Key] = item.Value; + } + + // ReSharper disable once InvertIf + if (response.Content?.Headers != null) + { + foreach (var item in response.Content.Headers) + { + headers[item.Key] = item.Value; + } + } + + return headers; + } + + private static async Task ReadServerSentEventsAsync( + Stream stream, + Func onEvent, + string? expectedEventName, + string defaultEventName, + CancellationToken cancellationToken) + { + using var reader = new StreamReader(stream); + string? eventName = null; + var dataLines = new List(); + + while (!cancellationToken.IsCancellationRequested) + { + var line = await reader.ReadLineAsync().ConfigureAwait(false); + if (line == null) + { + break; + } + + if (line.Length == 0) + { + if (dataLines.Count > 0) + { + var resolvedEventName = eventName ?? defaultEventName; + if (expectedEventName == null || + string.Equals(resolvedEventName, expectedEventName, StringComparison.Ordinal)) + { + await onEvent(string.Join("\n", dataLines), cancellationToken).ConfigureAwait(false); + } + } + + eventName = null; + dataLines.Clear(); + continue; + } + + if (line.StartsWith(":", StringComparison.Ordinal)) + { + continue; + } + + if (line.StartsWith("event:", StringComparison.Ordinal)) + { + eventName = line.Substring("event:".Length).Trim(); + continue; + } + + if (line.StartsWith("data:", StringComparison.Ordinal)) + { + dataLines.Add(line.Substring("data:".Length).TrimStart()); + } + } + + cancellationToken.ThrowIfCancellationRequested(); + } + + private async Task> ReadObjectResponseAsync( + HttpResponseMessage response, + IReadOnlyDictionary> headers, + CancellationToken cancellationToken) + { + if (response.Content == null) + { + return new ObjectResponseResult(default!, string.Empty); + } + + var responseText = await ReadAsStringAsync(response.Content, cancellationToken).ConfigureAwait(false); + try + { + var typedBody = JsonSerializer.Deserialize(responseText, JsonSerializerSettings); + return new ObjectResponseResult(typedBody!, responseText); + } + catch (JsonException exception) + { + var message = "Could not deserialize the response body string as " + typeof(T).FullName + "."; + throw new SwaggerException(message, (int) response.StatusCode, responseText, headers, exception); + } + } + + private static JsonSerializerOptions CreateJsonSerializerSettings() + { + return new JsonSerializerOptions(); + } + +#if NET5_0_OR_GREATER + private static Task ReadAsStringAsync(HttpContent content, CancellationToken cancellationToken) => + content.ReadAsStringAsync(cancellationToken); + + private static Task ReadAsStreamAsync(HttpContent content, CancellationToken cancellationToken) => + content.ReadAsStreamAsync(cancellationToken); +#else + // ReSharper disable UnusedParameter.Local + private static Task ReadAsStringAsync(HttpContent content, CancellationToken cancellationToken) + { + return content.ReadAsStringAsync(); + } + + private static Task ReadAsStreamAsync(HttpContent content, CancellationToken cancellationToken) + { + return content.ReadAsStreamAsync(); + } + // ReSharper restore UnusedParameter.Local +#endif + + // ReSharper disable once MemberCanBePrivate.Global + protected readonly struct ObjectResponseResult + { + // ReSharper disable once ConvertToPrimaryConstructor + public ObjectResponseResult(T responseObject, string responseText) + { + Object = responseObject; + Text = responseText; + } + + public T Object { get; } + + public string Text { get; } + } +} + +public interface IMessagesEventStreamListener +{ + /// Callback method + /// + /// A cancellation token that can be used by other objects or threads to receive notice of + /// cancellation. + /// + /// A server side error occurred. + Task StreamEventsAsync( + Action onEvent, + CancellationToken cancellationToken = default); + + /// Callback method + /// + /// A cancellation token that can be used by other objects or threads to receive notice of + /// cancellation. + /// + /// A server side error occurred. + Task StreamEventsAsync( + Func onEvent, + CancellationToken cancellationToken = default); +} + +public sealed class MessagesEventStreamListener : EventStreamListener, IMessagesEventStreamListener +{ + // ReSharper disable once ConvertToPrimaryConstructor + public MessagesEventStreamListener( + string baseUrl, + HttpClient httpClient, + string eventEndpointPath = "/messages/event-stream") + : base(baseUrl, httpClient, eventEndpointPath) + { + } + + protected override string DefaultEventName => "message"; + + /// Callback function + /// + /// A cancellation token that can be used by other objects or threads to receive notice of + /// cancellation. + /// + /// A server side error occurred. + public new Task StreamEventsAsync( + Action onEvent, + CancellationToken cancellationToken = default) + { + return base.StreamEventsAsync(onEvent, cancellationToken); + } + + /// Callback function + /// + /// A cancellation token that can be used by other objects or threads to receive notice of + /// cancellation. + /// + /// A server side error occurred. + public new Task StreamEventsAsync( + Func onEvent, + CancellationToken cancellationToken = default) + { + return base.StreamEventsAsync(onEvent, cancellationToken); + } + + protected override Task StreamEventsCoreAsync( + Func onEvent, + CancellationToken cancellationToken) + { + return OpenEventStreamAsync(onEvent, DefaultEventName, cancellationToken); + } +} \ No newline at end of file diff --git a/swagger/RedShirt.Example.Api.Interop/ProblemDetails.cs b/swagger/RedShirt.Example.Api.Interop/ProblemDetails.cs new file mode 100644 index 0000000..65cbe75 --- /dev/null +++ b/swagger/RedShirt.Example.Api.Interop/ProblemDetails.cs @@ -0,0 +1,29 @@ +using System.Collections.Generic; +using System.Text.Json.Serialization; + +namespace RedShirt.Example.Api.Interop; + +// Hand-maintained DTO listed in nswag.json excludedTypeNames so NSwag does not emit a duplicate. +// Same cold-build constraint as SwaggerException.cs. +// EventStreamListener references this type before Client.generated.cs exists. +// It must live outside the generated file even though OpenAPI client methods also use it. +public class ProblemDetails +{ + [JsonPropertyName("type")] + public string Type { get; set; } = string.Empty; + + [JsonPropertyName("title")] + public string Title { get; set; } = string.Empty; + + [JsonPropertyName("status")] + public int? Status { get; set; } + + [JsonPropertyName("detail")] + public string Detail { get; set; } = string.Empty; + + [JsonPropertyName("instance")] + public string Instance { get; set; } = string.Empty; + + [JsonExtensionData] + public IDictionary? AdditionalProperties { get; set; } +} diff --git a/swagger/RedShirt.Example.Api.Interop/RedShirt.Example.Api.Interop.csproj b/swagger/RedShirt.Example.Api.Interop/RedShirt.Example.Api.Interop.csproj index 05505a0..491aba0 100644 --- a/swagger/RedShirt.Example.Api.Interop/RedShirt.Example.Api.Interop.csproj +++ b/swagger/RedShirt.Example.Api.Interop/RedShirt.Example.Api.Interop.csproj @@ -15,8 +15,12 @@ - - + + diff --git a/swagger/RedShirt.Example.Api.Interop/SwaggerException.cs b/swagger/RedShirt.Example.Api.Interop/SwaggerException.cs new file mode 100644 index 0000000..0e2509c --- /dev/null +++ b/swagger/RedShirt.Example.Api.Interop/SwaggerException.cs @@ -0,0 +1,58 @@ +using System; +using System.Collections.Generic; + +namespace RedShirt.Example.Api.Interop; + +// Hand-maintained counterpart to NSwag exceptionClass output (see nswag.json, generateExceptionClasses=false). +// EventStreamListener.cs is compiled in the same MSBuild pass as Client.generated.cs. +// Compile inputs are resolved before the NSwag CoreCompile target runs. +// These types must already exist on disk for a cold build from before Client.generated.cs to succeed. +// Do not remove in favour of generated copies without fixing that ordering. +// BeforeTargets=CoreCompile is not enough on its own. +public class SwaggerException : Exception +{ + public int StatusCode { get; } + + public string? Response { get; } + + public IReadOnlyDictionary> Headers { get; } + + // ReSharper disable once ConvertToPrimaryConstructor + public SwaggerException( + string message, + int statusCode, + string? response, + IReadOnlyDictionary> headers, + Exception? innerException) + : base( + message + "\n\nStatus: " + statusCode + "\nResponse: \n" + + (response == null ? "(null)" : response.Substring(0, response.Length >= 512 ? 512 : response.Length)), + innerException) + { + StatusCode = statusCode; + Response = response; + Headers = headers; + } + + public override string ToString() + { + return string.Format("HTTP Response: \n\n{0}\n\n{1}", Response, base.ToString()); + } +} + +public class SwaggerException : SwaggerException +{ + public TResult Result { get; } + + public SwaggerException( + string message, + int statusCode, + string? response, + IReadOnlyDictionary> headers, + TResult result, + Exception? innerException) + : base(message, statusCode, response, headers, innerException) + { + Result = result; + } +} diff --git a/swagger/RedShirt.Example.Api.Interop/nswag.json b/swagger/RedShirt.Example.Api.Interop/nswag.json index 02ac53a..5088afa 100644 --- a/swagger/RedShirt.Example.Api.Interop/nswag.json +++ b/swagger/RedShirt.Example.Api.Interop/nswag.json @@ -13,7 +13,7 @@ "generateDtoTypes": true, "injectHttpClient": true, "disposeHttpClient": true, - "generateExceptionClasses": true, + "generateExceptionClasses": false, "exceptionClass": "SwaggerException", "wrapDtoExceptions": true, "useHttpClientCreationMethod": false, @@ -54,7 +54,9 @@ "classStyle": "Poco", "generateDefaultValues": true, "generateDataAnnotations": true, - "excludedTypeNames": [], + "excludedTypeNames": [ + "ProblemDetails" + ], "handleReferences": false, "generateImmutableArrayProperties": false, "generateImmutableDictionaryProperties": false, diff --git a/swagger/RedShirt.Example.Api.Interop/swagger.json b/swagger/RedShirt.Example.Api.Interop/swagger.json index 1a63f6e..d70478c 100644 --- a/swagger/RedShirt.Example.Api.Interop/swagger.json +++ b/swagger/RedShirt.Example.Api.Interop/swagger.json @@ -830,6 +830,72 @@ } } }, + "/messages/event-stream": { + "get": { + "tags": [ + "Message" + ], + "summary": "Event stream", + "description": "This GET /messages/event-stream endpoint is not directly supported by OpenAPI.\nThe generated NSwag MessageClient.GetEventStreamAsync method is intended for connectivity checks only.\nUse MessagesEventStreamListener in RedShirt.Example.Api.Interop project to receive message events.", + "operationId": "Message_GetEventStream", + "responses": { + "200": { + "description": "" + }, + "400": { + "description": "", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ProblemDetails" + } + } + } + }, + "502": { + "description": "" + } + } + } + }, + "/messages": { + "post": { + "tags": [ + "Message" + ], + "operationId": "Message_Post", + "requestBody": { + "x-name": "request", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ExampleMessagePostRequest" + } + } + }, + "required": true, + "x-position": 1 + }, + "responses": { + "202": { + "description": "" + }, + "400": { + "description": "", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ProblemDetails" + } + } + } + }, + "502": { + "description": "" + } + } + } + }, "/orders/{id}": { "delete": { "tags": [ @@ -2544,6 +2610,18 @@ } } }, + "ExampleMessagePostRequest": { + "type": "object", + "additionalProperties": false, + "required": [ + "message" + ], + "properties": { + "message": { + "type": "string" + } + } + }, "OrderDto": { "type": "object", "additionalProperties": false, diff --git a/test/RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws.UnitTests/RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws.UnitTests.csproj b/test/RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws.UnitTests/RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws.UnitTests.csproj new file mode 100644 index 0000000..f921156 --- /dev/null +++ b/test/RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws.UnitTests/RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws.UnitTests.csproj @@ -0,0 +1,37 @@ + + + + net10.0 + enable + enable + + false + true + + + + + + all + runtime; build; native; contentfiles; analyzers; buildtransitive + + + + + + + + all + runtime; build; native; contentfiles; analyzers; buildtransitive + + + + + + + + + + + + diff --git a/test/RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws.UnitTests/Support/TestUtilities.cs b/test/RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws.UnitTests/Support/TestUtilities.cs new file mode 100644 index 0000000..396d0a1 --- /dev/null +++ b/test/RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws.UnitTests/Support/TestUtilities.cs @@ -0,0 +1,38 @@ +namespace RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws.UnitTests.Support; + +internal static class TestUtilities +{ + internal static void WrapEnvironment(IReadOnlyDictionary values, Action action) + { + var previous = new Dictionary(StringComparer.Ordinal); + + foreach (var (key, value) in values) + { + previous[key] = Environment.GetEnvironmentVariable(key); + Environment.SetEnvironmentVariable(key, value); + } + + try + { + action(); + } + finally + { + foreach (var (key, value) in previous) + { + Environment.SetEnvironmentVariable(key, value); + } + } + } + + internal static void WrapLocalAwsEnvironment(Action action) + { + WrapEnvironment(new Dictionary + { + ["AWS_SERVICE_URL"] = "http://localhost:4566", + ["AWS_ACCESS_KEY_ID"] = "foo", + ["AWS_SECRET_ACCESS_KEY"] = "bar", + ["AWS_REGION"] = "us-east-1" + }, action); + } +} \ No newline at end of file diff --git a/test/RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws.UnitTests/Tests/Extensions/ServiceCollectionExtensionsTests.cs b/test/RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws.UnitTests/Tests/Extensions/ServiceCollectionExtensionsTests.cs new file mode 100644 index 0000000..8b48d05 --- /dev/null +++ b/test/RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws.UnitTests/Tests/Extensions/ServiceCollectionExtensionsTests.cs @@ -0,0 +1,48 @@ +using Amazon.IoT; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws.Extensions; +using RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws.Services; +using RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws.UnitTests.Support; +using RedShirt.Example.Api.ClientEvents.Library.Mqtt.Services; + +namespace RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws.UnitTests.Tests.Extensions; + +public class ServiceCollectionExtensionsTests +{ + [Fact] + public void AddAwsMqttClientEvents_RegistersAmazonIoTClient() + { + var configuration = new ConfigurationBuilder().Build(); + + TestUtilities.WrapLocalAwsEnvironment(() => + { + var provider = new ServiceCollection() + .AddAwsMqttClientEvents(configuration) + .BuildServiceProvider(); + + Assert.NotNull(provider.GetService()); + }); + } + + [Fact] + public void AddAwsMqttClientEvents_RegistersBrokerUrlResolver() + { + var configuration = new ConfigurationBuilder() + .AddInMemoryCollection(new Dictionary + { + ["ClientEvents:Mqtt:AWS:BrokerConnectHost"] = "ministack" + }) + .Build(); + + TestUtilities.WrapLocalAwsEnvironment(() => + { + var provider = new ServiceCollection() + .AddAwsMqttClientEvents(configuration) + .BuildServiceProvider(); + + var resolver = provider.GetRequiredService(); + Assert.IsType(resolver); + }); + } +} \ No newline at end of file diff --git a/test/RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws.UnitTests/Tests/Services/AwsIotMqttBrokerUrlResolverTests.cs b/test/RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws.UnitTests/Tests/Services/AwsIotMqttBrokerUrlResolverTests.cs new file mode 100644 index 0000000..a5dee68 --- /dev/null +++ b/test/RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws.UnitTests/Tests/Services/AwsIotMqttBrokerUrlResolverTests.cs @@ -0,0 +1,107 @@ +using Amazon.IoT; +using Amazon.IoT.Model; +using Microsoft.Extensions.Options; +using Moq; +using RedShirt.Example.Api.ClientEvents.Library.Core.Exceptions; +using RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws.Services; + +namespace RedShirt.Example.Api.ClientEvents.Library.Mqtt.Aws.UnitTests.Tests.Services; + +public class AwsIotMqttBrokerUrlResolverTests +{ + private static AwsIotMqttBrokerUrlResolver CreateSut( + Mock amazonIoT, + AwsIotMqttBrokerUrlResolver.ConfigurationModel? configuration = null) + { + return new AwsIotMqttBrokerUrlResolver( + amazonIoT.Object, + Options.Create(configuration ?? new AwsIotMqttBrokerUrlResolver.ConfigurationModel())); + } + + [Fact] + public async Task ResolveBrokerUrlAsync_ReturnsConnectHostAndHostHeader_WhenBrokerConnectHostSet() + { + const string endpointAddress = "abc123-ats.iot.us-east-1.ministack:4566"; + var amazonIoT = new Mock(MockBehavior.Strict); + amazonIoT + .Setup(i => i.DescribeEndpointAsync(It.IsAny(), It.IsAny())) + .ReturnsAsync(new DescribeEndpointResponse + { + EndpointAddress = endpointAddress + }); + + var sut = CreateSut(amazonIoT, new AwsIotMqttBrokerUrlResolver.ConfigurationModel + { + BrokerConnectHost = "ministack" + }); + + var target = await sut.ResolveBrokerUrlAsync(TestContext.Current.CancellationToken); + + Assert.Equal("ws://ministack:4566/mqtt", target.BrokerUrl); + Assert.Equal(endpointAddress, target.WebSocketHostHeader); + amazonIoT.VerifyAll(); + } + + [Fact] + public async Task ResolveBrokerUrlAsync_ReturnsDirectEndpoint_WhenBrokerConnectHostUnset() + { + const string endpointAddress = "abc123-ats.iot.us-east-1.amazonaws.com"; + var amazonIoT = new Mock(MockBehavior.Strict); + amazonIoT + .Setup(i => i.DescribeEndpointAsync( + It.Is(request => request.EndpointType == "iot:Data-ATS"), + It.IsAny())) + .ReturnsAsync(new DescribeEndpointResponse + { + EndpointAddress = endpointAddress + }); + + var sut = CreateSut(amazonIoT); + + var target = await sut.ResolveBrokerUrlAsync(TestContext.Current.CancellationToken); + + Assert.Equal($"ws://{endpointAddress}/mqtt", target.BrokerUrl); + Assert.Null(target.WebSocketHostHeader); + amazonIoT.VerifyAll(); + } + + [Fact] + public async Task ResolveBrokerUrlAsync_Throws_WhenEndpointAddressInvalid() + { + var amazonIoT = new Mock(MockBehavior.Strict); + amazonIoT + .Setup(i => i.DescribeEndpointAsync(It.IsAny(), It.IsAny())) + .ReturnsAsync(new DescribeEndpointResponse + { + EndpointAddress = "://not-a-valid-host" + }); + + var sut = CreateSut(amazonIoT); + + var exception = await Assert.ThrowsAsync(() => + sut.ResolveBrokerUrlAsync(TestContext.Current.CancellationToken)); + + Assert.False(exception.CouldBeTransient); + amazonIoT.VerifyAll(); + } + + [Fact] + public async Task ResolveBrokerUrlAsync_Throws_WhenEndpointAddressMissing() + { + var amazonIoT = new Mock(MockBehavior.Strict); + amazonIoT + .Setup(i => i.DescribeEndpointAsync(It.IsAny(), It.IsAny())) + .ReturnsAsync(new DescribeEndpointResponse + { + EndpointAddress = null + }); + + var sut = CreateSut(amazonIoT); + + var exception = await Assert.ThrowsAsync(() => + sut.ResolveBrokerUrlAsync(TestContext.Current.CancellationToken)); + + Assert.True(exception.CouldBeTransient); + amazonIoT.VerifyAll(); + } +} \ No newline at end of file diff --git a/test/RedShirt.Example.Api.ClientEvents.Library.Mqtt.UnitTests/RedShirt.Example.Api.ClientEvents.Library.Mqtt.UnitTests.csproj b/test/RedShirt.Example.Api.ClientEvents.Library.Mqtt.UnitTests/RedShirt.Example.Api.ClientEvents.Library.Mqtt.UnitTests.csproj new file mode 100644 index 0000000..94058f1 --- /dev/null +++ b/test/RedShirt.Example.Api.ClientEvents.Library.Mqtt.UnitTests/RedShirt.Example.Api.ClientEvents.Library.Mqtt.UnitTests.csproj @@ -0,0 +1,36 @@ + + + + net10.0 + enable + enable + + false + true + + + + + all + runtime; build; native; contentfiles; analyzers; buildtransitive + + + + + + + + all + runtime; build; native; contentfiles; analyzers; buildtransitive + + + + + + + + + + + + diff --git a/test/RedShirt.Example.Api.ClientEvents.Library.Mqtt.UnitTests/Tests/Factories/ApiMqttClientFactoryTests.cs b/test/RedShirt.Example.Api.ClientEvents.Library.Mqtt.UnitTests/Tests/Factories/ApiMqttClientFactoryTests.cs new file mode 100644 index 0000000..417c9e9 --- /dev/null +++ b/test/RedShirt.Example.Api.ClientEvents.Library.Mqtt.UnitTests/Tests/Factories/ApiMqttClientFactoryTests.cs @@ -0,0 +1,270 @@ +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using Moq; +using MQTTnet; +using MQTTnet.Formatter; +using RedShirt.Example.Api.ClientEvents.Library.Core.Exceptions; +using RedShirt.Example.Api.ClientEvents.Library.Mqtt.Factories; +using RedShirt.Example.Api.ClientEvents.Library.Mqtt.Models; +using RedShirt.Example.Api.ClientEvents.Library.Mqtt.Services; +using RedShirt.Example.Api.Common.SecretManagers.Core.Services; +using System.Reflection; + +namespace RedShirt.Example.Api.ClientEvents.Library.Mqtt.UnitTests.Tests.Factories; + +public class ApiMqttClientFactoryTests +{ + private static ApiMqttClientFactory CreateSut( + ApiMqttClientFactory.ConfigurationModel configuration, + Mock? brokerUrlResolver = null, + Mock? secretManager = null, + ILogger? logger = null) + { + brokerUrlResolver ??= new Mock(MockBehavior.Strict); + secretManager ??= new Mock(MockBehavior.Strict); + + return new ApiMqttClientFactory( + Options.Create(configuration), + secretManager.Object, + brokerUrlResolver.Object, + logger ?? NullLogger.Instance); + } + + private static void VerifyLogWarningContaining( + Mock> logger, + string expectedSubstring, + Times times) + { + logger.Verify( + x => x.Log( + LogLevel.Warning, + It.IsAny(), + It.Is((state, _) => + state.ToString()!.Contains(expectedSubstring, StringComparison.Ordinal)), + It.IsAny(), + It.IsAny>()), + times); + } + + private static async Task InvokeGetOptionsBuilderAsync( + ApiMqttClientFactory factory, + ApiMqttClientFactory.ConfigurationModel sutConfiguration) + { + var method = typeof(ApiMqttClientFactory).GetMethod( + "GetOptionsBuilderAsync", + BindingFlags.Instance | BindingFlags.NonPublic); + Assert.NotNull(method); + + var invokeResult = method.Invoke(factory, [sutConfiguration, CancellationToken.None]); + Assert.NotNull(invokeResult); + + var resultTask = (Task) invokeResult; + await resultTask.ConfigureAwait(false); + + var result = resultTask.GetType().GetProperty("Result")!.GetValue(resultTask); + Assert.NotNull(result); + + var optionsBuilder = result.GetType().GetProperty("OptionsBuilder")!.GetValue(result); + Assert.NotNull(optionsBuilder); + + var buildMethod = optionsBuilder.GetType().GetMethod("Build"); + Assert.NotNull(buildMethod); + + var options = buildMethod.Invoke(optionsBuilder, null); + Assert.IsType(options); + + return (MqttClientOptions) options; + } + + [Fact] + public async Task CreateConnectedClientAsync_WhenBrokerUrlMissing_ThrowsBeforeConnecting() + { + var brokerUrlResolver = new Mock(MockBehavior.Strict); + var sut = CreateSut( + new ApiMqttClientFactory.ConfigurationModel + { + ResolveBrokerAddressExternally = false, + BrokerUrl = null + }, + brokerUrlResolver); + + var exception = await Assert.ThrowsAsync(() => + sut.CreateConnectedClientAsync(TestContext.Current.CancellationToken)); + + Assert.Contains("broker URL", exception.Message, StringComparison.OrdinalIgnoreCase); + brokerUrlResolver.VerifyNoOtherCalls(); + } + + [Fact] + public async Task CreateConnectedClientAsync_WhenCredentialsConfigured_ResolvesSecretsBeforeConnecting() + { + var brokerUrlResolver = new Mock(MockBehavior.Strict); + var secretManager = new Mock(MockBehavior.Strict); + secretManager + .Setup(s => s.GetSecretsAsync( + It.Is>(paths => + paths.Contains("/mqtt/user") && paths.Contains("/mqtt/pass")), + It.IsAny())) + .ReturnsAsync(new Dictionary + { + ["/mqtt/user"] = "mqtt-user", + ["/mqtt/pass"] = "mqtt-pass" + }); + + var sut = CreateSut( + new ApiMqttClientFactory.ConfigurationModel + { + ResolveBrokerAddressExternally = false, + BrokerUrl = "mqtt://127.0.0.1:1", + UsernameSecretPath = "/mqtt/user", + PasswordSecretPath = "/mqtt/pass" + }, + brokerUrlResolver, + secretManager); + + await Assert.ThrowsAnyAsync(() => + sut.CreateConnectedClientAsync(TestContext.Current.CancellationToken)); + + secretManager.Verify( + s => s.GetSecretsAsync(It.IsAny>(), It.IsAny()), + Times.Once); + brokerUrlResolver.VerifyNoOtherCalls(); + } + + [Fact] + public async Task CreateConnectedClientAsync_WhenProtocolVersionInvalid_LogsWarningAndStillAttemptsConnection() + { + var logger = new Mock>(); + var sut = CreateSut( + new ApiMqttClientFactory.ConfigurationModel + { + ResolveBrokerAddressExternally = false, + BrokerUrl = "mqtt://127.0.0.1:1", + ProtocolVersion = "not-a-version" + }, + logger: logger.Object); + + var exception = await Assert.ThrowsAsync(() => + sut.CreateConnectedClientAsync(TestContext.Current.CancellationToken)); + + Assert.NotEqual("not-a-version", exception.Message, StringComparer.Ordinal); + VerifyLogWarningContaining(logger, "Failed to parse MQTT protocol version", Times.Once()); + } + + [Fact] + public async Task CreateConnectedClientAsync_WhenProtocolVersionValid_DoesNotLogParseWarning() + { + var logger = new Mock>(); + var sut = CreateSut( + new ApiMqttClientFactory.ConfigurationModel + { + ResolveBrokerAddressExternally = false, + BrokerUrl = "mqtt://127.0.0.1:1", + ProtocolVersion = "V311" + }, + logger: logger.Object); + + await Assert.ThrowsAsync(() => + sut.CreateConnectedClientAsync(TestContext.Current.CancellationToken)); + + VerifyLogWarningContaining(logger, "Failed to parse MQTT protocol version", Times.Never()); + } + + [Fact] + public async Task CreateConnectedClientAsync_WhenResolveExternallyFalse_DoesNotCallResolver() + { + var brokerUrlResolver = new Mock(MockBehavior.Strict); + var sut = CreateSut( + new ApiMqttClientFactory.ConfigurationModel + { + ResolveBrokerAddressExternally = false, + BrokerUrl = "mqtt://127.0.0.1:1" + }, + brokerUrlResolver); + + await Assert.ThrowsAnyAsync(() => + sut.CreateConnectedClientAsync(TestContext.Current.CancellationToken)); + + brokerUrlResolver.VerifyNoOtherCalls(); + } + + [Fact] + public async Task CreateConnectedClientAsync_WhenResolveExternallyTrue_CallsResolver() + { + var brokerUrlResolver = new Mock(MockBehavior.Strict); + brokerUrlResolver + .Setup(r => r.ResolveBrokerUrlAsync(It.IsAny())) + .ReturnsAsync(new MqttBrokerTarget + { + BrokerUrl = "mqtt://127.0.0.1:1" + }); + + var sut = CreateSut( + new ApiMqttClientFactory.ConfigurationModel + { + ResolveBrokerAddressExternally = true + }, + brokerUrlResolver); + + await Assert.ThrowsAnyAsync(() => + sut.CreateConnectedClientAsync(TestContext.Current.CancellationToken)); + + brokerUrlResolver.Verify(r => r.ResolveBrokerUrlAsync(It.IsAny()), Times.Once); + brokerUrlResolver.VerifyNoOtherCalls(); + } + + [Theory] + [InlineData(null, "RedShirt.Example.Api")] + [InlineData("", "RedShirt.Example.Api")] + [InlineData(" ", "RedShirt.Example.Api")] + [InlineData("My.Custom.Prefix", "My.Custom.Prefix")] + public async Task GetOptionsBuilderAsync_WhenClientIdPrefixConfigured_UsesExpectedClientIdPrefix( + string? clientIdPrefix, + string expectedPrefix) + { + var config = new ApiMqttClientFactory.ConfigurationModel + { + ResolveBrokerAddressExternally = false, + BrokerUrl = "mqtt://127.0.0.1:1", + ClientIdPrefix = clientIdPrefix + }; + + var sut = CreateSut(config); + var options = await InvokeGetOptionsBuilderAsync(sut, config); + + var serverHostname = string.IsNullOrWhiteSpace(Environment.MachineName) + ? "unknown" + : Environment.MachineName; + + Assert.StartsWith($"{expectedPrefix}/{serverHostname}/", options.ClientId); + + var clientIdSegments = options.ClientId.Split('/'); + Assert.Equal(3, clientIdSegments.Length); + Assert.Matches("^[0-9a-f]{32}$", clientIdSegments[2]); + } + + [Theory] + [InlineData(null, MqttProtocolVersion.V500)] + [InlineData("", MqttProtocolVersion.V500)] + [InlineData(" ", MqttProtocolVersion.V500)] + [InlineData("V310", MqttProtocolVersion.V310)] + [InlineData("V311", MqttProtocolVersion.V311)] + [InlineData("V500", MqttProtocolVersion.V500)] + public async Task GetOptionsBuilderAsync_WhenProtocolVersionConfigured_AppliesExpectedMqttProtocolVersion( + string? protocolVersion, + MqttProtocolVersion expectedProtocolVersion) + { + var config = new ApiMqttClientFactory.ConfigurationModel + { + ResolveBrokerAddressExternally = false, + BrokerUrl = "mqtt://127.0.0.1:1", + ProtocolVersion = protocolVersion + }; + + var sut = CreateSut(config); + var options = await InvokeGetOptionsBuilderAsync(sut, config); + + Assert.Equal(expectedProtocolVersion, options.ProtocolVersion); + } +} \ No newline at end of file diff --git a/test/RedShirt.Example.Api.ClientEvents.Library.Mqtt.UnitTests/Tests/Services/MqttApiClientEventSenderTests.cs b/test/RedShirt.Example.Api.ClientEvents.Library.Mqtt.UnitTests/Tests/Services/MqttApiClientEventSenderTests.cs new file mode 100644 index 0000000..d856e8f --- /dev/null +++ b/test/RedShirt.Example.Api.ClientEvents.Library.Mqtt.UnitTests/Tests/Services/MqttApiClientEventSenderTests.cs @@ -0,0 +1,73 @@ +using Microsoft.Extensions.Logging.Abstractions; +using Moq; +using RedShirt.Example.Api.ClientEvents.Library.Core.Exceptions; +using RedShirt.Example.Api.ClientEvents.Library.Core.Models; +using RedShirt.Example.Api.ClientEvents.Library.Mqtt.Factories; +using RedShirt.Example.Api.ClientEvents.Library.Mqtt.Services; +using RedShirt.Example.Api.ClientEvents.Library.Mqtt.Services.Resilience; + +namespace RedShirt.Example.Api.ClientEvents.Library.Mqtt.UnitTests.Tests.Services; + +public class MqttApiClientEventSenderTests +{ + [Theory] + [InlineData(null)] + [InlineData("")] + [InlineData(" ")] + public async Task SendAsync_WhenTopicMissing_ThrowsWithoutCallingDependencies(string? topic) + { + var mqttClientFactory = new Mock(MockBehavior.Strict); + var retryWrapper = new Mock(MockBehavior.Strict); + + var sut = new MqttApiClientEventSender( + mqttClientFactory.Object, + retryWrapper.Object, + NullLogger>.Instance); + + var exception = await Assert.ThrowsAsync(() => + sut.SendAsync(new ApiClientEventSendRequest + { + Topic = topic, + Payload = "hello" + }, TestContext.Current.CancellationToken)); + + Assert.Contains("topic", exception.Message, StringComparison.OrdinalIgnoreCase); + mqttClientFactory.VerifyNoOtherCalls(); + retryWrapper.VerifyNoOtherCalls(); + } + + [Fact] + public async Task SendAsync_WhenTopicPresent_DelegatesToRetryWrapper() + { + var mqttClientFactory = new Mock(MockBehavior.Strict); + var retryWrapper = new Mock(MockBehavior.Strict); + retryWrapper + .Setup(r => r.RunAsync(It.IsAny>(), It.IsAny())) + .Returns, CancellationToken>((func, token) => func(token)); + + mqttClientFactory + .Setup(f => f.CreateConnectedClientAsync(It.IsAny())) + .ThrowsAsync(new ApiClientEventsException("connect failed") + { + CouldBeTransient = true, + IsHandled = false + }); + + var sut = new MqttApiClientEventSender( + mqttClientFactory.Object, + retryWrapper.Object, + NullLogger>.Instance); + + await Assert.ThrowsAsync(() => + sut.SendAsync(new ApiClientEventSendRequest + { + Topic = "example-message/user/test", + Payload = "hello" + }, TestContext.Current.CancellationToken)); + + retryWrapper.Verify( + r => r.RunAsync(It.IsAny>(), It.IsAny()), + Times.Once); + mqttClientFactory.Verify(f => f.CreateConnectedClientAsync(It.IsAny()), Times.Once); + } +} \ No newline at end of file diff --git a/test/RedShirt.Example.Api.ClientEvents.Library.Mqtt.UnitTests/Tests/Services/Resilience/MqttClientEventsExceptionArbiterServiceTests.cs b/test/RedShirt.Example.Api.ClientEvents.Library.Mqtt.UnitTests/Tests/Services/Resilience/MqttClientEventsExceptionArbiterServiceTests.cs new file mode 100644 index 0000000..b6ff0be --- /dev/null +++ b/test/RedShirt.Example.Api.ClientEvents.Library.Mqtt.UnitTests/Tests/Services/Resilience/MqttClientEventsExceptionArbiterServiceTests.cs @@ -0,0 +1,122 @@ +using RedShirt.Example.Api.ClientEvents.Library.Core.Exceptions; +using RedShirt.Example.Api.ClientEvents.Library.Mqtt.Services.Resilience; +using RedShirt.Example.Api.Common.SecretManagers.Core.Exceptions; +using System.Net.Sockets; +using System.Text.Json; + +namespace RedShirt.Example.Api.ClientEvents.Library.Mqtt.UnitTests.Tests.Services.Resilience; + +public class MqttClientEventsExceptionArbiterServiceTests +{ + private readonly MqttClientEventsExceptionArbiterService _sut = new(); + + [Fact] + public void GetReport_ApiClientEventsException_IsAlreadyHandled() + { + var report = _sut.GetReport(new ApiClientEventsException("done") + { + IsHandled = true, + CouldBeTransient = true + }); + + Assert.True(report.AlreadyHandled); + Assert.True(report.IsExpected); + Assert.True(report.CouldBeTransient); + } + + [Fact] + public void GetReport_ApiClientEventsException_NonTransient_IsExpected() + { + var report = _sut.GetReport(new ApiClientEventsException("config") + { + IsHandled = false, + CouldBeTransient = false + }); + + Assert.False(report.AlreadyHandled); + Assert.True(report.IsExpected); + Assert.False(report.CouldBeTransient); + } + + [Fact] + public void GetReport_ApiClientEventsException_Transient_IsExpected() + { + var report = _sut.GetReport(new ApiClientEventsException("timeout") + { + IsHandled = false, + CouldBeTransient = true + }); + + Assert.False(report.AlreadyHandled); + Assert.True(report.IsExpected); + Assert.True(report.CouldBeTransient); + } + + [Fact] + public void GetReport_ApiSecretManagerException_IsAlreadyHandled() + { + var report = _sut.GetReport(new ApiSecretManagerException(new InvalidOperationException("ssm")) + { + IsHandled = true, + CouldBeTransient = true, + CouldBeExternallySolvable = true + }); + + Assert.True(report.AlreadyHandled); + Assert.True(report.IsExpected); + } + + [Fact] + public void GetReport_JsonException_IsExpectedNonTransient() + { + var report = _sut.GetReport(new JsonException("bad payload")); + + Assert.True(report.IsExpected); + Assert.False(report.CouldBeTransient); + } + + [Fact] + public void GetReport_OperationCanceledException_IsNotTransient() + { + var report = _sut.GetReport(new OperationCanceledException()); + + Assert.True(report.IsExpected); + Assert.False(report.CouldBeTransient); + } + + [Fact] + public void GetReport_SocketException_IsTransient() + { + var report = _sut.GetReport(new SocketException()); + + Assert.True(report.IsExpected); + Assert.True(report.CouldBeTransient); + } + + [Fact] + public void GetReport_TaskCanceledException_IsTransient() + { + var report = _sut.GetReport(new TaskCanceledException()); + + Assert.True(report.IsExpected); + Assert.True(report.CouldBeTransient); + } + + [Fact] + public void GetReport_UnknownException_IsUnexpected() + { + var report = _sut.GetReport(new InvalidOperationException("surprise")); + + Assert.False(report.IsExpected); + Assert.False(report.CouldBeTransient); + } + + [Fact] + public void GetReport_UnwrapsSingleAggregateException() + { + var report = _sut.GetReport(new AggregateException(new SocketException())); + + Assert.True(report.IsExpected); + Assert.True(report.CouldBeTransient); + } +} \ No newline at end of file diff --git a/test/RedShirt.Example.Api.ClientEvents.Library.Mqtt.UnitTests/Tests/Services/Resilience/MqttClientEventsRetryWrapperServiceTests.cs b/test/RedShirt.Example.Api.ClientEvents.Library.Mqtt.UnitTests/Tests/Services/Resilience/MqttClientEventsRetryWrapperServiceTests.cs new file mode 100644 index 0000000..b54f849 --- /dev/null +++ b/test/RedShirt.Example.Api.ClientEvents.Library.Mqtt.UnitTests/Tests/Services/Resilience/MqttClientEventsRetryWrapperServiceTests.cs @@ -0,0 +1,111 @@ +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using Moq; +using RedShirt.Example.Api.ClientEvents.Library.Core.Exceptions; +using RedShirt.Example.Api.ClientEvents.Library.Mqtt.Services.Resilience; +using RedShirt.Example.Api.Common.Services.Utility; +using System.Net.Sockets; +using System.Text.Json; + +namespace RedShirt.Example.Api.ClientEvents.Library.Mqtt.UnitTests.Tests.Services.Resilience; + +public class MqttClientEventsRetryWrapperServiceTests +{ + private static MqttClientEventsRetryWrapperService CreateSut( + IMqttClientEventsExceptionArbiterService? arbiter = null, + ISleepService? sleep = null, + int? retryCount = 3) + { + var sleepMock = new Mock(); + sleepMock + .Setup(s => s.DelayAsync(It.IsAny(), It.IsAny())) + .Returns(Task.CompletedTask); + + return new MqttClientEventsRetryWrapperService( + arbiter ?? new MqttClientEventsExceptionArbiterService(), + NullLogger.Instance, + sleep ?? sleepMock.Object, + Options.Create(new MqttClientEventsRetryWrapperService.ConfigurationModel + { + RetryCount = retryCount + })); + } + + [Fact] + public async Task RunAsync_DoesNotWrap_UnexpectedExceptions() + { + var sut = CreateSut(); + + await Assert.ThrowsAsync(() => + sut.RunAsync(_ => throw new InvalidOperationException("surprise"), + TestContext.Current.CancellationToken)); + } + + [Fact] + public async Task RunAsync_PassesThrough_HandledApiClientEventsException() + { + var sut = CreateSut(); + + var original = new ApiClientEventsException("done") + { + IsHandled = true, + CouldBeTransient = true + }; + + var thrown = await Assert.ThrowsAsync(() => + sut.RunAsync(_ => throw original, TestContext.Current.CancellationToken)); + + Assert.Same(original, thrown); + } + + [Fact] + public async Task RunAsync_RetriesTransientFailures_ThenSucceeds() + { + var attempts = 0; + var sleep = new Mock(MockBehavior.Strict); + sleep + .Setup(s => s.DelayAsync(It.IsAny(), It.IsAny())) + .Returns(Task.CompletedTask); + + var sut = CreateSut(sleep: sleep.Object); + + var result = await sut.RunAsync(_ => + { + attempts++; + if (attempts < 3) + { + throw new SocketException(); + } + + return Task.FromResult("ok"); + }, TestContext.Current.CancellationToken); + + Assert.Equal("ok", result); + Assert.Equal(3, attempts); + sleep.Verify(s => s.DelayAsync(It.IsAny(), It.IsAny()), Times.Exactly(2)); + } + + [Fact] + public async Task RunAsync_ReturnsResult_WhenFuncSucceeds() + { + var sut = CreateSut(); + + var result = await sut.RunAsync(_ => Task.FromResult(42), TestContext.Current.CancellationToken); + + Assert.Equal(42, result); + } + + [Fact] + public async Task RunAsync_WrapsExpectedNonTransient_AsApiClientEventsException() + { + var sut = CreateSut(); + + var wrapped = await Assert.ThrowsAsync(() => + sut.RunAsync(_ => throw new JsonException("bad payload"), + TestContext.Current.CancellationToken)); + + Assert.True(wrapped.IsHandled); + Assert.False(wrapped.CouldBeTransient); + Assert.IsType(wrapped.InnerException); + } +} \ No newline at end of file diff --git a/test/RedShirt.Example.Api.Interop.UnitTests/RedShirt.Example.Api.Interop.UnitTests.csproj b/test/RedShirt.Example.Api.Interop.UnitTests/RedShirt.Example.Api.Interop.UnitTests.csproj new file mode 100644 index 0000000..7ef1a61 --- /dev/null +++ b/test/RedShirt.Example.Api.Interop.UnitTests/RedShirt.Example.Api.Interop.UnitTests.csproj @@ -0,0 +1,34 @@ + + + + net10.0 + enable + enable + + false + true + + + + + all + runtime; build; native; contentfiles; analyzers; buildtransitive + + + + + + all + runtime; build; native; contentfiles; analyzers; buildtransitive + + + + + + + + + + + + diff --git a/test/RedShirt.Example.Api.Interop.UnitTests/Tests/EventStreamListenerTests.cs b/test/RedShirt.Example.Api.Interop.UnitTests/Tests/EventStreamListenerTests.cs new file mode 100644 index 0000000..147ea09 --- /dev/null +++ b/test/RedShirt.Example.Api.Interop.UnitTests/Tests/EventStreamListenerTests.cs @@ -0,0 +1,406 @@ +using Moq; +using System.Net; +using System.Text; + +namespace RedShirt.Example.Api.Interop.UnitTests.Tests; + +public class EventStreamListenerTests +{ + private const string BaseUrl = "https://api.example.test/"; + + private static HttpResponseMessage EmptyOkResponse() + { + return new HttpResponseMessage(HttpStatusCode.OK) + { + Content = new StringContent(string.Empty, Encoding.UTF8, "text/event-stream") + }; + } + + private static HttpResponseMessage SseOkResponse(string sseBody) + { + return new HttpResponseMessage(HttpStatusCode.OK) + { + Content = new StringContent(sseBody, Encoding.UTF8, "text/event-stream") + }; + } + + [Fact] + public void Constructor_ThrowsArgumentNullException_WhenEventEndpointPathIsNull() + { + using var httpClient = new HttpClient(); + + var exception = Assert.Throws(() => + new TestEventStreamListener(BaseUrl, httpClient, null!)); + + Assert.Equal("eventEndpointPath", exception.ParamName); + } + + [Fact] + public void Constructor_ThrowsArgumentNullException_WhenHttpClientIsNull() + { + var exception = Assert.Throws(() => + new TestEventStreamListener(BaseUrl, null!, "events/stream")); + + Assert.Equal("httpClient", exception.ParamName); + } + + [Fact] + public async Task StreamEventsAsync_ActionOverload_DeliversEventData() + { + var handler = new StubHandler(_ => SseOkResponse(SseBuilder.Event("message", "hello"))); + using var httpClient = new HttpClient(handler); + var listener = new TestEventStreamListener(BaseUrl, httpClient, "events/stream") + { + ExpectedEventName = "message" + }; + var received = new List(); + + await listener.StreamEventsAsyncAction(received.Add, TestContext.Current.CancellationToken); + + Assert.Equal(["hello"], received); + } + + [Fact] + public async Task StreamEventsAsync_CompletesWithoutCallbacks_WhenResponseContentIsNull() + { + var handler = new StubHandler(_ => new HttpResponseMessage(HttpStatusCode.OK)); + using var httpClient = new HttpClient(handler); + var listener = new TestEventStreamListener(BaseUrl, httpClient, "events/stream"); + var callbackCount = 0; + + await listener.StreamAsync( + (_, _) => + { + callbackCount++; + return Task.CompletedTask; + }, + TestContext.Current.CancellationToken); + + Assert.Equal(0, callbackCount); + } + + [Fact] + public async Task StreamEventsAsync_FiltersEvents_WhenExpectedEventNameIsSet() + { + var sseBody = SseBuilder.Events( + ("other", "ignored"), + ("message", "kept")); + var handler = new StubHandler(_ => SseOkResponse(sseBody)); + using var httpClient = new HttpClient(handler); + var listener = new TestEventStreamListener(BaseUrl, httpClient, "events/stream") + { + ExpectedEventName = "message" + }; + var received = new List(); + + await listener.StreamAsync( + (data, _) => + { + received.Add(data); + return Task.CompletedTask; + }, + TestContext.Current.CancellationToken); + + Assert.Equal(["kept"], received); + } + + [Fact] + public async Task StreamEventsAsync_FuncOverload_PassesCancellationTokenToCallback() + { + var handler = new StubHandler(_ => SseOkResponse(SseBuilder.Data("payload"))); + using var httpClient = new HttpClient(handler); + var listener = new TestEventStreamListener(BaseUrl, httpClient, "events/stream"); + CancellationToken? callbackToken = null; + + await listener.StreamAsync( + (_, cancellationToken) => + { + callbackToken = cancellationToken; + return Task.CompletedTask; + }, + TestContext.Current.CancellationToken); + + Assert.Equal(TestContext.Current.CancellationToken, callbackToken); + } + + [Fact] + public async Task StreamEventsAsync_IgnoresCommentLines() + { + var sseBody = ": keep-alive\n\n" + SseBuilder.Data("hello"); + var handler = new StubHandler(_ => SseOkResponse(sseBody)); + using var httpClient = new HttpClient(handler); + var listener = new TestEventStreamListener(BaseUrl, httpClient, "events/stream"); + var received = new List(); + + await listener.StreamAsync( + (data, _) => + { + received.Add(data); + return Task.CompletedTask; + }, + TestContext.Current.CancellationToken); + + Assert.Equal(["hello"], received); + } + + [Fact] + public async Task StreamEventsAsync_InvokesPrepareRequestAndProcessResponseHooks() + { + var prepareRequest = new Mock>(); + var processResponse = new Mock>(); + var handler = new StubHandler(_ => EmptyOkResponse()); + using var httpClient = new HttpClient(handler); + var listener = new TestEventStreamListener(BaseUrl, httpClient, "events/stream") + { + PrepareRequestHook = prepareRequest.Object, + ProcessResponseHook = processResponse.Object + }; + + await listener.StreamAsync((_, _) => Task.CompletedTask, TestContext.Current.CancellationToken); + + prepareRequest.Verify( + hook => hook(httpClient, handler.LastRequest!, handler.LastRequest!.RequestUri!.ToString()), + Times.Once); + processResponse.Verify( + hook => hook(httpClient, It.Is(response => response.StatusCode == HttpStatusCode.OK)), + Times.Once); + } + + [Fact] + public async Task StreamEventsAsync_NormalizesBaseUrlAndEventEndpointPath() + { + var handler = new StubHandler(_ => EmptyOkResponse()); + using var httpClient = new HttpClient(handler); + var listener = new TestEventStreamListener("https://api.example.test", httpClient, "/messages/event-stream"); + + await listener.StreamAsync((_, _) => Task.CompletedTask, TestContext.Current.CancellationToken); + + Assert.Equal( + "https://api.example.test/messages/event-stream", + handler.LastRequest!.RequestUri!.ToString()); + } + + [Fact] + public async Task StreamEventsAsync_ParsesMultipleEventsAndMultilineData() + { + var sseBody = + "event: message\ndata: line-one\ndata: line-two\n\n" + + SseBuilder.Event("message", "second"); + var handler = new StubHandler(_ => SseOkResponse(sseBody)); + using var httpClient = new HttpClient(handler); + var listener = new TestEventStreamListener(BaseUrl, httpClient, "events/stream") + { + ExpectedEventName = "message" + }; + var received = new List(); + + await listener.StreamAsync( + (data, _) => + { + received.Add(data); + return Task.CompletedTask; + }, + TestContext.Current.CancellationToken); + + Assert.Equal(["line-one\nline-two", "second"], received); + } + + [Fact] + public async Task StreamEventsAsync_PropagatesCancellation_BeforeRequestIsSent() + { + using var cancellationTokenSource = new CancellationTokenSource(); + await cancellationTokenSource.CancelAsync(); + var handler = new StubHandler(_ => EmptyOkResponse()); + using var httpClient = new HttpClient(handler); + var listener = new TestEventStreamListener(BaseUrl, httpClient, "events/stream"); + + await Assert.ThrowsAnyAsync(() => + listener.StreamAsync((_, _) => Task.CompletedTask, cancellationTokenSource.Token)); + + Assert.Null(handler.LastRequest); + } + + [Fact] + public async Task StreamEventsAsync_SendsGetRequest_WithExpectedUrlAndAcceptHeader() + { + var handler = new StubHandler(_ => EmptyOkResponse()); + using var httpClient = new HttpClient(handler); + var listener = new TestEventStreamListener(BaseUrl, httpClient, "/messages/event-stream"); + + await listener.StreamAsync((_, _) => Task.CompletedTask, TestContext.Current.CancellationToken); + + Assert.NotNull(handler.LastRequest); + Assert.Equal(HttpMethod.Get, handler.LastRequest!.Method); + Assert.Equal( + "https://api.example.test/messages/event-stream", + handler.LastRequest.RequestUri!.ToString()); + Assert.Contains( + handler.LastRequest.Headers.Accept, + value => value.MediaType == "text/event-stream"); + } + + [Fact] + public async Task StreamEventsAsync_ThrowsArgumentNullException_WhenActionCallbackIsNull() + { + using var httpClient = new HttpClient(new StubHandler(_ => EmptyOkResponse())); + var listener = new TestEventStreamListener(BaseUrl, httpClient, "events/stream"); + + var exception = await Assert.ThrowsAsync(() => + listener.StreamEventsAsyncAction(null!, TestContext.Current.CancellationToken)); + + Assert.Equal("onEvent", exception.ParamName); + } + + [Fact] + public async Task StreamEventsAsync_ThrowsArgumentNullException_WhenFuncCallbackIsNull() + { + using var httpClient = new HttpClient(new StubHandler(_ => EmptyOkResponse())); + var listener = new TestEventStreamListener(BaseUrl, httpClient, "events/stream"); + + var exception = await Assert.ThrowsAsync(() => + listener.StreamAsync(null!, TestContext.Current.CancellationToken)); + + Assert.Equal("onEvent", exception.ParamName); + } + + [Fact] + public async Task StreamEventsAsync_ThrowsSwaggerExceptionWithProblemDetails_On400() + { + const string problemJson = """ + { + "title": "Bad Request", + "status": 400, + "detail": "Authenticated user id is required." + } + """; + var handler = new StubHandler(_ => new HttpResponseMessage(HttpStatusCode.BadRequest) + { + Content = new StringContent(problemJson, Encoding.UTF8, "application/json") + }); + using var httpClient = new HttpClient(handler); + var listener = new TestEventStreamListener(BaseUrl, httpClient, "events/stream"); + + var exception = await Assert.ThrowsAsync>(() => + listener.StreamAsync((_, _) => Task.CompletedTask, TestContext.Current.CancellationToken)); + + Assert.Equal(400, exception.StatusCode); + Assert.Equal("Bad Request", exception.Result.Title); + Assert.Equal("Authenticated user id is required.", exception.Result.Detail); + } + + [Fact] + public async Task StreamEventsAsync_ThrowsSwaggerException_On502() + { + var handler = new StubHandler(_ => new HttpResponseMessage(HttpStatusCode.BadGateway) + { + Content = new StringContent("upstream unavailable", Encoding.UTF8, "text/plain") + }); + using var httpClient = new HttpClient(handler); + var listener = new TestEventStreamListener(BaseUrl, httpClient, "events/stream"); + + var exception = await Assert.ThrowsAsync(() => + listener.StreamAsync((_, _) => Task.CompletedTask, TestContext.Current.CancellationToken)); + + Assert.Equal(502, exception.StatusCode); + Assert.Equal("upstream unavailable", exception.Response); + } + + [Fact] + public async Task StreamEventsAsync_ThrowsSwaggerException_OnUnexpectedStatusCode() + { + var handler = new StubHandler(_ => new HttpResponseMessage(HttpStatusCode.NotFound) + { + Content = new StringContent("missing", Encoding.UTF8, "text/plain") + }); + using var httpClient = new HttpClient(handler); + var listener = new TestEventStreamListener(BaseUrl, httpClient, "events/stream"); + + var exception = await Assert.ThrowsAsync(() => + listener.StreamAsync((_, _) => Task.CompletedTask, TestContext.Current.CancellationToken)); + + Assert.Equal(404, exception.StatusCode); + Assert.Contains("was not expected (404)", exception.Message); + Assert.Equal("missing", exception.Response); + } + + [Fact] + public async Task StreamEventsAsync_UsesDefaultEventName_WhenEventLineIsMissing() + { + var handler = new StubHandler(_ => SseOkResponse(SseBuilder.Data("implicit-message"))); + using var httpClient = new HttpClient(handler); + var listener = new TestEventStreamListener(BaseUrl, httpClient, "events/stream") + { + ExpectedEventName = "event" + }; + var received = new List(); + + await listener.StreamAsync( + (data, _) => + { + received.Add(data); + return Task.CompletedTask; + }, + TestContext.Current.CancellationToken); + + Assert.Equal(["implicit-message"], received); + } + + private sealed class TestEventStreamListener : EventStreamListener + { + public string? ExpectedEventName { get; init; } + + public Action? PrepareRequestHook { get; init; } + + public Action? ProcessResponseHook { get; init; } + + public TestEventStreamListener(string baseUrl, HttpClient httpClient, string eventEndpointPath) + : base(baseUrl, httpClient, eventEndpointPath) + { + } + + protected override void PrepareRequest(HttpClient client, HttpRequestMessage request, string url) + { + PrepareRequestHook?.Invoke(client, request, url); + } + + protected override void ProcessResponse(HttpClient client, HttpResponseMessage response) + { + ProcessResponseHook?.Invoke(client, response); + } + + public Task StreamAsync( + Func onEvent, + CancellationToken cancellationToken = default) + { + return StreamEventsAsync(onEvent, cancellationToken); + } + + public Task StreamEventsAsyncAction( + Action onEvent, + CancellationToken cancellationToken = default) + { + return StreamEventsAsync(onEvent, cancellationToken); + } + + protected override Task StreamEventsCoreAsync( + Func onEvent, + CancellationToken cancellationToken) + { + return OpenEventStreamAsync(onEvent, ExpectedEventName, cancellationToken); + } + } + + private sealed class StubHandler(Func responder) : HttpMessageHandler + { + public HttpRequestMessage? LastRequest { get; private set; } + + protected override Task SendAsync( + HttpRequestMessage request, + CancellationToken cancellationToken) + { + cancellationToken.ThrowIfCancellationRequested(); + LastRequest = request; + return Task.FromResult(responder(request)); + } + } +} \ No newline at end of file diff --git a/test/RedShirt.Example.Api.Interop.UnitTests/Tests/MessagesEventStreamListenerTests.cs b/test/RedShirt.Example.Api.Interop.UnitTests/Tests/MessagesEventStreamListenerTests.cs new file mode 100644 index 0000000..64e32d1 --- /dev/null +++ b/test/RedShirt.Example.Api.Interop.UnitTests/Tests/MessagesEventStreamListenerTests.cs @@ -0,0 +1,146 @@ +using System.Net; +using System.Text; + +namespace RedShirt.Example.Api.Interop.UnitTests.Tests; + +public class MessagesEventStreamListenerTests +{ + private static HttpResponseMessage EmptyOkResponse() + { + return new HttpResponseMessage(HttpStatusCode.OK) + { + Content = new StringContent(string.Empty, Encoding.UTF8, "text/event-stream") + }; + } + + private static HttpResponseMessage SseOkResponse(string sseBody) + { + return new HttpResponseMessage(HttpStatusCode.OK) + { + Content = new StringContent(sseBody, Encoding.UTF8, "text/event-stream") + }; + } + + [Fact] + public async Task StreamEventsAsync_ActionOverload_DeliversMessageEvents() + { + var handler = new StubHandler(_ => SseOkResponse(SseBuilder.Data("hello"))); + using var httpClient = new HttpClient(handler); + var listener = new MessagesEventStreamListener("https://api.example.test/", httpClient); + var received = new List(); + + await listener.StreamEventsAsync(received.Add, TestContext.Current.CancellationToken); + + Assert.Equal(["hello"], received); + } + + [Fact] + public async Task StreamEventsAsync_AllowsCustomEventEndpointPath() + { + var handler = new StubHandler(_ => EmptyOkResponse()); + using var httpClient = new HttpClient(handler); + var listener = new MessagesEventStreamListener( + "https://api.example.test/", + httpClient, + "/custom/event-stream"); + + await listener.StreamEventsAsync((_, _) => Task.CompletedTask, TestContext.Current.CancellationToken); + + Assert.Equal( + "https://api.example.test/custom/event-stream", + handler.LastRequest!.RequestUri!.ToString()); + } + + [Fact] + public async Task StreamEventsAsync_CanBeUsedThroughInterface() + { + var handler = new StubHandler(_ => SseOkResponse(SseBuilder.Data("via-interface"))); + using var httpClient = new HttpClient(handler); + IMessagesEventStreamListener listener = + new MessagesEventStreamListener("https://api.example.test/", httpClient); + var received = new List(); + + await listener.StreamEventsAsync( + (data, _) => + { + received.Add(data); + return Task.CompletedTask; + }, + TestContext.Current.CancellationToken); + + Assert.Equal(["via-interface"], received); + } + + [Fact] + public async Task StreamEventsAsync_DeliversOnlyMessageEvents() + { + var sseBody = SseBuilder.Events( + ("heartbeat", "ignored"), + ("message", "hello")); + var handler = new StubHandler(_ => SseOkResponse(sseBody)); + using var httpClient = new HttpClient(handler); + var listener = new MessagesEventStreamListener("https://api.example.test/", httpClient); + var received = new List(); + + await listener.StreamEventsAsync( + (data, _) => + { + received.Add(data); + return Task.CompletedTask; + }, + TestContext.Current.CancellationToken); + + Assert.Equal(["hello"], received); + } + + [Fact] + public async Task StreamEventsAsync_UsesDefaultEventEndpointPath_InRequestUrl() + { + var handler = new StubHandler(_ => EmptyOkResponse()); + using var httpClient = new HttpClient(handler); + var listener = new MessagesEventStreamListener("https://api.example.test/", httpClient); + + await listener.StreamEventsAsync((_, _) => Task.CompletedTask, TestContext.Current.CancellationToken); + + Assert.Equal( + "https://api.example.test/messages/event-stream", + handler.LastRequest!.RequestUri!.ToString()); + } + + private sealed class StubHandler(Func responder) : HttpMessageHandler + { + public HttpRequestMessage? LastRequest { get; private set; } + + protected override Task SendAsync( + HttpRequestMessage request, + CancellationToken cancellationToken) + { + LastRequest = request; + return Task.FromResult(responder(request)); + } + } +} + +internal static class SseBuilder +{ + internal static string Data(string data) + { + return $"data: {data}\n\n"; + } + + internal static string Event(string eventName, string data) + { + return $"event: {eventName}\ndata: {data}\n\n"; + } + + internal static string Events(params (string EventName, string Data)[] events) + { + var builder = new StringBuilder(); + foreach (var (eventName, data) in events) + { + builder.Append(Event(eventName, data)); + } + + return builder.ToString(); + } +} \ No newline at end of file diff --git a/test/local/README.md b/test/local/README.md index 2c9b1a3..7cfb986 100644 --- a/test/local/README.md +++ b/test/local/README.md @@ -383,3 +383,48 @@ To update the client secret in SSM *and* WireMock's in-memory stubs (token bodyP ``` This only updates in-memory WireMock stubs. Restarting `wiremock-bar` restores the mapping files under `wiremock/bar/`. After the script finishes, call `GET /bar/{id}` or `POST /bar` again — the connector should 401 once with the old bearer, refresh client credentials + token, then succeed with the rotated bearer. + +## Messages (Server-Sent Events) + +The API publishes example message events to MiniStack's embedded IoT MQTT broker and streams them to clients over Server-Sent Events at `GET /messages/event-stream`. Each authenticated user receives events on the MQTT topic `example-message/user/{userId}` derived from the JWT subject. + +Local Compose resolves the MQTT broker address from IoT `DescribeEndpoint` (the same hostname AWS SDK clients use). MiniStack routes WebSocket upgrades by that IoT endpoint hostname, not by the bare gateway host `ministack:4566`; connecting without it returns HTTP `403` instead of `101 Switching Protocols`. The API connects to `ministack:4566` and sends the DescribeEndpoint value as the HTTP `Host` header. Username and password secret paths are intentionally `null` so the MQTT client connects without credentials. + +| Variable | Local default | +|-----------------------------------------------------------------------|------------------------| +| `CLIENT_EVENTS__MQTT__RESOLVE_BROKER_ADDRESS_EXTERNALLY` | `true` | +| `CLIENT_EVENTS__MQTT__AWS__BROKER_CONNECT_HOST` | unset (connect to DescribeEndpoint host) | +| `CLIENT_EVENTS__MQTT__BROKER_URL` | unset (use DescribeEndpoint) | +| `CLIENT_EVENTS__MQTT__CLIENT_ID_PREFIX` | `RedShirt.Example.Api` | +| `CLIENT_EVENTS__MQTT__USERNAME_SECRET_PATH` | `null` | +| `CLIENT_EVENTS__MQTT__PASSWORD_SECRET_PATH` | `null` | +| `CLIENT_EVENTS__MQTT__RETRY_COUNT` | `3` | + +MiniStack advertises IoT endpoint hostnames under `MINISTACK_HOST` (set to `ministack` in compose). Docker maps that hostname to the MiniStack container via a network alias in `docker-compose.yaml`; if you reset MiniStack state, run `aws iot describe-endpoint` and update the alias to match. MiniStack also exposes native MQTT over TLS on port `8883` when needed; the compose file maps that port for host-side tooling. + +### Scripts + +Mock Python scripts under `scripts/messages/` exercise the event flow locally. Run them from the `test/local` directory: + +```bash +chmod +x ./scripts/messages/*.py # once +export API_JWT_TOKEN="$(./get-bearer-token.py)" +``` + +In one terminal, listen for events: + +```bash +./scripts/messages/listen-messages.py +``` + +In another terminal, publish a message (connected clients for the same user should receive it immediately): + +```bash +./scripts/messages/send-message.py "Hello over SSE" +``` + +Notes: + +* Both endpoints require a bearer token with `api:write` for `POST /messages` and `api:read` for `GET /messages/event-stream`. +* Events are scoped to the authenticated user's id (`sub` / name identifier claim). +* If MQTT connectivity fails, the API returns **502 Bad Gateway** with a problem-details body. diff --git a/test/local/docker-compose.yaml b/test/local/docker-compose.yaml index 04f9c3a..8d57d92 100644 --- a/test/local/docker-compose.yaml +++ b/test/local/docker-compose.yaml @@ -3,6 +3,14 @@ services: image: ministackorg/ministack:latest ports: - 4566:4566 + - 8883:8883 + environment: + MINISTACK_HOST: ministack + networks: + default: + aliases: + # Must match `aws iot describe-endpoint` (account-specific prefix persists in MiniStack state). + - f7b11509f4d675-ats.iot.us-east-1.ministack redis: image: redis:latest @@ -171,3 +179,14 @@ services: UPLOADS__BUCKET_VERIFIED_ITEMS: "${UPLOADS__BUCKET_VERIFIED_ITEMS-verified-uploads}" UPLOADS__MAX_UPLOAD_SIZE_BYTES: "${UPLOADS__MAX_UPLOAD_SIZE_BYTES-}" UPLOADS__PRESIGNED_URL_LIFETIME_MINUTES: "${UPLOADS__PRESIGNED_URL_LIFETIME_MINUTES-15}" + # Client events (MiniStack IoT MQTT broker) + ## + CLIENT_EVENTS__MQTT__RESOLVE_BROKER_ADDRESS_EXTERNALLY: "${CLIENT_EVENTS__MQTT__RESOLVE_BROKER_ADDRESS_EXTERNALLY-true}" + CLIENT_EVENTS__MQTT__AWS__BROKER_CONNECT_HOST: "${CLIENT_EVENTS__MQTT__AWS__BROKER_CONNECT_HOST-}" + CLIENT_EVENTS__MQTT__BROKER_URL: "${CLIENT_EVENTS__MQTT__BROKER_URL-}" + CLIENT_EVENTS__MQTT__CLIENT_ID_PREFIX: "${CLIENT_EVENTS__MQTT__CLIENT_ID_PREFIX-RedShirt.Example.Api}" + # Set specific protocol version + CLIENT_EVENTS__MQTT__PROTOCOL_VERSION: "${CLIENT_EVENTS__MQTT__PROTOCOL_VERSION-V311}" + CLIENT_EVENTS__MQTT__USERNAME_SECRET_PATH: null + CLIENT_EVENTS__MQTT__PASSWORD_SECRET_PATH: null + CLIENT_EVENTS__MQTT__RETRY_COUNT: "${CLIENT_EVENTS__MQTT__RETRY_COUNT-3}" diff --git a/test/local/scripts/messages/listen-messages.py b/test/local/scripts/messages/listen-messages.py new file mode 100755 index 0000000..c70af0c --- /dev/null +++ b/test/local/scripts/messages/listen-messages.py @@ -0,0 +1,86 @@ +#!/usr/bin/env python3 +"""Listen to GET /messages/event-stream and print each example-message event.""" + +from __future__ import annotations + +import sys +import urllib.error +import urllib.request + +from messages_script_common import ( + create_parser, + get_api_base_url, + require_api_jwt_token, +) + + +def parse_sse_stream(response: urllib.response.addinfourl) -> None: + event_name = None + data_lines: list[str] = [] + + while True: + line = response.readline() + if not line: + break + + text = line.decode("utf-8", errors="replace").rstrip("\r\n") + if not text: + if data_lines: + payload = "\n".join(data_lines) + label = event_name or "message" + print(f"[{label}] {payload}") + sys.stdout.flush() + event_name = None + data_lines = [] + continue + + if text.startswith(":"): + continue + + if text.startswith("event:"): + event_name = text[len("event:") :].strip() + continue + + if text.startswith("data:"): + data_lines.append(text[len("data:") :].strip()) + continue + + +def main() -> int: + parser = create_parser( + "Listen to GET /messages/event-stream and print each example-message event" + ) + args = parser.parse_args() + _ = args + + token = require_api_jwt_token() + base_url = get_api_base_url() + url = f"{base_url.rstrip('/')}/messages/event-stream" + + request = urllib.request.Request( + url, + method="GET", + headers={ + "Authorization": f"Bearer {token}", + "Accept": "text/event-stream", + }, + ) + + print(f"Listening on {url} (Ctrl+C to stop)...", flush=True) + + try: + with urllib.request.urlopen(request, timeout=None) as response: + parse_sse_stream(response) + except KeyboardInterrupt: + print("\nStopped.", flush=True) + return 0 + except urllib.error.HTTPError as error: + body_text = error.read().decode("utf-8", errors="replace") + print(f"HTTP {error.code}: {body_text}", file=sys.stderr) + return 1 + + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/test/local/scripts/messages/messages_script_common.py b/test/local/scripts/messages/messages_script_common.py new file mode 100755 index 0000000..c529293 --- /dev/null +++ b/test/local/scripts/messages/messages_script_common.py @@ -0,0 +1,39 @@ +"""Shared helpers for local message event-stream test scripts.""" + +from __future__ import annotations + +import argparse +import os + +DEFAULT_API_BASE = "http://localhost:9000" +API_JWT_TOKEN_ENV = "API_JWT_TOKEN" +API_BASE_URL_ENV = "API_BASE_URL" + +ENV_HELP_EPILOG = f"""\ +environment variables: + {API_JWT_TOKEN_ENV} Bearer access token (required). Populate from get-bearer-token.py, e.g.: + export {API_JWT_TOKEN_ENV}="$(./test/local/get-bearer-token.py)" + {API_BASE_URL_ENV} API base URL (default: {DEFAULT_API_BASE}) +""" + + +def create_parser(description: str) -> argparse.ArgumentParser: + return argparse.ArgumentParser( + description=description, + epilog=ENV_HELP_EPILOG, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + + +def require_api_jwt_token() -> str: + token = os.environ.get(API_JWT_TOKEN_ENV) + if not token: + raise SystemExit( + f"Set {API_JWT_TOKEN_ENV} to a bearer access token " + f'(see --help; e.g. export {API_JWT_TOKEN_ENV}="$(./test/local/get-bearer-token.py)").' + ) + return token + + +def get_api_base_url() -> str: + return os.environ.get(API_BASE_URL_ENV, DEFAULT_API_BASE) diff --git a/test/local/scripts/messages/send-message.py b/test/local/scripts/messages/send-message.py new file mode 100755 index 0000000..ec0bede --- /dev/null +++ b/test/local/scripts/messages/send-message.py @@ -0,0 +1,57 @@ +#!/usr/bin/env python3 +"""Publish an example message event via POST /messages.""" + +from __future__ import annotations + +import json +import sys +import urllib.error +import urllib.request + +from messages_script_common import ( + create_parser, + get_api_base_url, + require_api_jwt_token, +) + + +def main() -> int: + parser = create_parser("Publish an example message event via POST /messages") + parser.add_argument( + "message", + nargs="?", + default="Hello from local test script", + help="Message text to publish (default: Hello from local test script)", + ) + args = parser.parse_args() + + token = require_api_jwt_token() + base_url = get_api_base_url() + url = f"{base_url.rstrip('/')}/messages" + body = json.dumps({"message": args.message}).encode("utf-8") + + request = urllib.request.Request( + url, + data=body, + method="POST", + headers={ + "Authorization": f"Bearer {token}", + "Accept": "application/json", + "Content-Type": "application/json", + }, + ) + + try: + with urllib.request.urlopen(request, timeout=30) as response: + print(f"HTTP {response.status} {response.reason}") + except urllib.error.HTTPError as error: + body_text = error.read().decode("utf-8", errors="replace") + print(f"HTTP {error.code}: {body_text}", file=sys.stderr) + return 1 + + print(f"Published message for the authenticated user: {args.message!r}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())