dohertj2/CBDD
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Reformat / cleanup
All checks were successful
NuGet Publish / build-and-pack (push) Successful in 46s
Details
NuGet Publish / publish-to-gitea (push) Successful in 56s
Details

Browse Source
This commit is contained in:
Joseph Doherty
2026-02-21 08:10:36 -05:00
parent 4c6aaa5a3f
commit a70d8befae
176 changed files with 50555 additions and 49587 deletions
Download Patch File Download Diff File Expand all files Collapse all files

57
tests/CBDD.Tests/Cdc/CdcScalabilityTests.cs
View File

@@ -1,29 +1,34 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using ZB.MOM.WW.CBDD.Core.CDC;
using ZB.MOM.WW.CBDD.Shared;
using Xunit;
namespace ZB.MOM.WW.CBDD.Tests;
public class CdcScalabilityTests : IDisposable
{
private readonly Shared.TestDbContext _db;
private readonly TestDbContext _db;
private readonly string _dbPath;
/// <summary>
/// Initializes a new instance of the <see cref="CdcScalabilityTests"/> class.
/// Initializes a new instance of the <see cref="CdcScalabilityTests" /> class.
/// </summary>
public CdcScalabilityTests()
{
_dbPath = Path.Combine(Path.GetTempPath(), $"cdc_scaling_{Guid.NewGuid()}.db");
_db = new Shared.TestDbContext(_dbPath);
_db = new TestDbContext(_dbPath);
}
/// <summary>
/// Verifies CDC dispatch reaches all registered subscribers.
/// Disposes test resources and removes temporary files.
/// </summary>
public void Dispose()
{
_db.Dispose();
if (File.Exists(_dbPath)) File.Delete(_dbPath);
string wal = Path.ChangeExtension(_dbPath, ".wal");
if (File.Exists(wal)) File.Delete(wal);
}
/// <summary>
/// Verifies CDC dispatch reaches all registered subscribers.
/// </summary>
[Fact]
public async Task Test_Cdc_1000_Subscribers_Receive_Events()
@@ -34,13 +39,10 @@ public class CdcScalabilityTests : IDisposable
var subscriptions = new List<IDisposable>();
// 1. Create 1000 subscribers
for (int i = 0; i < SubscriberCount; i++)
for (var i = 0; i < SubscriberCount; i++)
{
int index = i;
var sub = _db.People.Watch().Subscribe(_ =>
{
Interlocked.Increment(ref eventCounts[index]);
});
var sub = _db.People.Watch().Subscribe(_ => { Interlocked.Increment(ref eventCounts[index]); });
subscriptions.Add(sub);
}
@@ -53,16 +55,13 @@ public class CdcScalabilityTests : IDisposable
await Task.Delay(1000, ct);
// 4. Verify all subscribers received both events
for (int i = 0; i < SubscriberCount; i++)
{
eventCounts[i].ShouldBe(2);
}
for (var i = 0; i < SubscriberCount; i++) eventCounts[i].ShouldBe(2);
foreach (var sub in subscriptions) sub.Dispose();
}
/// <summary>
/// Verifies a slow subscriber does not block other subscribers.
/// Verifies a slow subscriber does not block other subscribers.
/// </summary>
[Fact(Skip = "Performance test - run manually when needed")]
public async Task Test_Cdc_Slow_Subscriber_Does_Not_Block_Others()
@@ -80,10 +79,7 @@ public class CdcScalabilityTests : IDisposable
});
// 2. Register a fast subscriber
using var fastSub = _db.People.Watch().Subscribe(_ =>
{
Interlocked.Increment(ref fastEventCount);
});
using var fastSub = _db.People.Watch().Subscribe(_ => { Interlocked.Increment(ref fastEventCount); });
// 3. Perform a write
_db.People.Insert(new Person { Id = 1, Name = "John", Age = 30 });
@@ -107,15 +103,4 @@ public class CdcScalabilityTests : IDisposable
await Task.Delay(2500, ct); // Wait for the second one in slow sub to be processed after the first Sleep
slowEventCount.ShouldBe(2);
}
/// <summary>
/// Disposes test resources and removes temporary files.
/// </summary>
public void Dispose()
{
_db.Dispose();
if (File.Exists(_dbPath)) File.Delete(_dbPath);
var wal = Path.ChangeExtension(_dbPath, ".wal");
if (File.Exists(wal)) File.Delete(wal);
}
}
}

94
tests/CBDD.Tests/Cdc/CdcTests.cs
View File

@@ -1,15 +1,8 @@
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
using System.Linq;
using System.Threading.Tasks;
using ZB.MOM.WW.CBDD.Bson;
using ZB.MOM.WW.CBDD.Core.CDC;
using ZB.MOM.WW.CBDD.Core.Transactions;
using ZB.MOM.WW.CBDD.Shared;
using Xunit;
namespace ZB.MOM.WW.CBDD.Tests;
@@ -17,33 +10,43 @@ public class CdcTests : IDisposable
{
private static readonly TimeSpan DefaultEventTimeout = TimeSpan.FromSeconds(3);
private static readonly TimeSpan PollInterval = TimeSpan.FromMilliseconds(10);
private readonly TestDbContext _db;
private readonly string _dbPath = $"cdc_test_{Guid.NewGuid()}.db";
private readonly Shared.TestDbContext _db;
/// <summary>
/// Initializes a new instance of the <see cref="CdcTests"/> class.
/// Initializes a new instance of the <see cref="CdcTests" /> class.
/// </summary>
public CdcTests()
{
_db = new Shared.TestDbContext(_dbPath);
_db = new TestDbContext(_dbPath);
}
/// <summary>
/// Verifies that an insert operation publishes a CDC event.
/// Disposes test resources and removes temporary files.
/// </summary>
public void Dispose()
{
_db.Dispose();
if (File.Exists(_dbPath)) File.Delete(_dbPath);
if (File.Exists(_dbPath + "-wal")) File.Delete(_dbPath + "-wal");
}
/// <summary>
/// Verifies that an insert operation publishes a CDC event.
/// </summary>
[Fact]
public async Task Test_Cdc_Basic_Insert_Fires_Event()
{
var ct = TestContext.Current.CancellationToken;
var events = new ConcurrentQueue<ChangeStreamEvent<int, Person>>();
using var subscription = _db.People.Watch(capturePayload: true).Subscribe(events.Enqueue);
using var subscription = _db.People.Watch(true).Subscribe(events.Enqueue);
var person = new Person { Id = 1, Name = "John", Age = 30 };
_db.People.Insert(person);
_db.SaveChanges();
await WaitForEventCountAsync(events, expectedCount: 1, ct);
await WaitForEventCountAsync(events, 1, ct);
var snapshot = events.ToArray();
snapshot.Length.ShouldBe(1);
@@ -54,20 +57,20 @@ public class CdcTests : IDisposable
}
/// <summary>
/// Verifies payload is omitted when CDC capture payload is disabled.
/// Verifies payload is omitted when CDC capture payload is disabled.
/// </summary>
[Fact]
public async Task Test_Cdc_No_Payload_When_Not_Requested()
{
var ct = TestContext.Current.CancellationToken;
var events = new ConcurrentQueue<ChangeStreamEvent<int, Person>>();
using var subscription = _db.People.Watch(capturePayload: false).Subscribe(events.Enqueue);
using var subscription = _db.People.Watch(false).Subscribe(events.Enqueue);
var person = new Person { Id = 1, Name = "John", Age = 30 };
_db.People.Insert(person);
_db.SaveChanges();
await WaitForEventCountAsync(events, expectedCount: 1, ct);
await WaitForEventCountAsync(events, 1, ct);
var snapshot = events.ToArray();
snapshot.Length.ShouldBe(1);
@@ -75,14 +78,14 @@ public class CdcTests : IDisposable
}
/// <summary>
/// Verifies CDC events are published only for committed changes.
/// Verifies CDC events are published only for committed changes.
/// </summary>
[Fact]
public async Task Test_Cdc_Commit_Only()
{
var ct = TestContext.Current.CancellationToken;
var events = new ConcurrentQueue<ChangeStreamEvent<int, Person>>();
using var subscription = _db.People.Watch(capturePayload: true).Subscribe(events.Enqueue);
using var subscription = _db.People.Watch(true).Subscribe(events.Enqueue);
using (var txn = _db.BeginTransaction())
{
@@ -101,21 +104,21 @@ public class CdcTests : IDisposable
txn.Commit();
}
await WaitForEventCountAsync(events, expectedCount: 1, ct);
await WaitForEventCountAsync(events, 1, ct);
var snapshot = events.ToArray();
snapshot.Length.ShouldBe(1);
snapshot[0].DocumentId.ShouldBe(2);
}
/// <summary>
/// Verifies update and delete operations publish CDC events.
/// Verifies update and delete operations publish CDC events.
/// </summary>
[Fact]
public async Task Test_Cdc_Update_And_Delete()
{
var ct = TestContext.Current.CancellationToken;
var events = new ConcurrentQueue<ChangeStreamEvent<int, Person>>();
using var subscription = _db.People.Watch(capturePayload: true).Subscribe(events.Enqueue);
using var subscription = _db.People.Watch(true).Subscribe(events.Enqueue);
var person = new Person { Id = 1, Name = "John", Age = 30 };
_db.People.Insert(person);
@@ -128,7 +131,7 @@ public class CdcTests : IDisposable
_db.People.Delete(1);
_db.SaveChanges();
await WaitForEventCountAsync(events, expectedCount: 3, ct);
await WaitForEventCountAsync(events, 3, ct);
var snapshot = events.ToArray();
snapshot.Length.ShouldBe(3);
@@ -140,16 +143,6 @@ public class CdcTests : IDisposable
snapshot[2].DocumentId.ShouldBe(1);
}
/// <summary>
/// Disposes test resources and removes temporary files.
/// </summary>
public void Dispose()
{
_db.Dispose();
if (File.Exists(_dbPath)) File.Delete(_dbPath);
if (File.Exists(_dbPath + "-wal")) File.Delete(_dbPath + "-wal");
}
private static async Task WaitForEventCountAsync(
ConcurrentQueue<ChangeStreamEvent<int, Person>> events,
int expectedCount,
@@ -158,10 +151,7 @@ public class CdcTests : IDisposable
var sw = Stopwatch.StartNew();
while (sw.Elapsed < DefaultEventTimeout)
{
if (events.Count >= expectedCount)
{
return;
}
if (events.Count >= expectedCount) return;
await Task.Delay(PollInterval, ct);
}
@@ -174,12 +164,12 @@ public class CdcTests : IDisposable
public static class ObservableExtensions
{
/// <summary>
/// Subscribes to an observable sequence using an action callback.
/// Subscribes to an observable sequence using an action callback.
/// </summary>
/// <typeparam name="T">The event type.</typeparam>
/// <param name="observable">The observable sequence.</param>
/// <param name="onNext">The callback for next events.</param>
/// <returns>An <see cref="IDisposable"/> subscription.</returns>
/// <returns>An <see cref="IDisposable" /> subscription.</returns>
public static IDisposable Subscribe<T>(this IObservable<T> observable, Action<T> onNext)
{
return observable.Subscribe(new AnonymousObserver<T>(onNext));
@@ -190,26 +180,36 @@ public static class ObservableExtensions
private readonly Action<T> _onNext;
/// <summary>
/// Initializes a new instance of the <see cref="AnonymousObserver{T}"/> class.
/// Initializes a new instance of the <see cref="AnonymousObserver{T}" /> class.
/// </summary>
/// <param name="onNext">The callback for next events.</param>
public AnonymousObserver(Action<T> onNext) => _onNext = onNext;
public AnonymousObserver(Action<T> onNext)
{
_onNext = onNext;
}
/// <summary>
/// Handles completion.
/// Handles completion.
/// </summary>
public void OnCompleted() { }
public void OnCompleted()
{
}
/// <summary>
/// Handles an observable error.
/// Handles an observable error.
/// </summary>
/// <param name="error">The observed error.</param>
public void OnError(Exception error) { }
public void OnError(Exception error)
{
}
/// <summary>
/// Handles the next value.
/// Handles the next value.
/// </summary>
/// <param name="value">The observed value.</param>
public void OnNext(T value) => _onNext(value);
public void OnNext(T value)
{
_onNext(value);
}
}
}
}