The Demux Pipe

The pipes library has gone through an in-depth refactoring to become what it is now, and one of the components that changed the most is the demultiplexer, a.k.a. demux pipe.

I think this refactoring illustrates two principles or phenomena that we observe in software refactoring: Single Responsibility Principle and Refactoring breakthrough.

They contributed to make the code simpler, clearer and more modular. Let’s reflect on how that happened, in order to get inspiration for future refactoring projects.

EDIT: The demux pipe of the pipes library has been renamed into fork. Thanks to Arno Schödl for this insight.

The old demux

As a reminder, the goal of demux was to send data to several outputs:

std::copy(begin(inputs), end(inputs),
    demux(demux_if(predicate1).send_to(back_inserter(v1)),
          demux_if(predicate2).send_to(back_inserter(v2)),
          demux_if(predicate3).send_to(back_inserter(v3))));

Every piece of data that is sent to demux by the STL algorithm is checked by predicate1. If predicate1 returns true then the data is sent on to back_inserter(v1), and that’s it.

If predicate1 returns false, then the value is checked by predicate2. If it returns true it gets sent to back_inserter(v2). And so on with predicate3.

And if none of the three predicates returned true, then the data is not sent anywhere.

demux can be combined with other components of the library to create elaborate treatments of the incoming data:

std::copy(begin(inputs), end(inputs),
    demux(demux_if(predicate1).send_to(transform(f) >>= back_inserter(v1)),
          demux_if(predicate2).send_to(filter(p) >>= back_inserter(v2)),
          demux_if(predicate3).send_to(begin(v3))));

What is wrong with demux

We had already talked about this initial version of demux in a previous post, and you, readers of Fluent C++, reacted to its design by leaving comments.

I’m so grateful for those comments. They helped point out what didn’t make sense in that version of demux, and how it could be improved.

The first pointed flaws of that demux is that it only sends the data to the first branch that matches. If several branches match, they won’t all get the data. That can be what you want or not, depending on the situation. It would be nice to be able to select one of the two behaviours: first that matches or all that match.

Another issue is that there is no “default” clause, to ensure that the incoming piece of data goes somewhere even if all the predicates return false.

The last problem is the syntax. It would be nice to simplify the cumbersome demux(demux_if(predicate1).send_to(back_inserter(v1).

Let’s see how to remedy to those three issues.

Sending data to several directions

The pipes library wasn’t always called that way. It used to be called Smart Output Iterators. Its transformation into pipes was a refactoring breakthrough, in the sense that it sheds a new light on how to represent the components of the library.

The concept of refactoring breakthrough is explained in more detail the Domain Driven Design book.

The initial intent of demux was to send data to several directions. The analogy with plumbing of the intent of sending data to all directions looks like this:

In the above picture, fluid pours in on the left hand side and comes out on the three pipes on the right.

In this vision, demux should send to all branches, and there is not even a notion of predicate.

Then if we want to filter with predicates, we can always tack on some filter pipes:

This assembly of pipes sends the incoming data to all outputs that match.

Its equivalent in code would look like this:

demux(filter(predicate1) >>= back_inserter(v1),
      filter(predicate2) >>= back_inserter(v2),
      filter(predicate3) >>= back_inserter(v3));

Now demux has only one responsibility, sending the same piece of data to all its output pipes. The responsibility of checking a predicate is left to the good old filter, who is focused on this responsibility solely.

This is an application of the Single Responsibility Principle, and as a result the syntax has become much simpler.

Implementation of the new demux

The implementation of demux becomes very simple. The pipe contains a std::tuple of the output pipes to which it needs to send the data. It loops over them with the for_each algorithm on tuples, and sends the incoming value to each one of them:

template<typename T>
void onReceive(T&& value)
{
    for_each(outputPipes_, [&value](auto&& outputPipe){ send(outputPipe, value); });
}

And that’s all for demux.

Sending to the first one that matches

Now we have a demux pipe that sends to all outputs, and we can combine it with other pipes such as filter to add predicates to the branches.

But what if we do need to send data only to the first branch that matches?

I can’t see how demux can do that, because it always sends to all branches, and each branch doesn’t know what happened in the other branches.

So we’re back to the old version of demux, that sends to the first branch that matches.

We can do three things to improve it though:

• give it another name,
• lighten its syntax,
• include a “default” branch that gets used if all the other predicates return false.

A new name

What to call a component that activates one of several branches depending on an incoming value?

One of the suggestions was to use the words “switch” and “case”, like the native constructs of C++ (and of several other languages).

Let’s see what the renaming looks like. The previous version of demux looked like this:

demux(demux_if(predicate1).send_to(back_inserter(v1)),
      demux_if(predicate2).send_to(back_inserter(v2)),
      demux_if(predicate3).send_to(back_inserter(v3)));

With the new names it looks like this:

switch_(case_(predicate1).send_to(back_inserter(v1)),
        case_(predicate2).send_to(back_inserter(v2)),
        case_(predicate3).send_to(back_inserter(v3)));

A lighter syntax

The above code has already become more understandable. But we can also make the syntax more idiomatic to the library, by using the operator>>= instead of a class method called “send_to”:

switch_(case_(predicate1) >>= back_inserter(v1),
        case_(predicate2) >>= back_inserter(v2),
        case_(predicate3) >>= back_inserter(v3));

There is less noise, less parentheses and a better consistency with the rest of the library.

We’re skipping over the implementation of this here, because its has the same technical aspects as the initial demux iterator.

A default branch

Finally, we want to add a branch that offers a fallback option in case none of the predicates of the case_ branches return true. To be consistent with switch_ and case_, let’s call it default_.

Its implementation is very straightforward: default_ is merely a case_ branch with a predicate that always returns true:

auto const default_ = case_([](auto&&){ return true; });

We can now use it this way:

switch_(case_(predicate1) >>= back_inserter(v1),
        case_(predicate2) >>= back_inserter(v2),
        case_(predicate3) >>= back_inserter(v3),
        default_ >>= back_inserter(v4));

If switch_ receives a value for which predicate1, predicate2 and predicate3 return false, then that value will be sent to v4.

Like all pipes, switch_ can be the output of an STL algorithm:

std::set_difference(begin(input1), end(input1),
                    begin(input2), end(input2),
                    switch_(case_(predicate1) >>= back_inserter(v1),
                            case_(predicate2) >>= back_inserter(v2),
                            case_(predicate3) >>= back_inserter(v3),
                            default_ >>= back_inserter(v4));

Or we can send the data of a range or an STL container by using funnel:

inputs >>= funnel
       >>= switch_(case_(predicate1) >>= back_inserter(v1),
                   case_(predicate2) >>= back_inserter(v2),
                   case_(predicate3) >>= back_inserter(v3),
                   default_ >>= back_inserter(v4));

Or it can be an output of another pipe:

inputs >>= funnel
       >>= transform(f)
       >>= switch_(case_(predicate1) >>= back_inserter(v1),
                   case_(predicate2) >>= back_inserter(v2),
                   case_(predicate3) >>= back_inserter(v3),
                   default_ >>= back_inserter(v4));

Refactoring pipes

We’ve seen how the concepts of refactoring breakthrough and single responsibility principle helped refactor the demux pipes into two components of the pipes library. Those two components are arguably clearer thanks to this change.

Would you have gone differently about a part of this refactoring?

Can you think of other pipes you would like to add to the library?

Leave a comment below to let me know.

You will also like

• The Demultiplexer Iterator: Routing Data to Any Numbers of Outputs
• STL Algorithms on Tuples
• Write Your Own Dependency-Injection Container
• Smart Output Iterators >>= become(Pipes)

Don't want to miss out ? Follow:   
Share this post!