stevejgordon

{"attributes":{"controlGroupInput":{"chainingSystem":"HIERARCHICAL","controlStyle":"oneLine","ignoreParentSettingsJSON":"{\"ignoreFilters\":false,\"ignoreQuery\":false,\"ignoreTimerange\":false,\"ignoreValidations\":false}","panelsJSON":"{\"2be66584-9de4-4a36-ba54-bfdd1b4ccfb4\":{\"grow\":true,\"order\":0,\"type\":\"optionsListControl\",\"width\":\"medium\",\"explicitInput\":{\"dataViewId\":\"apm_static_data_view_id_default\",\"fieldName\":\"service.node.name\",\"title\":\"Instance\",\"exclude\":null,\"existsSelected\":true,\"selectedOptions\":[],\"searchTechnique\":\"exact\",\"singleSelect\":null,\"runPastTimeout\":null,\"sort\":{\"by\":\"_count\",\"direction\":\"desc\"},\"placeholder\":null,\"hideActionBar\":null,\"hideExclude\":null,\"hideExists\":null,\"hideSort\":null}}}","showApplySelections":false},"description":"","kibanaSavedObjectMeta":{"searchSourceJSON":"{\"filter\":[],\"query\":{\"query\":\"\",\"language\":\"kuery\"}}"},"optionsJSON":"{\"useMargins\":true,\"syncColors\":false,\"syncCursor\":true,

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;