stevejgordon

if (parent._doneWriting != null)
{
    return ChannelUtilities.GetInvalidCompletionValueTask<T>(parent._doneWriting);
}

stevejgordon

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

    // Try to dequeue again, now that we hold the lock.
    if (parent._items.TryDequeue(out item))
    {
        CompleteIfDone(parent);
        return new ValueTask<T>(item);
    }

stevejgordon

private void CompleteIfDone(UnboundedChannel<T> parent)
{
    if (parent._doneWriting != null && parent._items.IsEmpty)
    {
        // If we've now emptied the items queue and we're not getting any more, complete.
        ChannelUtilities.Complete(parent._completion, parent._doneWriting);
    }
}

stevejgordon

UnboundedChannel<T> parent = _parent;
if (parent._items.TryDequeue(out T item))
{
    CompleteIfDone(parent);
    return new ValueTask<T>(item);
}

stevejgordon

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

stevejgordon

internal UnboundedChannelReader(UnboundedChannel<T> parent)
{
    _parent = parent;
    _readerSingleton = new AsyncOperation<T>(parent._runContinuationsAsynchronously, pooled: true);
    _waiterSingleton = new AsyncOperation<bool>(parent._runContinuationsAsynchronously, pooled: true);
}

stevejgordon

if (blockedReader != null)
{
    // Complete the reader.  It's possible the reader was canceled, in which
    // case we loop around to try everything again.
    if (blockedReader.TrySetResult(item))
    {
        return true;
    }
}
else

stevejgordon

public void Complete(Exception? error = null)
{
    if (!TryComplete(error))
    {
        throw ChannelUtilities.CreateInvalidCompletionException();
    }
}

stevejgordon

ChannelUtilities.FailOperations<AsyncOperation<T>, T>(parent._blockedReaders, ChannelUtilities.CreateInvalidCompletionException(error));
ChannelUtilities.WakeUpWaiters(ref parent._waitingReadersTail, result: false, error: error);

return true;

stevejgordon

internal static void Complete(TaskCompletionSource<VoidResult> tcs, Exception? error = null)
{
    if (error is OperationCanceledException oce)
    {
        tcs.TrySetCanceled(oce.CancellationToken);
    }
    else if (error != null && error != s_doneWritingSentinel)
    {
        tcs.TrySetException(error);
    }