scalaz/scalaz-stream — pull request #70 by pchlupacek (2013-12-31 12:15:41+00:00)
Title: MergeX

Ok @pchiusano, I needed for nio-streams bounded queue implementation. And hence this will again mean create new actor and all the orchestration around. I thought about it for a while, and actually created this MergeX combinator.

What MergeX does:

- provides MergeX algebra (see https://github.com/scalaz/scalaz-stream/compare/mergex2?expand=1#diff-17d2fc5501bfaed2ae5ef9d7c6231063R49) that allows generalize merging of streams that provide `I` (upstreams) and take `O` (downstreams). The `MergeXStrategy` is taking form of Process1, look here (https://github.com/scalaz/scalaz-stream/compare/mergex2?expand=1#diff-150b5a103571209b0d45191a4df61a12R11)

- with that algebra we can quite simply express mergeN, boundedQ, signal, topic just name it.

What this PR is about:
- MergeX implementation (is quite hairy now, but I`ll keep improving it)
- Introduced BoundedQueue
- Signal, Topic is implemented in terms of MergeX
- I have deprecated Ref, and just used Signal, hence I didn`t see any usages of Ref that can`t be expressed by signal. Also there were like 1-2 places where ref was actually used, and mainly these was tests. I can reintroduce it back, but I really disliked its signature A => Unit, and much prefer A => Task[Unit] rather.
- Implemented MergeN in terms of MergeX + added tests
- `Process.state` and `Process.awakeEvery` was reimplemented to use Signal instead of ref
- introduced `wye.echoLeft`
- added `toSignal` combinator to `async` to convert any discrete process to signal

What needs to be done
- I would like to implement queue as boundedQueue (that will require to deprecate `dequeue` method, although if we really need it it may be probably kept there. I would like to change Queue signature to be more like BoundedQueue, actually, again bacause of A => Unit.
- Cleanup implementation of MergeX a bit

Can you go through the code and share your thoughts?

Pavel.