Сборщик данных производителя с возможностью чтения и записи пакетов данных.

Question

Я ищу коллекцию типа BufferBlock

но с такими методами, как:

SendAsync<T>(T[])
T[] ReceiveAsync<T>()

Кто-нибудь может помочь?

Answer 1

Эти методы недоступны, SendAsync<T> принимает только один T, а RecieveAsync<T> возвращает только один T, а не массивы.

SendAsync<T>(T[])
T[] ReceiveAsync<T>()

Однако существует TryReceiveAll<T>(out IList<T> items) и вы можете вызвать SendAsync<T> в цикле, чтобы отправить массив в BufferBlock или написать свой собственный метод расширения, что-то вроде этого:

public static async Task SendAllAsync<T>(this ITargetBlock<T> block, IEnumerable<T> items)
{
    foreach(var item in items)
    {
         await block.SendAsync(item)
    }
}

Обратите внимание, что SendAsyncвозвращает логическое значение, указывающее на принятие сообщений, вы можете вернуть массив логических значений или просто вернуть, если какое-либо из них вернет false, но это ваше дело.

Скорее всего, было бы проще использовать BatchBlock<T>что вы можете отправлять элементы как одиночные, используя цикл, но испускайте элементы партиями, что было бы проще, чем использовать TryRecieveAll, если вы строите конвейер. BatchBlock Walkthrough и Пример BatchBlock

Answer 2

ReceiveAsync и SendAsync доступны в качестве методов расширения на интерфейсах ISourceBlock и ITargetBlockT <>. Это означает, что вы должны привести блок к этим интерфейсам, чтобы использовать методы расширения, например:

var buffer=new BufferBlock<string>();

var source=(ISourceBlock<string>)buffer;
var target=(ITargetBlock<string>)buffer;

await target.SendAsync("something");

Обычно это , а не проблема, потому что все методы потока данных принимают интерфейсы, а не конкретные типы, например:

async Task MyProducer(ITargetBlock<string> target)
{
    ...
    await target.SendAsync(..);
    ...
    target.Complete();
}

async Task MyConsumer(ISourceBlock<string> target)
{
    ...
    var message=await target.ReceiveAsync();
    ...
}

public static async Task Main()
{
    var buffer=new BufferBlock<string>();
    MyProducer(buffer);
    await MyConsumer(buffer);
}