Composability

Composability is improved by using stream fragments and wiring together the stream before turning the fragment into a stream and running it. Your application should be created using fragments connected with pipes. You use operations like map and filter to wrap a base stream fragment with your domain specific logic then connect the resulting fragments together to perform a sequence of operations. Either wrapping a base Stream object using map or connecting fragments together using pipe should be seen as applying domain specific knowledge. In fact, many methods on Streams are implemented using pipes.

Each fragment can be parameterized and returned in a function that you author. This allows the fragment to be customized and made more reusable. For example, if the fragment is to apply a string transformation to a specific field, the parameter may be the suffix to append:

def addSuffix(fragment: Process1[String, String], suffix: String) = fragment.map(_ + suffix)

These types of fragments are useful but may not as expressive as you would like. You could write the above differently:

def addSuffix(suffix: String) = pipe.lift(base:String => base + suffix)

This approach should be preferred because it allows a more visually decoupled syntax using the pipe operations e.g. Stream.emits(Seq("blah", "hah")).through(addSuffix("-nah") ... versus addSuffix(Stream.emits(...)) ....