Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -76,6 +76,11 @@ public async Task SendAsync_Should_RouteToErrorQueue_When_HandlerThrows()
t.Endpoint("payment-ep")
.Handler<ThrowingPaymentHandler>()
.FaultEndpoint(new Uri("azuresb:q/payment-q_error"));
// the handler sits on an explicitly named queue, so sends have to be directed
// there instead of to the message-type derived queue the convention would pick
t.DispatchEndpoint("payment-send-ep")
.ToQueue("payment-ep")
.Send<ProcessPayment>();
t.Endpoint("payment-error-ep")
.Queue("payment-q_error")
// we mark it as an error because only then no route will be provisioned for the
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -208,15 +208,18 @@ public async Task Inbox_Should_ProcessMessage_When_MessageIdIsNull()
await messageBus.PublishAsync(new InboxEvent { Payload = "null-id-1" }, CancellationToken.None);
await messageBus.PublishAsync(new InboxEvent { Payload = "null-id-2" }, CancellationToken.None);

// assert - both should be processed since null MessageId cannot be deduplicated
// assert - both are processed, a cleared MessageId leaves nothing to deduplicate on
Assert.True(
await recorder.WaitAsync(s_timeout, expectedCount: 2),
"Handler should receive both messages when MessageId is null");

Assert.Equal(2, recorder.Messages.Count);

// Inbox should NOT have recorded anything since MessageId was null
Assert.Empty(inbox.RecordedEnvelopes);
// the broker stamps its own MessageId when the envelope carries none, so the inbox claims
// each message under a distinct id rather than collapsing them into one
Assert.Equal(
2,
inbox.RecordedEnvelopes.Select(e => e.MessageId).Distinct(StringComparer.Ordinal).Count());
}

private static async Task WaitUntilAsync(Func<bool> condition, TimeSpan timeout)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -45,10 +45,11 @@ public async Task UseNativeDeadLetterForwarding_Should_SetForwardDeadLetteredMes
// assert
Assert.Equal($"{queueName}_error", queue.ForwardDeadLetteredMessagesTo);

// and the broker reflects it
// and the broker reflects it, echoing the target back as an absolute entity URI
var adminClient = new ServiceBusAdministrationClient(ctx.AdminConnectionString);
var properties = await adminClient.GetQueueAsync(queueName, Xunit.TestContext.Current.CancellationToken);
Assert.Equal($"{queueName}_error", properties.Value.ForwardDeadLetteredMessagesTo);
var forwardedTo = new Uri(properties.Value.ForwardDeadLetteredMessagesTo);
Assert.Equal($"/{queueName}_error", forwardedTo.AbsolutePath);
Comment thread
PascalSenn marked this conversation as resolved.
}

[Fact]
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -48,12 +48,15 @@ public async Task RequestAsync_Should_CorrelateResponses_When_ConcurrentRequests
.AddAzureServiceBus(ctx)
.BuildTestBusAsync();

// a single scope that outlives every in-flight request, the requests only complete once
// their responses have been correlated back
using var scope = bus.Provider.CreateScope();
var messageBus = scope.ServiceProvider.GetRequiredService<IMessageBus>();

// act
var tasks = new Task<OrderStatusResponse>[10];
for (var i = 0; i < 10; i++)
{
using var scope = bus.Provider.CreateScope();
var messageBus = scope.ServiceProvider.GetRequiredService<IMessageBus>();
tasks[i] = messageBus
.RequestAsync(new GetOrderStatus { OrderId = $"ORD-{i}" }, CancellationToken.None)
.AsTask();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -107,10 +107,15 @@ internal async ValueTask CleanupEntitiesAsync()
}
}

/// <summary>
/// Builds a lower-kebab prefix. Names derived from an entity name, such as the fault queue,
/// are run through the naming conventions, which lowercase and replace underscores. Keeping the
/// prefix in that shape makes the derived names line up with the entity they were derived from.
/// </summary>
private static string GeneratePrefix(string testName, string filePath)
{
var hash = Convert.ToHexString(SHA256.HashData(Encoding.UTF8.GetBytes(filePath)))[..8];
return $"{testName}_{hash}";
return $"{testName}-{hash}".ToLowerInvariant().Replace('_', '-');
}
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,14 @@ namespace ChilliCream.Nitro.CommandLine.Tests.Tui.Runtime;
public sealed class TaskDbWatcherTests : IDisposable
{
private static readonly TimeSpan Debounce = TimeSpan.FromMilliseconds(50);

/// <summary>
/// The debounce used by the coalescing test. It is far wider than the time the burst takes to
/// write so that neither file system event delivery nor a scheduler delay stretched by a loaded
/// machine can push two writes of the same burst into separate debounce windows.
/// </summary>
private static readonly TimeSpan BurstDebounce = TimeSpan.FromMilliseconds(500);

private static readonly TimeSpan TestTimeout = TimeSpan.FromSeconds(5);

private readonly string _directory =
Expand Down Expand Up @@ -72,27 +80,28 @@ public async Task RunAsync_Should_CoalesceBurstOfWrites_IntoSingleEvent()
var testToken = TestContext.Current.CancellationToken;
var databasePath = Path.Combine(_directory, "tasks.db");
File.WriteAllText(databasePath, "initial");
var watcher = new TaskDbWatcher(databasePath, Debounce);
var watcher = new TaskDbWatcher(databasePath, BurstDebounce);
var channel = Channel.CreateUnbounded<TuiEvent>();
using var cts = CancellationTokenSource.CreateLinkedTokenSource(testToken);

// act
var runTask = watcher.RunAsync(channel.Writer, cts.Token);
await Task.Delay(Debounce, testToken);

// A burst of writes to the db and its WAL/SHM siblings within one debounce
// window resets the same timer rather than each scheduling its own event.
// A burst of writes to the db and its WAL sibling within one debounce window resets the
// same timer rather than each write scheduling its own event. The writes are issued back
// to back so that the burst cannot be stretched across two windows by a delay that a
// loaded machine overruns.
for (var i = 0; i < 5; i++)
{
File.WriteAllText(databasePath, "changed-" + i);
File.WriteAllText(databasePath + "-wal", "wal-" + i);
await Task.Delay(Debounce / 5, testToken);
}

var first = await ReadOneAsync(channel.Reader, testToken);

// No further event should follow once the burst settles.
await Task.Delay(Debounce * 4, testToken);
await Task.Delay(BurstDebounce * 2, testToken);
cts.Cancel();
await runTask;

Expand Down
Loading