.NET 4.5 Baby Steps, Part 3: IDataFlowBlock

introduction

a new interface in .net is the idataflowblock, which is implemented in many interesting ways, so to look at those we will start off with the simplest implementation. actionblock<tinput> is a completely new class in .net 4.5 and provides a way of working with data in a very task orientated way. i simplistically think of this as the implementation of the iobserver interface we got in .net 4. but do not limit your thinking to just that.

to use it, you first must add a reference to system.threading.tasks.dataflow

in this simple first example i am doing a fairly simple pub/sub demo:

var subscriber = new actionblock<string>(input =>
    {
        console.writeline("got: {0}", input);
    });

for (int i = 0; i < 10; i++)
{
    task.factory.startnew(() =>
        {
            thread.sleep(new random().next(200, 1000));
            subscriber.post(datetime.now.tolongtimestring());
        });
}

console.readline();

as iobserver<t>

so the first fantastic feature is that it does have the ability (via extension method) to be an iobsserver<t> so it really solves the need to build up your own subscriber classes when implementing a pub/sub model.

first is the code for the publisher class – this is normal for the iobservable<t> as we had in .net 4. this just means our new code can play well with our existing code.

public class publisher : iobservable<string>
{
    list<iobserver<string>> subscribers = new list<iobserver<string>>();

    public idisposable subscribe(iobserver<string> observer)
    {
        subscribers.add(observer);
        return null;
    }

    public void send()
    {
        foreach (var item in subscribers)
        {
            item.onnext(datetime.now.tolongtimestring());
        }
    }
}

for our demo code, which produces the same as above:

var publisher = new publisher();

var subscriber = new actionblock<string>(input =>
    {
        console.writeline("got: {0}", input);
    });

publisher.subscribe(subscriber.asobserver());

for (int i = 0; i < 10; i++)
{
    task.factory.startnew(() =>
        {
            thread.sleep(new random().next(200, 1000));
            publisher.send();
        });
}

complete

the next awesome feature is the complete method which can be used to stop accepting of input when called – this is great for services where you want to shut down.

in this demo code it will run until you press enter:

var subscriber = new actionblock<string>(input =>
{
    console.writeline("got: {0}", input);
});


task.factory.startnew(() =>
{
    while (true)
    {

        thread.sleep(new random().next(200, 1000));
        subscriber.post(datetime.now.tolongtimestring());
    }
});

console.writeline("press any key to stop input");
console.readline();
subscriber.complete();