func NewTaskManager(coordinator Coordinator, receiver InputManager, taskStarter TaskStarter) TaskManager {
	c := metrics.NewCounter()
	metrics.Register("processing", c)
	tm := &taskManager{
		tasks:        make(map[StreamID]Task),
		coordinator:  coordinator,
		receiver:     receiver,
		taskStarter:  taskStarter,
		counter:      c,
		done:         make(chan struct{}),
		taskFailedCh: make(chan *TaskInfo),
	}
	receiver.SetTaskManager(tm)
	return tm
}
NewTaskManager creates a new task manager
package dagger
import (
	"fmt"
	log "github.com/Sirupsen/logrus"
	"math/rand"
	"strings"
	"github.com/rcrowley/go-metrics"
)
type Task interface {
	RecordProcessor
	GetSnapshot() ([]byte, error)
	Sync() (Timestamp, error)
	Run(chan error)
	Stop()
}
type TaskManager interface {
	ManageTasks() error
	GetTaskSnapshot(StreamID) ([]byte, error)
	Stop()
}
type taskManager struct {
	tasks        map[StreamID]Task
	coordinator  Coordinator
	receiver     InputManager
	taskStarter  TaskStarter
	done         chan struct{}
	taskFailedCh chan *TaskInfo
	counter metrics.Counter
}