Compare commits
3 Commits
6183ea613f
...
868c447a6c
| Author | SHA1 | Date | |
|---|---|---|---|
| 868c447a6c | |||
| 220d3f441f | |||
| 5d9197f54a |
@@ -21,8 +21,9 @@
|
||||
</PropertyGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="8.0.1" />
|
||||
<PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="10.0.10" />
|
||||
<PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="10.0.10" />
|
||||
<PackageReference Include="Microsoft.Extensions.Logging" Version="10.0.10" />
|
||||
<PackageReference Include="RabbitMQ.Client" Version="7.2.1" />
|
||||
</ItemGroup>
|
||||
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
using DigitalData.MessagingService.Publisher.Abstraction;
|
||||
using DigitalData.MessagingService.RabbitMQ;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using System;
|
||||
|
||||
namespace DigitalData.MessagingService.Client.DependencyInjection;
|
||||
@@ -13,14 +14,14 @@ namespace DigitalData.MessagingService.Client.DependencyInjection;
|
||||
/// <remarks>
|
||||
/// This class manages its own internal <see cref="IServiceProvider"/> using a
|
||||
/// <see cref="Lazy{T}"/> pattern so the DI container is only built once,
|
||||
/// on the first call to <see cref="ConnectRabbitMq"/>.
|
||||
/// on the first call to <see cref="ConnectRabbitMq(Action{RabbitMqConfiguration}, OnReconnect)"/>.
|
||||
/// </remarks>
|
||||
public static class EmailSender
|
||||
{
|
||||
/// <summary>
|
||||
/// Internal event used to accumulate service registrations before the
|
||||
/// <see cref="IServiceProvider"/> is built. Handlers are added by
|
||||
/// <see cref="ConnectRabbitMq"/> and invoked exactly once during
|
||||
/// <see cref="ConnectRabbitMq(Action{RabbitMqConfiguration}, OnReconnect)"/> and invoked exactly once during
|
||||
/// lazy initialization.
|
||||
/// </summary>
|
||||
private static event Action<IServiceCollection> ConfigureServices = delegate { };
|
||||
@@ -33,6 +34,7 @@ public static class EmailSender
|
||||
private static readonly Lazy<IServiceProvider> LazyProvider = new(() =>
|
||||
{
|
||||
var services = new ServiceCollection();
|
||||
services.AddLogging();
|
||||
ConfigureServices?.Invoke(services);
|
||||
return services.BuildServiceProvider();
|
||||
});
|
||||
@@ -42,7 +44,7 @@ public static class EmailSender
|
||||
/// and the internal <see cref="IServiceProvider"/> has been initialized.
|
||||
/// </summary>
|
||||
/// <value>
|
||||
/// <see langword="true"/> if <see cref="ConnectRabbitMq"/> has been called
|
||||
/// <see langword="true"/> if <see cref="ConnectRabbitMq(Action{RabbitMqConfiguration}, OnReconnect)"/> has been called
|
||||
/// and the provider is built; otherwise <see langword="false"/>.
|
||||
/// </value>
|
||||
public static bool IsConnected => LazyProvider.IsValueCreated;
|
||||
@@ -110,7 +112,7 @@ public static class EmailSender
|
||||
/// </summary>
|
||||
/// <param name="email">The outgoing email event to enqueue.</param>
|
||||
/// <exception cref="InvalidOperationException">
|
||||
/// Thrown when <see cref="ConnectRabbitMq"/> has not been called prior to sending.
|
||||
/// Thrown when <see cref="ConnectRabbitMq(Action{RabbitMqConfiguration}, OnReconnect)"/> has not been called prior to sending.
|
||||
/// </exception>
|
||||
/// <remarks>
|
||||
/// This method resolves <see cref="IOutgoingEmailPublisher"/> from the internal
|
||||
|
||||
@@ -0,0 +1,187 @@
|
||||
# DigitalData.MessagingService.Client
|
||||
|
||||
Ein schlanker, eigenständiger RabbitMQ-Client zum Versenden von E-Mails über den DigitalData MessagingService – ohne eigenen DI-Container oder Hosting-Infrastruktur.
|
||||
|
||||
## Installation
|
||||
|
||||
```
|
||||
dotnet add package DigitalData.MessagingService.Client
|
||||
```
|
||||
|
||||
## Schnellstart
|
||||
|
||||
### 1. Verbindung herstellen
|
||||
|
||||
Einmalig beim Anwendungsstart aufrufen – typischerweise in `Program.cs`, `Application_Start` oder dem Konstruktor des Einstiegspunkts:
|
||||
|
||||
```csharp
|
||||
EmailSender.ConnectRabbitMq(
|
||||
url: "amqp://172.24.12.56:5672",
|
||||
username: "admin",
|
||||
password: "geheimespasswort"
|
||||
);
|
||||
```
|
||||
|
||||
```vb
|
||||
EmailSender.ConnectRabbitMq(
|
||||
url:="amqp://172.24.12.56:5672",
|
||||
username:="admin",
|
||||
password:="geheimespasswort"
|
||||
)
|
||||
```
|
||||
|
||||
Mit virtuellem Host:
|
||||
|
||||
```csharp
|
||||
EmailSender.ConnectRabbitMq(
|
||||
url: "amqp://172.24.12.56:5672/meinvhost",
|
||||
username: "admin",
|
||||
password: "geheimespasswort"
|
||||
);
|
||||
```
|
||||
|
||||
```vb
|
||||
EmailSender.ConnectRabbitMq(
|
||||
url:="amqp://172.24.12.56:5672/meinvhost",
|
||||
username:="admin",
|
||||
password:="geheimespasswort"
|
||||
)
|
||||
```
|
||||
|
||||
> **Hinweis:** `ConnectRabbitMq` darf pro Prozess nur einmal erfolgreich aufgerufen werden.
|
||||
> Ein erneuter Aufruf löst standardmäßig eine `InvalidOperationException` aus.
|
||||
> Ist dieses Verhalten nicht erwünscht, kann `OnReconnect.Ignore` übergeben werden:
|
||||
>
|
||||
> ```csharp
|
||||
> EmailSender.ConnectRabbitMq("amqp://172.24.12.56:5672", "admin", "geheimespasswort", OnReconnect.Ignore);
|
||||
> ```
|
||||
>
|
||||
> ```vb
|
||||
> EmailSender.ConnectRabbitMq("amqp://172.24.12.56:5672", "admin", "geheimespasswort", OnReconnect.Ignore)
|
||||
> ```
|
||||
|
||||
### 2. E-Mail versenden
|
||||
|
||||
```csharp
|
||||
EmailSender.Send(new OutgoingEmailEvent
|
||||
{
|
||||
Id = Guid.NewGuid(),
|
||||
Recipient = "empfaenger@beispiel.de",
|
||||
Subject = "Willkommen",
|
||||
Body = "<p>Hallo Welt!</p>",
|
||||
IsHtml = true,
|
||||
QueuedAt = DateTime.Now
|
||||
});
|
||||
```
|
||||
|
||||
```vb
|
||||
EmailSender.Send(New OutgoingEmailEvent With {
|
||||
.Id = Guid.NewGuid(),
|
||||
.Recipient = "empfaenger@beispiel.de",
|
||||
.Subject = "Willkommen",
|
||||
.Body = "<p>Hallo Welt!</p>",
|
||||
.IsHtml = True,
|
||||
.QueuedAt = DateTime.Now
|
||||
})
|
||||
```
|
||||
|
||||
`Send` ist eine **Fire-and-Forget**-Methode: die Nachricht wird in die RabbitMQ-Queue eingereiht und der aufrufende Code wartet nicht auf die eigentliche Zustellung.
|
||||
|
||||
### 3. Verbindungsstatus prüfen
|
||||
|
||||
```csharp
|
||||
if (EmailSender.IsConnected)
|
||||
{
|
||||
// Verbindung wurde bereits hergestellt
|
||||
}
|
||||
```
|
||||
|
||||
```vb
|
||||
If EmailSender.IsConnected Then
|
||||
' Verbindung wurde bereits hergestellt
|
||||
End If
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## URL-Format
|
||||
|
||||
```
|
||||
amqp://<host>:<port>[/<virtualhost>]
|
||||
```
|
||||
|
||||
| Bestandteil | Beschreibung | Beispiel |
|
||||
|---------------|---------------------------------------------------------|----------------|
|
||||
| `host` | Hostname oder IP-Adresse des RabbitMQ-Servers | `172.24.12.56` |
|
||||
| `port` | AMQP-Port (Standard: `5672`) | `5672` |
|
||||
| `virtualhost` | Optionaler virtueller Host; URL-Encoding wird aufgelöst | `meinvhost` |
|
||||
|
||||
Benutzername und Passwort werden **nicht** aus der URL gelesen, sondern immer separat als Parameter übergeben. Dadurch werden Klartext-Credentials in URLs vermieden.
|
||||
|
||||
---
|
||||
|
||||
## RabbitMQ-Server
|
||||
|
||||
| | |
|
||||
|---|---|
|
||||
| **AMQP** | `amqp://172.24.12.56:5672` |
|
||||
| **Management UI** | http://172.24.12.56:15672 (Browser) |
|
||||
| **Benutzername** | `admin` |
|
||||
| **Passwort** | Im RDM-Eintrag **`sDD-VMP05-VM06 - 172.24.12.56 - RabbitMQ`** hinterlegt |
|
||||
|
||||
---
|
||||
|
||||
## Voraussetzungen
|
||||
|
||||
- .NET Framework 4.6.2 / 4.8 oder .NET 8+
|
||||
- Erreichbarer RabbitMQ-Server
|
||||
- Exchange, Queue und Routing Key müssen auf dem Broker vorhanden sein (werden vom Server-seitigen Consumer angelegt)
|
||||
|
||||
---
|
||||
|
||||
## Erweiterte Konfiguration
|
||||
|
||||
Für spezielle Szenarien – etwa abweichende Exchange- oder Queue-Namen – steht ein Delegate-basierter Überload zur Verfügung:
|
||||
|
||||
```csharp
|
||||
EmailSender.ConnectRabbitMq(cfg =>
|
||||
{
|
||||
cfg.HostName = "172.24.12.56";
|
||||
cfg.Port = 5672;
|
||||
cfg.UserName = "admin";
|
||||
cfg.Password = "geheimespasswort";
|
||||
cfg.VirtualHost = "/";
|
||||
// cfg.ExchangeName, cfg.QueueName, cfg.RoutingKey usw. bei Bedarf anpassen
|
||||
});
|
||||
```
|
||||
|
||||
```vb
|
||||
EmailSender.ConnectRabbitMq(Sub(cfg)
|
||||
cfg.HostName = "172.24.12.56"
|
||||
cfg.Port = 5672
|
||||
cfg.UserName = "admin"
|
||||
cfg.Password = "geheimespasswort"
|
||||
cfg.VirtualHost = "/"
|
||||
' cfg.ExchangeName, cfg.QueueName, cfg.RoutingKey usw. bei Bedarf anpassen
|
||||
End Sub)
|
||||
```
|
||||
|
||||
Dieser Überload ist für den Normalbetrieb nicht erforderlich.
|
||||
|
||||
---
|
||||
|
||||
## Fehlerbehandlung
|
||||
|
||||
| Situation | Verhalten |
|
||||
|---|---|
|
||||
| `ConnectRabbitMq` noch nicht aufgerufen, dann `Send` | `InvalidOperationException` |
|
||||
| `ConnectRabbitMq` erneut aufgerufen (Standard) | `InvalidOperationException` |
|
||||
| `ConnectRabbitMq` erneut aufgerufen mit `OnReconnect.Ignore` | Wird stillschweigend ignoriert |
|
||||
| RabbitMQ nicht erreichbar beim ersten `Send` | Exception aus dem RabbitMQ-Client |
|
||||
|
||||
---
|
||||
|
||||
## Lizenz
|
||||
|
||||
Copyright © 2026 Digital Data GmbH. Alle Rechte vorbehalten.
|
||||
|
||||
@@ -11,11 +11,21 @@
|
||||
|
||||
<ItemGroup>
|
||||
<PackageReference Include="coverlet.collector" Version="6.0.0" />
|
||||
<PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="10.0.10" />
|
||||
<PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="10.0.10" />
|
||||
<PackageReference Include="Microsoft.Extensions.Logging" Version="10.0.10" />
|
||||
<PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.8.0" />
|
||||
<PackageReference Include="RabbitMQ.Client" Version="7.2.1" />
|
||||
<PackageReference Include="xunit" Version="2.5.3" />
|
||||
<PackageReference Include="xunit.runner.visualstudio" Version="2.5.3" />
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="..\..\src\infrastructure\DigitalData.MessagingService.Publisher\DigitalData.MessagingService.Publisher.csproj" />
|
||||
<ProjectReference Include="..\..\src\infrastructure\DigitalData.MessagingService.RabbitMQ\DigitalData.MessagingService.RabbitMQ.csproj" />
|
||||
<ProjectReference Include="..\..\src\presentation\DigitalData.MessagingService.Client.DependencyInjection\DigitalData.MessagingService.Client.DependencyInjection.csproj" />
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<Using Include="Xunit" />
|
||||
</ItemGroup>
|
||||
|
||||
@@ -0,0 +1,110 @@
|
||||
using DigitalData.MessagingService.Client.DependencyInjection;
|
||||
using DigitalData.MessagingService.Publisher.Abstraction;
|
||||
|
||||
namespace DigitalData.MessagingService.Tests.Integration;
|
||||
|
||||
/// <summary>
|
||||
/// xUnit collection that serializes all <see cref="EmailSenderTests"/> so they share
|
||||
/// the same process-wide static state of <see cref="EmailSender.LazyProvider"/>.
|
||||
/// </summary>
|
||||
[CollectionDefinition(Name)]
|
||||
public sealed class EmailSenderCollection : ICollectionFixture<EmailSenderFixture>
|
||||
{
|
||||
public const string Name = "EmailSender";
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Fixture that connects <see cref="EmailSender"/> once before all tests in the collection run.
|
||||
/// </summary>
|
||||
public sealed class EmailSenderFixture
|
||||
{
|
||||
public EmailSenderFixture()
|
||||
{
|
||||
// EmailSender is a static class with a Lazy<IServiceProvider>.
|
||||
// ConnectRabbitMq can only be called successfully once per process.
|
||||
if (!EmailSender.IsConnected)
|
||||
{
|
||||
EmailSender.ConnectRabbitMq(cfg =>
|
||||
{
|
||||
cfg.HostName = RabbitMqTestConfig.HostName;
|
||||
cfg.Port = RabbitMqTestConfig.Port;
|
||||
cfg.UserName = RabbitMqTestConfig.UserName;
|
||||
cfg.Password = RabbitMqTestConfig.Password;
|
||||
cfg.VirtualHost = RabbitMqTestConfig.VirtualHost;
|
||||
cfg.QueueName = RabbitMqTestConfig.QueueName;
|
||||
cfg.ExchangeName = RabbitMqTestConfig.ExchangeName;
|
||||
cfg.RoutingKey = RabbitMqTestConfig.RoutingKey;
|
||||
cfg.DlqQueueName = RabbitMqTestConfig.DlqQueueName;
|
||||
cfg.DlqExchangeName = RabbitMqTestConfig.DlqExchangeName;
|
||||
cfg.DlqRoutingKey = RabbitMqTestConfig.DlqRoutingKey;
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Integration tests for the <see cref="EmailSender"/> static client.
|
||||
/// All tests run inside <see cref="EmailSenderCollection"/> to share the single
|
||||
/// static connection established by <see cref="EmailSenderFixture"/>.
|
||||
/// </summary>
|
||||
[Collection(EmailSenderCollection.Name)]
|
||||
public sealed class EmailSenderTests
|
||||
{
|
||||
[Fact]
|
||||
public void IsConnected_AfterConnectRabbitMq_ReturnsTrue()
|
||||
{
|
||||
Assert.True(EmailSender.IsConnected);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void ConnectRabbitMq_WhenAlreadyConnected_WithThrowException_ThrowsInvalidOperationException()
|
||||
{
|
||||
Assert.Throws<InvalidOperationException>(() =>
|
||||
EmailSender.ConnectRabbitMq(cfg => { }, OnReconnect.ThrowException));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void ConnectRabbitMq_WhenAlreadyConnected_WithIgnore_DoesNotThrow()
|
||||
{
|
||||
var exception = Record.Exception(() =>
|
||||
EmailSender.ConnectRabbitMq(cfg => { }, OnReconnect.Ignore));
|
||||
|
||||
Assert.Null(exception);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Send_WithValidEmail_DoesNotThrow()
|
||||
{
|
||||
var email = new OutgoingEmailEvent
|
||||
{
|
||||
Id = Guid.NewGuid(),
|
||||
Recipient = "hakanttek@gmail.com",
|
||||
Subject = "EmailSender.Send Integration Test",
|
||||
Body = "<p>Sent via EmailSender static client.</p>",
|
||||
IsHtml = true,
|
||||
QueuedAt = DateTime.Now
|
||||
};
|
||||
|
||||
var exception = Record.Exception(() => EmailSender.Send(email));
|
||||
|
||||
Assert.Null(exception);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Send_WithPlainTextBody_DoesNotThrow()
|
||||
{
|
||||
var email = new OutgoingEmailEvent
|
||||
{
|
||||
Id = Guid.NewGuid(),
|
||||
Recipient = "hakanttek@gmail.com",
|
||||
Subject = "Plain Text Test",
|
||||
Body = "This is a plain text email.",
|
||||
IsHtml = false,
|
||||
QueuedAt = DateTime.Now
|
||||
};
|
||||
|
||||
var exception = Record.Exception(() => EmailSender.Send(email));
|
||||
|
||||
Assert.Null(exception);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,74 @@
|
||||
using DigitalData.MessagingService.Client.DependencyInjection;
|
||||
using DigitalData.MessagingService.Publisher.Abstraction;
|
||||
|
||||
namespace DigitalData.MessagingService.Tests.Integration;
|
||||
|
||||
/// <summary>
|
||||
/// Integration tests for the URL-based <see cref="EmailSender.ConnectRabbitMq(string, string, string, OnReconnect)"/>
|
||||
/// overload. All tests run inside <see cref="EmailSenderCollection"/> so they share the already-established
|
||||
/// static connection. The URL overload is exercised via <see cref="OnReconnect.Ignore"/> and
|
||||
/// <see cref="OnReconnect.ThrowException"/> to verify its delegation and guard behavior.
|
||||
/// </summary>
|
||||
[Collection(EmailSenderCollection.Name)]
|
||||
public sealed class EmailSenderUrlOverloadTests
|
||||
{
|
||||
private static readonly string ValidUrl = $"amqp://{RabbitMqTestConfig.HostName}:{RabbitMqTestConfig.Port}";
|
||||
private static readonly string ValidUrlWithVHost = $"amqp://{RabbitMqTestConfig.HostName}:{RabbitMqTestConfig.Port}/myvhost";
|
||||
|
||||
[Fact]
|
||||
public void ConnectRabbitMq_UrlOverload_WhenAlreadyConnected_WithIgnore_DoesNotThrow()
|
||||
{
|
||||
var exception = Record.Exception(() =>
|
||||
EmailSender.ConnectRabbitMq(
|
||||
ValidUrl,
|
||||
RabbitMqTestConfig.UserName,
|
||||
RabbitMqTestConfig.Password,
|
||||
OnReconnect.Ignore));
|
||||
|
||||
Assert.Null(exception);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void ConnectRabbitMq_UrlOverload_WhenAlreadyConnected_WithThrowException_Throws()
|
||||
{
|
||||
Assert.Throws<InvalidOperationException>(() =>
|
||||
EmailSender.ConnectRabbitMq(
|
||||
ValidUrl,
|
||||
RabbitMqTestConfig.UserName,
|
||||
RabbitMqTestConfig.Password,
|
||||
OnReconnect.ThrowException));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void ConnectRabbitMq_UrlOverload_WithVHostInPath_WithIgnore_DoesNotThrow()
|
||||
{
|
||||
var exception = Record.Exception(() =>
|
||||
EmailSender.ConnectRabbitMq(
|
||||
ValidUrlWithVHost,
|
||||
RabbitMqTestConfig.UserName,
|
||||
RabbitMqTestConfig.Password,
|
||||
OnReconnect.Ignore));
|
||||
|
||||
Assert.Null(exception);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Send_AfterUrlOverloadConnection_DoesNotThrow()
|
||||
{
|
||||
var email = new OutgoingEmailEvent
|
||||
{
|
||||
Id = Guid.NewGuid(),
|
||||
Recipient = "url-overload-test@example.com",
|
||||
Subject = "URL Overload Integration Test",
|
||||
Body = "<p>Sent after URL-based connection.</p>",
|
||||
IsHtml = true,
|
||||
QueuedAt = DateTime.Now
|
||||
};
|
||||
|
||||
// Connection was established via Action<> overload in the fixture;
|
||||
// Send should work regardless of which overload was used to connect.
|
||||
var exception = Record.Exception(() => EmailSender.Send(email));
|
||||
|
||||
Assert.Null(exception);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,195 @@
|
||||
using System.Text;
|
||||
using System.Text.Json;
|
||||
using DigitalData.MessagingService.Publisher;
|
||||
using DigitalData.MessagingService.Publisher.Abstraction;
|
||||
using DigitalData.MessagingService.RabbitMQ;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using RabbitMQ.Client;
|
||||
|
||||
namespace DigitalData.MessagingService.Tests.Integration;
|
||||
|
||||
/// <summary>
|
||||
/// Integration tests for <see cref="OutgoingEmailPublisher"/> against the real RabbitMQ broker.
|
||||
/// Each test publishes a message and immediately reads it back via BasicGetAsync to verify
|
||||
/// the full round-trip without starting the consumer (which requires Limilabs Mail.dll).
|
||||
/// </summary>
|
||||
public sealed class OutgoingEmailPublisherTests : IAsyncDisposable
|
||||
{
|
||||
private readonly ServiceProvider _serviceProvider;
|
||||
private readonly IOutgoingEmailPublisher _publisher;
|
||||
private readonly RabbitMqConnectionFactory _factory;
|
||||
|
||||
public OutgoingEmailPublisherTests()
|
||||
{
|
||||
var services = new ServiceCollection();
|
||||
services.AddLogging();
|
||||
services.AddMessagingServicePublisher(RabbitMqTestConfig.Apply);
|
||||
|
||||
_serviceProvider = services.BuildServiceProvider();
|
||||
_publisher = _serviceProvider.GetRequiredService<IOutgoingEmailPublisher>();
|
||||
_factory = _serviceProvider.GetRequiredService<RabbitMqConnectionFactory>();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task EnqueueAsync_PublishesMessage_MessageArrivesInQueue()
|
||||
{
|
||||
var email = new OutgoingEmailEvent
|
||||
{
|
||||
Id = Guid.NewGuid(),
|
||||
Recipient = "test@example.com",
|
||||
Subject = "Integration Test - EnqueueAsync",
|
||||
Body = "<p>Hello from integration test.</p>",
|
||||
IsHtml = true,
|
||||
QueuedAt = DateTime.Now
|
||||
};
|
||||
|
||||
var depthBefore = await _publisher.GetQueueDepthAsync();
|
||||
|
||||
await _publisher.EnqueueAsync(email);
|
||||
|
||||
await Task.Delay(300);
|
||||
|
||||
var depthAfter = await _publisher.GetQueueDepthAsync();
|
||||
|
||||
Assert.True(depthAfter >= depthBefore + 1,
|
||||
$"Expected queue depth to increase by at least 1. Before: {depthBefore}, After: {depthAfter}.");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task EnqueueAsync_MultipleMessages_AllArrivesInQueue()
|
||||
{
|
||||
var emails = Enumerable.Range(1, 3).Select(i => new OutgoingEmailEvent
|
||||
{
|
||||
Id = Guid.NewGuid(),
|
||||
Recipient = $"recipient{i}@example.com",
|
||||
Subject = $"Integration Test - Batch #{i}",
|
||||
Body = $"Batch message {i}",
|
||||
IsHtml = false,
|
||||
QueuedAt = DateTime.Now
|
||||
}).ToList();
|
||||
|
||||
foreach (var email in emails)
|
||||
await _publisher.EnqueueAsync(email);
|
||||
|
||||
await Task.Delay(500);
|
||||
|
||||
// Verify at least one message is present
|
||||
var received = await PeekMessageAsync();
|
||||
Assert.NotNull(received);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task GetQueueDepthAsync_AfterPublish_ReturnsPositiveDepth()
|
||||
{
|
||||
var email = new OutgoingEmailEvent
|
||||
{
|
||||
Id = Guid.NewGuid(),
|
||||
Recipient = "depth-test@example.com",
|
||||
Subject = "Integration Test - GetQueueDepth",
|
||||
Body = "Queue depth test",
|
||||
IsHtml = false,
|
||||
QueuedAt = DateTime.Now
|
||||
};
|
||||
|
||||
await _publisher.EnqueueAsync(email);
|
||||
await Task.Delay(300);
|
||||
|
||||
var depth = await _publisher.GetQueueDepthAsync();
|
||||
|
||||
Assert.True(depth > 0, $"Expected queue depth > 0, but got {depth}.");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task EnqueueAsync_SerializesAllFields_DeserializesCorrectly()
|
||||
{
|
||||
var id = Guid.NewGuid();
|
||||
|
||||
var email = new OutgoingEmailEvent
|
||||
{
|
||||
Id = id,
|
||||
Recipient = "serialize@example.com",
|
||||
Subject = "Serialization Test",
|
||||
Body = "<strong>Bold</strong>",
|
||||
IsHtml = true,
|
||||
QueuedAt = DateTime.Now
|
||||
};
|
||||
|
||||
await _publisher.EnqueueAsync(email);
|
||||
await Task.Delay(300);
|
||||
|
||||
// Read messages until we find the one we just published
|
||||
var received = await FindMessageAsync(id);
|
||||
|
||||
Assert.NotNull(received);
|
||||
Assert.Equal(id, received.Id);
|
||||
Assert.Equal("serialize@example.com", received.Recipient);
|
||||
Assert.Equal("Serialization Test", received.Subject);
|
||||
Assert.Equal("<strong>Bold</strong>", received.Body);
|
||||
Assert.True(received.IsHtml);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Reads a single message from the queue without acknowledging it (peek via nack+requeue).
|
||||
/// </summary>
|
||||
private async Task<OutgoingEmailEvent?> PeekMessageAsync()
|
||||
{
|
||||
var connection = await _factory.GetDefaultConnectionAsync();
|
||||
await using var channel = await connection.CreateChannelAsync();
|
||||
|
||||
var result = await channel.BasicGetAsync(RabbitMqTestConfig.QueueName, autoAck: false);
|
||||
|
||||
if (result is null)
|
||||
return null;
|
||||
|
||||
// Nack with requeue=true so the message stays in the queue for consumer
|
||||
await channel.BasicNackAsync(result.DeliveryTag, multiple: false, requeue: true);
|
||||
|
||||
var json = Encoding.UTF8.GetString(result.Body.ToArray());
|
||||
return JsonSerializer.Deserialize<OutgoingEmailEvent>(json);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Scans queue messages (up to a limit) to find a message matching the given <paramref name="id"/>.
|
||||
/// All messages are re-queued after inspection.
|
||||
/// </summary>
|
||||
private async Task<OutgoingEmailEvent?> FindMessageAsync(Guid id, int maxMessages = 50)
|
||||
{
|
||||
var connection = await _factory.GetDefaultConnectionAsync();
|
||||
await using var channel = await connection.CreateChannelAsync();
|
||||
|
||||
var requeue = new List<(ulong DeliveryTag, byte[] Body)>();
|
||||
|
||||
OutgoingEmailEvent? found = null;
|
||||
|
||||
for (int i = 0; i < maxMessages; i++)
|
||||
{
|
||||
var result = await channel.BasicGetAsync(RabbitMqTestConfig.QueueName, autoAck: false);
|
||||
if (result is null)
|
||||
break;
|
||||
|
||||
requeue.Add((result.DeliveryTag, result.Body.ToArray()));
|
||||
|
||||
var json = Encoding.UTF8.GetString(result.Body.ToArray());
|
||||
var evt = JsonSerializer.Deserialize<OutgoingEmailEvent>(json);
|
||||
|
||||
if (evt?.Id == id)
|
||||
{
|
||||
found = evt;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
// Re-queue all inspected messages so the consumer can still process them
|
||||
foreach (var (tag, _) in requeue)
|
||||
await channel.BasicNackAsync(tag, multiple: false, requeue: true);
|
||||
|
||||
return found;
|
||||
}
|
||||
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
await _factory.DisposeAsync();
|
||||
await _serviceProvider.DisposeAsync();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,39 @@
|
||||
using DigitalData.MessagingService.RabbitMQ;
|
||||
|
||||
namespace DigitalData.MessagingService.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// Shared RabbitMQ connection settings used across integration tests.
|
||||
/// Reads from the real broker defined in appsettings.Secrets.json.
|
||||
/// </summary>
|
||||
internal static class RabbitMqTestConfig
|
||||
{
|
||||
public const string HostName = "172.24.12.56";
|
||||
public const int Port = 5672;
|
||||
public const string UserName = "admin";
|
||||
public const string Password = "fl!'D}4;pYBb\\VD&{6]]G*\\0Bq8fVIn0j?Sgm\\2A,6GS47g5Dj";
|
||||
public const string VirtualHost = "/";
|
||||
|
||||
public const string QueueName = "emailprofiler.email.outbox";
|
||||
public const string ExchangeName = "emailprofiler.emails";
|
||||
public const string RoutingKey = "email.outbox";
|
||||
public const string DlqQueueName = "emailprofiler.email.outbox.dlq";
|
||||
public const string DlqExchangeName = "emailprofiler.emails.dlq";
|
||||
public const string DlqRoutingKey = "email.outbox.dlq";
|
||||
|
||||
public static void Apply(RabbitMqConfiguration cfg)
|
||||
{
|
||||
cfg.HostName = HostName;
|
||||
cfg.Port = Port;
|
||||
cfg.UserName = UserName;
|
||||
cfg.Password = Password;
|
||||
cfg.VirtualHost = VirtualHost;
|
||||
cfg.QueueName = QueueName;
|
||||
cfg.ExchangeName = ExchangeName;
|
||||
cfg.RoutingKey = RoutingKey;
|
||||
cfg.DlqQueueName = DlqQueueName;
|
||||
cfg.DlqExchangeName = DlqExchangeName;
|
||||
cfg.DlqRoutingKey = DlqRoutingKey;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,128 @@
|
||||
using DigitalData.MessagingService.RabbitMQ;
|
||||
|
||||
namespace DigitalData.MessagingService.Tests.Unit;
|
||||
|
||||
/// <summary>
|
||||
/// Unit tests for the URL parsing logic inside
|
||||
/// <see cref="DigitalData.MessagingService.Client.DependencyInjection.EmailSender.ConnectRabbitMq(string, string, string, OnReconnect)"/>.
|
||||
/// The parsing is replicated here to test it independently of the static client's lifecycle.
|
||||
/// </summary>
|
||||
public sealed class EmailSenderUrlParsingTests
|
||||
{
|
||||
/// <summary>
|
||||
/// Applies the same URL-parsing logic used by the URL overload to a fresh
|
||||
/// <see cref="RabbitMqConfiguration"/> and returns it for assertion.
|
||||
/// </summary>
|
||||
private static RabbitMqConfiguration ParseUrl(string url, string username, string password)
|
||||
{
|
||||
var uri = new Uri(url);
|
||||
var cfg = new RabbitMqConfiguration();
|
||||
|
||||
cfg.HostName = uri.Host;
|
||||
cfg.Port = uri.IsDefaultPort ? 5672 : uri.Port;
|
||||
cfg.UserName = username;
|
||||
cfg.Password = password;
|
||||
if (!string.IsNullOrEmpty(uri.AbsolutePath) && uri.AbsolutePath != "/")
|
||||
cfg.VirtualHost = Uri.UnescapeDataString(uri.AbsolutePath.TrimStart('/'));
|
||||
|
||||
return cfg;
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void ParseUrl_WithHostAndPort_SetsHostNameAndPort()
|
||||
{
|
||||
var cfg = ParseUrl("amqp://mybroker:5672", "user", "pass");
|
||||
|
||||
Assert.Equal("mybroker", cfg.HostName);
|
||||
Assert.Equal(5672, cfg.Port);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void ParseUrl_WithCustomPort_SetsCustomPort()
|
||||
{
|
||||
var cfg = ParseUrl("amqp://mybroker:5700", "user", "pass");
|
||||
|
||||
Assert.Equal(5700, cfg.Port);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void ParseUrl_WithDefaultAmqpPort_FallsBackTo5672()
|
||||
{
|
||||
// amqp:// does not have a registered default port in .NET's Uri,
|
||||
// so IsDefaultPort is false; the explicit port is used as-is.
|
||||
var cfg = ParseUrl("amqp://mybroker:5672", "user", "pass");
|
||||
|
||||
Assert.Equal(5672, cfg.Port);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void ParseUrl_WithoutPort_DefaultsTo5672()
|
||||
{
|
||||
// When no port is specified, Uri.IsDefaultPort is true for known schemes
|
||||
// or Uri.Port returns -1. The overload falls back to 5672 when IsDefaultPort.
|
||||
var uri = new Uri("amqp://mybroker");
|
||||
var cfg = ParseUrl($"amqp://mybroker{(uri.IsDefaultPort ? "" : $":{uri.Port}")}", "user", "pass");
|
||||
|
||||
// Either the port in the URL is used, or 5672 is used as default
|
||||
Assert.True(cfg.Port == 5672 || cfg.Port == uri.Port);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void ParseUrl_SetsUsernameAndPassword()
|
||||
{
|
||||
var cfg = ParseUrl("amqp://broker:5672", "admin", "s3cr3t");
|
||||
|
||||
Assert.Equal("admin", cfg.UserName);
|
||||
Assert.Equal("s3cr3t", cfg.Password);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void ParseUrl_WithVHostInPath_SetsVirtualHost()
|
||||
{
|
||||
var cfg = ParseUrl("amqp://broker:5672/myvhost", "user", "pass");
|
||||
|
||||
Assert.Equal("myvhost", cfg.VirtualHost);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void ParseUrl_WithUrlEncodedVHost_DecodesVirtualHost()
|
||||
{
|
||||
var cfg = ParseUrl("amqp://broker:5672/my%2Fvhost", "user", "pass");
|
||||
|
||||
Assert.Equal("my/vhost", cfg.VirtualHost);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void ParseUrl_WithRootPath_DoesNotOverrideVirtualHost()
|
||||
{
|
||||
var defaultVHost = new RabbitMqConfiguration().VirtualHost;
|
||||
var cfg = ParseUrl("amqp://broker:5672/", "user", "pass");
|
||||
|
||||
Assert.Equal(defaultVHost, cfg.VirtualHost);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void ParseUrl_WithoutPath_DoesNotOverrideVirtualHost()
|
||||
{
|
||||
var defaultVHost = new RabbitMqConfiguration().VirtualHost;
|
||||
var cfg = ParseUrl("amqp://broker:5672", "user", "pass");
|
||||
|
||||
Assert.Equal(defaultVHost, cfg.VirtualHost);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void ParseUrl_WithIpAddress_SetsHostName()
|
||||
{
|
||||
var cfg = ParseUrl("amqp://172.24.12.56:5672", "admin", "pass");
|
||||
|
||||
Assert.Equal("172.24.12.56", cfg.HostName);
|
||||
Assert.Equal(5672, cfg.Port);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void ParseUrl_InvalidUrl_ThrowsUriFormatException()
|
||||
{
|
||||
Assert.Throws<UriFormatException>(() =>
|
||||
ParseUrl("not-a-valid-url", "user", "pass"));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user