Skip to content

Commit db4ebd4

Browse files
authored
Add test to promote mirrored to regular stream (#1008)
1 parent af50d8b commit db4ebd4

File tree

1 file changed

+60
-0
lines changed

1 file changed

+60
-0
lines changed

tests/NATS.Client.JetStream.Tests/ManageStreamTest.cs

Lines changed: 60 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -377,4 +377,64 @@ public async Task PersistMode_property_should_be_set_on_stream()
377377
var updatedAsync = await js.GetStreamAsync($"{prefix}persist-async", cancellationToken: cts.Token);
378378
Assert.Equal(StreamConfigPersistMode.Async, updatedAsync.Info.Config.PersistMode);
379379
}
380+
381+
[SkipIfNatsServer(versionEarlierThan: "2.12")]
382+
public async Task Remove_mirror_config_to_promote_stream()
383+
{
384+
await using var nats = new NatsConnection(new NatsOpts { Url = _server.Url });
385+
var prefix = _server.GetNextId();
386+
await nats.ConnectRetryAsync();
387+
388+
var js = new NatsJSContext(nats);
389+
var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30));
390+
391+
// Create source stream
392+
await js.CreateStreamAsync(
393+
new StreamConfig($"{prefix}SOURCE", [$"{prefix}foo"]),
394+
cts.Token);
395+
396+
// Publish some messages to source
397+
for (var i = 0; i < 10; i++)
398+
{
399+
await js.PublishAsync($"{prefix}foo", new byte[] { (byte)i }, cancellationToken: cts.Token);
400+
}
401+
402+
// Create mirror stream
403+
var mirror = await js.CreateStreamAsync(
404+
new StreamConfig
405+
{
406+
Name = $"{prefix}MIRROR",
407+
Mirror = new StreamSource { Name = $"{prefix}SOURCE" },
408+
},
409+
cts.Token);
410+
411+
Assert.NotNull(mirror.Info.Config.Mirror);
412+
Assert.Equal($"{prefix}SOURCE", mirror.Info.Config.Mirror.Name);
413+
414+
// Wait for mirror to catch up
415+
await Retry.Until(
416+
"mirror caught up",
417+
async () =>
418+
{
419+
await mirror.RefreshAsync(cts.Token);
420+
return mirror.Info.State.Messages == 10;
421+
},
422+
timeout: TimeSpan.FromSeconds(10));
423+
424+
// Remove mirror configuration - promote to regular stream
425+
var promoted = await js.UpdateStreamAsync(
426+
new StreamConfig
427+
{
428+
Name = $"{prefix}MIRROR",
429+
Subjects = [$"{prefix}bar"],
430+
},
431+
cts.Token);
432+
433+
// Verify mirror is null after promotion
434+
Assert.Null(promoted.Info.Config.Mirror);
435+
Assert.Contains($"{prefix}bar", promoted.Info.Config.Subjects!);
436+
437+
// Verify messages are retained after promotion
438+
Assert.Equal(10, promoted.Info.State.Messages);
439+
}
380440
}

0 commit comments

Comments
 (0)