func createJobTask(
	client publisher.Client,
	name string,
	r monitors.TaskRunner,
) scheduler.TaskFunc {
	return func() []scheduler.TaskFunc {
		event, next, err := r.Run()
		if err != nil {
			logp.Err("Job %v failed with: ", err)
		}

		if event != nil {
			event["monitor"] = name
			client.PublishEvent(event)
		}

		if len(next) == 0 {
			return nil
		}

		cont := make([]scheduler.TaskFunc, len(next))
		for i, n := range next {
			cont[i] = createJobTask(client, name, n)
		}
		return cont
	}
}

func (w *worker) run(client publisher.Client) {
	if *max > 0 {
		m := *max
		for w.running() && m >= 0 {
			event := w.gen()
			client.PublishEvent(event)
			m--
		}
	} else {
		for w.running() {
			event := w.gen()
			client.PublishEvent(event)
		}
	}

}

// PublishChannels publishes the events read from each channel to the given
// publisher client. If the publisher client blocks for any reason then events
// will not be read from the given channels.
//
// This method blocks until all of the channels have been closed
// and are fully read. To stop the method immediately, close the channels and
// close the publisher client to ensure that publishing does not block. This
// may result is some events being discarded.
func PublishChannels(client publisher.Client, cs ...<-chan common.MapStr) {
	var wg sync.WaitGroup

	// output publishes values from c until c is closed, then calls wg.Done.
	output := func(c <-chan common.MapStr) {
		defer wg.Done()
		for event := range c {
			client.PublishEvent(event)
		}
	}

	// Start an output goroutine for each input channel in cs.
	wg.Add(len(cs))
	for _, c := range cs {
		go output(c)
	}

	wg.Wait()
}

func (w *worker) run(client publisher.Client) {
	ticker := time.NewTicker(w.period)
	for {
		stats := map[string]float64{}
		now := time.Now()
		last := now

		for {
			select {
			case <-w.done:
				return
			case <-ticker.C:
			}

			data, err := w.readStats()
			if err != nil {
				logp.Warn("Failed to read vars from %v: %v", w.host, err)
				break
			}

			last = now
			now = time.Now()
			dt := now.Sub(last).Seconds()

			event := common.MapStr{
				"@timestamp": common.Time(now),
				"type":       w.name,
				"remote":     w.host,
			}
			for name, v := range data {
				if old, ok := stats[name]; ok {
					event[fmt.Sprintf("d_%s", name)] = (v - old) / dt
				}
				event[fmt.Sprintf("total_%s", name)] = v
				stats[name] = v
			}
			client.PublishEvent(event)
		}
	}
}

func Scanner(config config.PathConfig, client publisher.Client) {

	duration, _ := time.ParseDuration(config.Period)

	ticker := time.NewTicker(duration)
	defer ticker.Stop()

	for {
		select {

		case <-ticker.C:
		}

		events := []common.MapStr{}

		logp.Debug("mod", "Scanning glob: %s", config.Glob)

		files, _ := filepath.Glob(config.Glob)
		for _, file := range files {
			logp.Debug("mod", "Get stats for file: %s", file)
			stat, _ := os.Lstat(file)

			event := common.MapStr{
				"@timestamp": common.Time(time.Now()),
				"type":       "modbeat",
				"name":       stat.Name(),
				"path":       file,
				"mode":       stat.Mode(),
				"modtime":    stat.ModTime(),
				"size":       stat.Size(),
			}
			events = append(events, event)
		}
		client.PublishEvents(events)
		logp.Debug("mod", "Events sent for %s files", len(events))
	}
}

func (w *worker) run(client publisher.Client) {
	for w.running() {
		event := w.gen()
		client.PublishEvent(event)
	}
}

func (w *worker) run(client publisher.Client) {
	ticker := time.NewTicker(w.period)
	for {
		stats := make(map[string]float64)
		now := time.Now()
		last := now
		for {
			select {
			case <-w.done:
				return
			case <-ticker.C:
			}

			resp, err := http.Get(fmt.Sprintf("http://%s/debug/vars", w.host))
			if err != nil {
				logp.Info("Failed to retrieve variables: %v", err)
				break
			}

			body, err := ioutil.ReadAll(resp.Body)
			resp.Body.Close()
			if err != nil {
				logp.Info("Error reading response: %v", err)
				break
			}

			last = now
			now = time.Now()
			dt := now.Sub(last).Seconds()

			var data map[string]interface{}
			err = json.Unmarshal(body, &data)
			if err != nil {
				logp.Warn("Failed to decode json: %v", err)
				break
			}

			event := common.MapStr{
				"@timestamp": common.Time(now),
				"type":       w.name,
			}
			for name, raw := range data {
				has := true
				var total float64
				switch v := raw.(type) {
				case int:
					total = float64(v)
				case float64:
					total = v
				default:
					has = false
				}
				if !has {
					continue
				}

				if old, ok := stats[name]; ok {
					event[fmt.Sprintf("total_%s", name)] = total
					event[fmt.Sprintf("d_%s", name)] = (total - old) / dt
				}
				stats[name] = total
			}
			client.PublishEvent(event)
		}
	}
}