.NET_Framework

.NET Core.NET Framework Acronym Glossary.NET Framework ADO.NET.NET Framework CLR.NET Framework Code Contracts.NET Framework Collections.NET Framework Custom Types.NET Framework DateTime parsing.NET Framework Dependency Injection.NET Framework Dictionaries.NET Framework Encryption / Cryptography.NET Framework Exceptions.NET Framework Expression Trees.NET Framework File Input/Output.NET Framework ForEach.NET Framework Garbage Collection.NET Framework Globalization in ASP.NET MVC using Smart internationalization for ASP.NET.NET Framework HTTP clients.NET Framework HTTP servers.NET Framework Introduction.NET Framework JIT compiler.NET Framework JSON Serialization.NET Framework LINQ.NET Framework Managed Extensibility.NET Framework Memory management.NET Framework Networking.NET Framework NuGet packaging system.NET Framework Platform Invoke.NET Framework Process and Thread affinity setting.NET Framework Reading and writing Zip files.NET Framework ReadOnlyCollections.NET Framework Reflection.NET Framework Regular Expressions (System.Text.RegularExpressions).NET Framework Serial Ports.NET Framework Settings.NET Framework SpeechRecognitionEngine class to recognize speech.NET Framework Stack and Heap.NET Framework Strings.NET Framework Synchronization Contexts.NET Framework System.Diagnostics.NET Framework System.IO.NET Framework System.IO.File class.NET Framework System.Net.Mail.NET Framework System.Reflection.Emit namespace.NET Framework System.Runtime.Caching.MemoryCache (ObjectCache).NET Framework Task Parallel Library (TPL).NET Framework Task Parallel Library (TPL) API Overviews.NET Framework Threading.NET Framework TPL Dataflow.NET Framework Unit testing.NET Framework Upload file and POST data to webserver.NET Framework Using ProgressT and IProgressT.NET Framework VB Forms.NET Framework Work with SHA1 in C Sharp.NET Framework Write to and read from StdErr stream.NET Framework XmlSerializerJSON in .NET with Newtonsoft.JsonParallel processing using .NET framework



.NET Framework TPL Dataflow

From WikiOD

Remarks[edit | edit source]

Libraries Used in Examples[edit | edit source]

System.Threading.Tasks.Dataflow

System.Threading.Tasks

System.Net.Http

System.Net

Difference between Post and SendAsync[edit | edit source]

To add items to a block you can either use Post or SendAsync.

Post will try to add the item synchronously and return a bool saying whether it succeeded or not. It may not succeed when, for example, a block has reached its BoundedCapcity and has no more room for new items yet. SendAsync on the other hand will return an uncompleted Task<bool> that you can await. That task will complete in the future with a true result when the block cleared its internal queue and can accept more items or with a false result if it's declining permanently (e.g. as a result of cancellation).

Asynchronous Producer Consumer With A Bounded BufferBlock[edit | edit source]

var bufferBlock = new BufferBlock<int>(new DataflowBlockOptions
{
    BoundedCapacity = 1000
});

var cancellationToken = new CancellationTokenSource(TimeSpan.FromSeconds(10)).Token;

var producerTask = Task.Run(async () =>
{
    var random = new Random();

    while (!cancellationToken.IsCancellationRequested)
    {
        var value = random.Next();
        await bufferBlock.SendAsync(value, cancellationToken);
    }
});

var consumerTask = Task.Run(async () =>
{
    while (await bufferBlock.OutputAvailableAsync())
    {
        var value = bufferBlock.Receive();
        Console.WriteLine(value);
    }
});

await Task.WhenAll(producerTask, consumerTask);

Posting to an ActionBlock and waiting for completion[edit | edit source]

// Create a block with an asynchronous action
var block = new ActionBlock<string>(async hostName =>
{
    IPAddress[] ipAddresses = await Dns.GetHostAddressesAsync(hostName);
    Console.WriteLine(ipAddresses[0]);
});

block.Post("google.com"); // Post items to the block's InputQueue for processing
block.Post("reddit.com");
block.Post("stackoverflow.com");

block.Complete(); // Tell the block to complete and stop accepting new items
await block.Completion; // Asynchronously wait until all items completed processingu

Linking blocks to create a pipeline[edit | edit source]

var httpClient = new HttpClient();

// Create a block the accepts a uri and returns its contents as a string
var downloaderBlock = new TransformBlock<string, string>(
    async uri => await httpClient.GetStringAsync(uri));

// Create a block that accepts the content and prints it to the console
var printerBlock = new ActionBlock<string>(
    contents => Console.WriteLine(contents));

// Make the downloaderBlock complete the printerBlock when its completed.
var dataflowLinkOptions = new DataflowLinkOptions {PropagateCompletion = true};

// Link the block to create a pipeline
downloaderBlock.LinkTo(printerBlock, dataflowLinkOptions);

// Post urls to the first block which will pass their contents to the second one.
downloaderBlock.Post("http://youtube.com");
downloaderBlock.Post("http://github.com");
downloaderBlock.Post("http://twitter.com");

downloaderBlock.Complete(); // Completion will propagate to printerBlock
await printerBlock.Completion; // Only need to wait for the last block in the pipeline

Synchronous Producer/Consumer with BufferBlock[edit | edit source]

public class Producer
{
    private static Random random = new Random((int)DateTime.UtcNow.Ticks);
    //produce the value that will be posted to buffer block
    public double Produce ( )
    {
        var value = random.NextDouble();
        Console.WriteLine($"Producing value: {value}");
        return value;
    }
}

public class Consumer
{
    //consume the value that will be received from buffer block
    public void Consume (double value) => Console.WriteLine($"Consuming value: {value}");
}

class Program
{
    private static BufferBlock<double> buffer = new BufferBlock<double>();
    static void Main (string[] args)
    {
        //start a task that will every 1 second post a value from the producer to buffer block
        var producerTask = Task.Run(async () =>
        {
            var producer = new Producer();
            while(true)
            {
                buffer.Post(producer.Produce());
                await Task.Delay(1000);
            }
        });
        //start a task that will recieve values from bufferblock and consume it
        var consumerTask = Task.Run(() => 
        {
            var consumer = new Consumer();
            while(true)
            {
                consumer.Consume(buffer.Receive());
            }
        });

        Task.WaitAll(new[] { producerTask, consumerTask });
    }
}

Credit:Stack_Overflow_Documentation