/* ========================================================================
* Copyright (c) 2005-2025 The OPC Foundation, Inc. All rights reserved.
*
* OPC Foundation MIT License 1.00
*
* Permission is hereby granted, free of charge, to any person
* obtaining a copy of this software and associated documentation
* files (the "Software"), to deal in the Software without
* restriction, including without limitation the rights to use,
* copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the
* Software is furnished to do so, subject to the following
* conditions:
*
* The above copyright notice and this permission notice shall be
* included in all copies or substantial portions of the Software.
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
* EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
* OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
* NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
* HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
* WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
* FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
* OTHER DEALINGS IN THE SOFTWARE.
*
* The complete license agreement can be found here:
* http://opcfoundation.org/License/MIT/1.00/
* ======================================================================*/
#if !NET8_0_OR_GREATER
using System.Collections.Generic;
using System.Threading.Tasks;
#endif
namespace System.Threading.Channels
{
#if !NET8_0_OR_GREATER
///
/// Options for .
/// Polyfill for UnboundedPrioritizedChannelOptions
/// which is only available in .NET 9+.
///
///
internal sealed class UnboundedPrioritizedChannelOptions
{
///
/// Whether only a single reader will access the channel.
///
public bool SingleReader { get; set; }
///
/// The comparer used to order items by priority.
///
public IComparer? Comparer { get; set; }
}
///
/// Polyfill for Channel.CreateUnboundedPrioritized on
/// runtimes before .NET 9. Uses an unbounded channel with a
/// priority queue that re-orders items on read.
///
internal static class PrioritizedChannelHelper
{
extension(Channel)
{
///
/// Create an unbounded channel that reads items in priority
/// order according to the provided comparer.
///
///
public static Channel CreateUnboundedPrioritized(
UnboundedPrioritizedChannelOptions options)
{
return new PrioritizedChannel(
options.Comparer ?? Comparer.Default);
}
}
}
///
/// A channel implementation that stores items in a priority
/// queue, yielding them in priority order on read. This is a
/// polyfill for the .NET 9 prioritized channel API.
///
///
internal sealed class PrioritizedChannel : Channel, IDisposable
{
///
/// Create channel with comparer
///
public PrioritizedChannel(IComparer comparer)
{
m_comparer = comparer;
m_heap = [];
m_semaphore = new SemaphoreSlim(0);
Writer = new PrioritizedWriter(this);
Reader = new PrioritizedReader(this);
}
///
public void Dispose()
{
m_semaphore.Dispose();
}
private void HeapPush(T item)
{
m_heap.Add(item);
int i = m_heap.Count - 1;
while (i > 0)
{
int parent = (i - 1) / 2;
if (m_comparer.Compare(m_heap[i], m_heap[parent]) >= 0)
{
break;
}
(m_heap[i], m_heap[parent]) = (m_heap[parent], m_heap[i]);
i = parent;
}
}
private T HeapPop()
{
T result = m_heap[0];
int last = m_heap.Count - 1;
m_heap[0] = m_heap[last];
m_heap.RemoveAt(last);
last--;
int i = 0;
while (true)
{
int left = (2 * i) + 1;
int right = (2 * i) + 2;
int smallest = i;
if (left <= last &&
m_comparer.Compare(m_heap[left], m_heap[smallest]) < 0)
{
smallest = left;
}
if (right <= last &&
m_comparer.Compare(m_heap[right], m_heap[smallest]) < 0)
{
smallest = right;
}
if (smallest == i)
{
break;
}
(m_heap[i], m_heap[smallest]) = (m_heap[smallest], m_heap[i]);
i = smallest;
}
return result;
}
private sealed class PrioritizedWriter : ChannelWriter
{
private readonly PrioritizedChannel m_channel;
public PrioritizedWriter(PrioritizedChannel channel)
{
m_channel = channel;
}
public override bool TryWrite(T item)
{
lock (m_channel.m_lock)
{
if (m_channel.m_completed)
{
return false;
}
m_channel.HeapPush(item);
}
m_channel.m_semaphore.Release();
return true;
}
public override ValueTask WaitToWriteAsync(
CancellationToken ct = default)
{
return m_channel.m_completed
? new ValueTask(false)
: new ValueTask(true);
}
public override bool TryComplete(Exception? error = null)
{
lock (m_channel.m_lock)
{
if (m_channel.m_completed)
{
return false;
}
m_channel.m_completed = true;
}
m_channel.m_semaphore.Release();
return true;
}
}
private sealed class PrioritizedReader : ChannelReader
{
private readonly PrioritizedChannel m_channel;
public PrioritizedReader(PrioritizedChannel channel)
{
m_channel = channel;
}
///
public override bool CanCount => true;
///
public override int Count
{
get
{
lock (m_channel.m_lock)
{
return m_channel.m_heap.Count;
}
}
}
public override bool TryRead(out T item)
{
lock (m_channel.m_lock)
{
if (m_channel.m_heap.Count > 0)
{
item = m_channel.HeapPop();
return true;
}
}
item = default!;
return false;
}
public override async ValueTask WaitToReadAsync(
CancellationToken ct = default)
{
while (true)
{
lock (m_channel.m_lock)
{
if (m_channel.m_heap.Count > 0)
{
return true;
}
if (m_channel.m_completed)
{
return false;
}
}
await m_channel.m_semaphore.WaitAsync(ct)
.ConfigureAwait(false);
// Consumed the semaphore signal — check again.
// If an item was added, we'll find it in the heap.
// If completed, we return false.
lock (m_channel.m_lock)
{
if (m_channel.m_heap.Count > 0)
{
return true;
}
if (m_channel.m_completed)
{
return false;
}
}
}
}
}
private readonly IComparer m_comparer;
private readonly List m_heap;
private readonly SemaphoreSlim m_semaphore;
private readonly object m_lock = new();
private bool m_completed;
}
#endif
}