func (d *dispatcher) Run(quit chan struct{}) {
	defer close(d.submitCh)
	d.log(nil).Info("starting dispatcher")
	for {
		select {
		case task := <-d.submitCh:
			fields := log.Fields{
				"ID":     task.ID(),
				"Status": task.Status(),
			}
			d.log(fields).Debug("received task")
			if err := task.Run(d.attackFn); err != nil {
				d.log(fields).WithError(err).Errorf("failed to run %s", task.ID())
				continue
			}
		case update := <-d.updateCh:
			fields := log.Fields{
				"ID":     update.ID,
				"Status": update.Status,
			}
			d.mu.RLock()
			task := d.tasks[update.ID]
			d.mu.RUnlock()
			if err := d.db.Update(task.ID(), attackDetailFromTask(task)); err != nil {
				d.log(fields).WithError(err).Error("attack update error")
				continue
			}
			d.log(fields).Debug("received update for attack")
		case <-quit:
			for _, task := range d.tasks {
				_ = task.Cancel()
			}
			d.log(nil).Warning("gracefully shutting down the dispatcher")
			return
		}
	}
}
Run the dispatcher event loop to dispatch new attacks,
receive status updates for scheduled attacks and update the
storage.
package dispatcher
import (
	"fmt"
	"vegeta-server/pkg/vegeta"
	log "github.com/sirupsen/logrus"
	"sync"
	"vegeta-server/models"
	"github.com/pkg/errors"
)
var (
	defaultDB       = models.NewTaskMap()
	defaultAttackFn = vegeta.Attack
)
type IDispatcher interface {
	Run(chan struct{})
	Dispatch(models.AttackParams) (*models.AttackResponse, error)
	Cancel(string, bool) error
	Get(string) (*models.AttackResponse, error)
	List(models.FilterParams) []*models.AttackResponse
}
type dispatcher struct {
	mu       *sync.RWMutex
	tasks    map[string]ITask
	attackFn AttackFunc
	submitCh chan ITask
	updateCh chan UpdateMessage
	db       models.IAttackStore
}
func NewDispatcher(db models.IAttackStore, fn AttackFunc) *dispatcher {  
	if db == nil {
		db = defaultDB
	}
	if fn == nil {
		fn = defaultAttackFn
	}
	d := &dispatcher{
		&sync.RWMutex{},
		make(map[string]ITask),
		fn,
		make(chan ITask, 10),
		make(chan UpdateMessage, 20),
		db,
	}
	d.log(nil).Info("creating new dispatcher")
	return d
}
func (d *dispatcher) Dispatch(params models.AttackParams) (*models.AttackResponse, error) {
	task := NewTask(d.updateCh, params)
	id := task.ID()
	status := task.Status()
	fields := log.Fields{
		"ID":     id,
		"Status": status,
	}
	d.mu.Lock()
	d.tasks[task.ID()] = task
	d.mu.Unlock()
	_ = d.db.Add(attackDetailFromTask(task))
	d.log(fields).Info("dispatching new attack")
	d.submitCh <- task
	attackDetails, err := d.db.GetByID(id)
	if err != nil {
		return nil, errors.Wrap(err, "failed to get attack by ID")
	}
	resp := models.AttackResponse(attackDetails.AttackInfo)
	return &resp, nil
}