Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

ObjStore empty list fix #572

Closed
wants to merge 3 commits into from
Closed

ObjStore empty list fix #572

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
02e13c3
Added OnNoData Async Func for break the async enumeration
Jul 20, 2024
de06b58
Consumer Dispose fix, CreateSub operation didn't have time for execut…
Jul 20, 2024
4f7a083
Added a comment to the dispose order
Jul 23, 2024
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: 4 additions & 3 deletions src/NATS.Client.JetStream/Internal/NatsJSOrderedPushConsumer.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -142,12 +142,13 @@ public async ValueTask DisposeAsync()
_nats.ConnectionDisconnected -= OnDisconnected;

_consumerCreateChannel.Writer.TryComplete();
_commandChannel.Writer.TryComplete();
_msgChannel.Writer.TryComplete();

await _consumerCreateTask;

_commandChannel.Writer.TryComplete();
await _commandTask;

_msgChannel.Writer.TryComplete();

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

why did we change these? did you encounter an issue? (you're probably right to make these changes, I just can't remember my reasoning here so just curious)

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, I encountered some problem.
Before exiting WatchAsync, if Enumerable is empty, call pushConsumer.Init();
The consumer is initialized and the same consumer is immediately deleted in the dispose method.
The creation operation _consumerCreateTask has not yet had time to complete and it requires an active _commandChannel, but it is already closed. Based on this, I decided to change the order of stopping channel reading.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Otherwise, when Dispose we get NatsJSApiException: consumer not found.

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

awesome, thanks fixing that 💯 should we stick this as comment on top of the relevant lines do you think?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good idea! Added.

await _context.DeleteConsumerAsync(_stream, Consumer, _cancellationToken);
}

Expand Down
19 changes: 19 additions & 0 deletions src/NATS.Client.ObjectStore/NatsObjStore.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -534,6 +534,7 @@ public IAsyncEnumerable<ObjectMetadata> ListAsync(NatsObjListOpts? opts = defaul
InitialSetOnly = true,
UpdatesOnly = false,
IgnoreDeletes = !opts.ShowDeleted,
OnNoData = opts.OnNoData,
};
return WatchAsync(watchOpts, cancellationToken);
}
Expand Down Expand Up @@ -583,6 +584,14 @@ public async IAsyncEnumerable<ObjectMetadata> WatchAsync(NatsObjWatchOpts? opts

pushConsumer.Init();

if (pushConsumer.Msgs.Count == 0 && opts.OnNoData != null)
{
if (await opts.OnNoData(cancellationToken))
{
yield break;
}
}

await foreach (var msg in pushConsumer.Msgs.ReadAllAsync(cancellationToken).ConfigureAwait(false))
{
if (pushConsumer.IsDone)
Expand Down Expand Up @@ -685,11 +694,21 @@ public record NatsObjWatchOpts
/// Only return the initial set of objects and don't watch for further updates.
/// </summary>
public bool InitialSetOnly { get; init; }

/// <summary>
/// Async function called when the enumerator reaches the end of data. Return True to break the async enumeration, False to allow the enumeration to continue.
/// </summary>
public Func<CancellationToken, ValueTask<bool>>? OnNoData { get; init; }
}

public record NatsObjListOpts
{
public bool ShowDeleted { get; init; }

/// <summary>
/// Async function called when the enumerator reaches the end of data. Return True to break the async enumeration, False to allow the enumeration to continue.
/// </summary>
public Func<CancellationToken, ValueTask<bool>>? OnNoData { get; init; }
}

public record NatsObjStatus(string Bucket, bool IsCompressed, StreamInfo Info);
43 changes: 43 additions & 0 deletions tests/NATS.Client.ObjectStore.Tests/ObjectStoreTest.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -389,6 +389,49 @@ public async Task List()
}
}

[Fact]
public async Task List_empty_store_for_end_of_data()
{
var timeout = TimeSpan.FromSeconds(10);
var cts = new CancellationTokenSource(timeout);
var cancellationToken = cts.Token;

await using var server = NatsServer.StartJS();
await using var nats = server.CreateClientConnection();
var js = new NatsJSContext(nats);
var obj = new NatsObjContext(js);

var store = await obj.CreateObjectStoreAsync(new NatsObjConfig("b1"), cancellationToken);

var signal = new WaitSignal(timeout);
var endOfDataHit = false;
var watchTask = Task.Run(
async () =>
{
var opts = new NatsObjListOpts()
{
OnNoData = async (_) =>
{
await Task.CompletedTask;
endOfDataHit = true;
signal.Pulse();
return true;
},
};

await foreach (var info in store.ListAsync(opts: opts, cancellationToken: cancellationToken))
{
}
},
cancellationToken);

await signal;

Assert.True(endOfDataHit, "End of Current Data not set");

await watchTask;
}

[SkipIfNatsServer(versionEarlierThan: "2.10")]
public async Task Compressed_storage()
{
Expand Down
Loading