stevejgordon

<Project Sdk="Microsoft.NET.Sdk">

  <PropertyGroup>
    <OutputType>Exe</OutputType>
    <TargetFramework>net5.0</TargetFramework>
  </PropertyGroup>

  <ItemGroup>
    <PackageReference Include="Elasticsearch.Net" Version="7.10.0" />
  </ItemGroup>

stevejgordon

public virtual async IAsyncEnumerable<T> ReadAllAsync([EnumeratorCancellation] CancellationToken cancellationToken = default)
{
    while (await WaitToReadAsync(cancellationToken).ConfigureAwait(false))
    {
        while (TryRead(out T item))
        {
            yield return item;
        }
    }
}

stevejgordon

var waiter = new AsyncOperation<bool>(parent._runContinuationsAsynchronously, cancellationToken);
ChannelUtilities.QueueWaiter(ref parent._waitingReadersTail, waiter);
return waiter.ValueTaskOfT;

stevejgordon

if (!cancellationToken.CanBeCanceled)
{
    AsyncOperation<bool> singleton = _waiterSingleton;
    if (singleton.TryOwnAndReset())
    {
        ChannelUtilities.QueueWaiter(ref parent._waitingReadersTail, singleton);
        return singleton.ValueTaskOfT;
    }
}

stevejgordon

if (parent._doneWriting != null)
{
    return parent._doneWriting != ChannelUtilities.s_doneWritingSentinel ?
        new ValueTask<bool>(Task.FromException<bool>(parent._doneWriting)) :
        default;
}

stevejgordon

UnboundedChannel<T> parent = _parent;

lock (parent.SyncObj)
{
    parent.AssertInvariants();

    // Try again to read now that we're synchronized with writers.
    if (!parent._items.IsEmpty)
    {
        return new ValueTask<bool>(true);

stevejgordon

if (cancellationToken.IsCancellationRequested)
{
    return new ValueTask<bool>(Task.FromCanceled<bool>(cancellationToken));
}

if (!_parent._items.IsEmpty)
{
    return new ValueTask<bool>(true);
}

stevejgordon

public override bool TryRead([MaybeNullWhen(false)] out T item)
{
    UnboundedChannel<T> parent = _parent;

    // Dequeue an item if we can
    if (parent._items.TryDequeue(out item))
    {
        CompleteIfDone(parent);
        return true;
    }

stevejgordon

var reader = new AsyncOperation<T>(parent._runContinuationsAsynchronously, cancellationToken);
parent._blockedReaders.EnqueueTail(reader);
return reader.ValueTaskOfT;

stevejgordon

if (!cancellationToken.CanBeCanceled)
{
    AsyncOperation<T> singleton = _readerSingleton;
    if (singleton.TryOwnAndReset())
    {
        parent._blockedReaders.EnqueueTail(singleton);
        return singleton.ValueTaskOfT;
    }
}