sampler-fork/data/sampler.go

package data

import (
	"fmt"
	"github.com/sqshq/sampler/config"
	"time"
)

type Sampler struct {
	consumer        *Consumer
	items           []*Item
	triggers        []*Trigger
	triggersChannel chan *Sample
	variables       []string
	pause           bool
}

func NewSampler(consumer *Consumer, items []*Item, triggers []*Trigger, options config.Options, fileVariables map[string]string, rateMs int) *Sampler {

	ticker := time.NewTicker(time.Duration(rateMs) * time.Millisecond)

	sampler := &Sampler{
		consumer,
		items,
		triggers,
		make(chan *Sample),
		mergeVariables(fileVariables, options.Environment),
		false,
	}

	go func() {
		for ; true; <-ticker.C {
			for _, item := range sampler.items {
				if !sampler.pause {
					go sampler.sample(item, options)
				}
			}
		}
	}()

	go func() {
		for {
			select {
			case sample := <-sampler.triggersChannel:
				for _, t := range sampler.triggers {
					if !sampler.pause {
						t.Execute(sample)
					}
				}
			}
		}
	}()

	return sampler
}

func (s *Sampler) sample(item *Item, options config.Options) {

	val, err := item.nextValue(s.variables)

	if len(val) > 0 {
		sample := &Sample{Label: item.label, Value: val, Color: item.color}
		s.consumer.SampleChannel <- sample
		s.triggersChannel <- sample
	} else if err != nil {
		s.consumer.AlertChannel <- &Alert{
			Title:       "Sampling failure",
			Text:        getErrorMessage(err),
			Color:       item.color,
			Recoverable: true,
		}
	}
}

// option variables takes precedence over the file variables with the same name
func mergeVariables(fileVariables map[string]string, optionsVariables []string) []string {

	result := optionsVariables

	for key, value := range fileVariables {
		result = append([]string{fmt.Sprintf("%s=%s", key, value)}, result...)
	}

	return result
}

func (s *Sampler) Pause(pause bool) {
	s.pause = pause
}