Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
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
7 changes: 7 additions & 0 deletions src/NATS.Client.JetStream/Models/StreamConfig.cs
Original file line number Diff line number Diff line change
Expand Up @@ -262,4 +262,11 @@ public StreamConfig()
[System.Text.Json.Serialization.JsonPropertyName("metadata")]
[System.Text.Json.Serialization.JsonIgnore(Condition = System.Text.Json.Serialization.JsonIgnoreCondition.WhenWritingDefault)]
public IDictionary<string, string>? Metadata { get; set; }

/// <summary>
/// AllowAtomicPublish allows atomic batch publishing into the stream.
/// </summary>
[System.Text.Json.Serialization.JsonPropertyName("allow_atomic")]
[System.Text.Json.Serialization.JsonIgnore(Condition = System.Text.Json.Serialization.JsonIgnoreCondition.WhenWritingDefault)]
public bool AllowAtomicPublish { get; set; }
}
39 changes: 39 additions & 0 deletions tests/NATS.Client.JetStream.Tests/ManageStreamTest.cs
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@
using NATS.Client.Core2.Tests;
using NATS.Client.JetStream.Models;
using NATS.Client.Platform.Windows.Tests;
using NATS.Client.TestUtilities;
using NATS.Client.TestUtilities2;

namespace NATS.Client.JetStream.Tests;
Expand Down Expand Up @@ -219,4 +220,42 @@ public async Task Create_or_update_stream_should_be_throwing_update_operation_er
await js.CreateOrUpdateStreamAsync(streamConfig, cts.Token);
await Assert.ThrowsAsync<NatsJSApiException>(async () => await js.CreateOrUpdateStreamAsync(streamConfigForUpdated, cts.Token));
}

[SkipIfNatsServer(versionEarlierThan: "2.12")]
public async Task AllowAtomicPublish_property_should_be_set_on_stream()
{
await using var nats = new NatsConnection(new NatsOpts { Url = _server.Url });
var prefix = _server.GetNextId();
await nats.ConnectRetryAsync();

var js = new NatsJSContext(nats);

var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));

// Create a stream with AllowAtomicPublish enabled
var streamConfig = new StreamConfig($"{prefix}atomic", [$"{prefix}atomic.*"])
{
AllowAtomicPublish = true,
};

var stream = await js.CreateStreamAsync(streamConfig, cts.Token);

// Verify the property is set on the created stream
Assert.True(stream.Info.Config.AllowAtomicPublish);

// Get the stream and verify the property is persisted
var retrievedStream = await js.GetStreamAsync($"{prefix}atomic", cancellationToken: cts.Token);
Assert.True(retrievedStream.Info.Config.AllowAtomicPublish);

// Update stream with AllowAtomicPublish disabled
var updatedConfig = streamConfig with { AllowAtomicPublish = false };
var updatedStream = await js.UpdateStreamAsync(updatedConfig, cts.Token);

// Verify the property is updated
Assert.False(updatedStream.Info.Config.AllowAtomicPublish);

// Get the stream and verify the update is persisted
var reRetrievedStream = await js.GetStreamAsync($"{prefix}atomic", cancellationToken: cts.Token);
Assert.False(reRetrievedStream.Info.Config.AllowAtomicPublish);
}
}