Fix thread issue in consumer pipeline

There is a bug in the scheduler need to find it.  Causes double call of
do function on begin.
Add more tests and reorganized some files.