tomasfabian
commented
2 years ago Hi @ArnoRienstra,
this is really unfortunate. Hopefully the .NET team will add a EndOfStreamAsync + ReadLineAsync which can accept a CancellationToken.
Could you help me to find a suitable solution, please? What do you think about registering an action for the cancellationToken? In the following way the while loop is exited either based on the streamReader.EndOfStream or by releasing the semaphoreSlim:
public async IAsyncEnumerable<T> Run<T>(object parameters, [EnumeratorCancellation] CancellationToken cancellationToken = default)
{
logger?.LogInformation($"Executing query {parameters}");
using var streamReader = await TryGetStreamReaderAsync<T>(parameters, cancellationToken).ConfigureAwait(false);
using var semaphoreSlim = new SemaphoreSlim(0, 1);
cancellationToken.Register(() =>
{
semaphoreSlim.Release();
});
await foreach (var entity in ConsumeAsync<T>(streamReader, semaphoreSlim, cancellationToken).WithCancellation(cancellationToken).ConfigureAwait(false))
yield return entity;
}
private Task<bool> EndOfStreamAsync(StreamReader streamReader, CancellationToken cancellationToken)
{
return Task.Run(() =>
{
try
{
return !streamReader.EndOfStream;
}
catch (Exception)
{
if (!cancellationToken.IsCancellationRequested)
throw;
return true;
}
}, cancellationToken);
}
private async IAsyncEnumerable<T> ConsumeAsync<T>(StreamReader streamReader, SemaphoreSlim semaphoreSlim, [EnumeratorCancellation] CancellationToken cancellationToken)
{
if (cancellationToken.IsCancellationRequested)
yield break;
var cancellationTask = Task.Run(async () =>
{
await semaphoreSlim.WaitAsync().ConfigureAwait(false);
return false;
}, cancellationToken);
while (await await Task.WhenAny(EndOfStreamAsync(streamReader, cancellationToken), cancellationTask).ConfigureAwait(false))
{
if (cancellationToken.IsCancellationRequested)
yield break;
var rawData = await streamReader.ReadLineAsync()
.ConfigureAwait(false);
logger?.LogDebug($"Raw data received: {rawData}");
var record = OnLineRead<T>(rawData);
if (record != null) yield return record.Value;
}
}Could you evaluate/test the above solution, please? Is streamReader.ReadLineAsync() still blocking in your case?
If you can come up with a better solution please let me know.
Thank you! Regards Tomas.
ArnoRienstra
Describe the bug KSQL-queries are not removed from KSQL if nothing received after cancelling. CreateQueryStream is calling ConsumeAsync, which is not responding on CancellationToken (while waiting on EndOfStream or ReadLineAsync)
To Reproduce After creating a push query using a filter on a certain ID, the ksql query is not removed from ksql if no new information is streamed. In our case, if a device is stopped or removed, there is no new record published and as such the library is waiting in ReadLineAsync. After initiating a new record the EndOfstream is continued and cancellation performed thereafter and as such ksql-query cancelled in ksql.
Expected behavior Perhaps exception is to be raised on thread from cancellationtoken, being handled by the ConsumeAsync? When .NET7 is released, use the cancellationtoken in the ReadLineAsync?
Environment (please complete the following information):