qb/mqe/mqe.go

package mqe

import (
	"sync"
	"time"

	"gordenko.dev/dima/qx"
)

type GroupBy string

// Для парсинга since и until
const (
	NoAggregateFunc byte = 1
	AggregateMin    byte = 2
	AggregateMax    byte = 4
	AggregateAvg    byte = 8

	ByHour  GroupBy = "h"
	ByDay   GroupBy = "d"
	ByMonth GroupBy = "m"

	hourPeriodLayout  = "2006010215"
	dayPeriodLayout   = "20060102"
	monthPeriodLayout = "200601"

	dateLayout = "2006-01-02"
)

type MetricMeasuresFilter struct {
	MetricID int64 `json:"metricID"`
	Since    int64 `json:"since"` // уже учтен firstHourOfDay
	Until    int64 `json:"until"` // уже учтен firstHourOfDay
}

type MeasureQueryEngine struct {
	db                     *qx.Db
	location               *time.Location
	cumulativeCh           chan *CumulativeTask
	instantCh              chan *InstantTask
	aggregatedCumulativeCh chan *AggregatedCumulativeTask
	aggregatedInstantCh    chan *AggregatedInstantTask
	cumulativeTotalCh      chan *CumulativeTotalTask
	metricReadingsCh       chan *MetricReadingsTask
}

type Options struct {
	Db         *qx.Db
	Location   *time.Location
	MinWorkers int
	QueueSize  int
}

func New(opt Options) *MeasureQueryEngine {
	if opt.Db == nil {
		panic("Db option is required")
	}

	if opt.Location == nil {
		panic("Location option is required")
	}

	if opt.MinWorkers <= 0 {
		panic("MinWorkers option is required")
	}

	if opt.QueueSize <= 0 {
		panic("QueueSize option is required")
	}

	s := new(MeasureQueryEngine)
	s.db = opt.Db
	s.location = opt.Location
	//
	s.cumulativeCh = make(chan *CumulativeTask, opt.QueueSize)
	s.instantCh = make(chan *InstantTask, opt.QueueSize)
	s.aggregatedCumulativeCh = make(chan *AggregatedCumulativeTask, opt.QueueSize)
	s.aggregatedInstantCh = make(chan *AggregatedInstantTask, opt.QueueSize)
	s.cumulativeTotalCh = make(chan *CumulativeTotalTask, opt.QueueSize)
	s.metricReadingsCh = make(chan *MetricReadingsTask, opt.QueueSize)

	//
	for i := 0; i < opt.MinWorkers; i++ {
		go s.worker()
	}
	return s
}

func (s *MeasureQueryEngine) worker() {
	for {
		select {
		case task := <-s.cumulativeCh:
			task.Result, task.Err = s.listCumulativeMeasures(task.In)
			task.WaitGroup.Done()

		case task := <-s.instantCh:
			task.Result, task.Err = s.listInstantMeasures(task.In)
			task.WaitGroup.Done()

		case task := <-s.aggregatedCumulativeCh:
			task.Result, task.Err = s.listAggregatedCumulativeMeasures(task.In)
			task.WaitGroup.Done()

		case task := <-s.aggregatedInstantCh:
			task.Result, task.Err = s.listAggregatedInstantMeasures(task.In)
			task.WaitGroup.Done()

		case task := <-s.cumulativeTotalCh:
			task.Result, task.Err = s.getCumulativeTotal(task.In)
			task.WaitGroup.Done()

		case task := <-s.metricReadingsCh:
			task.Result, task.Err = s.listMetricReadings(task.In)
			task.WaitGroup.Done()
		}
	}
}

//

type CumulativeTask struct {
	In        MetricMeasuresFilter
	WaitGroup *sync.WaitGroup
	Result    []Measure
	Err       error
}

type InstantTask struct {
	In        MetricMeasuresFilter
	WaitGroup *sync.WaitGroup
	Result    []Measure
	Err       error
}

type AggregatedCumulativeTask struct {
	In        AggregatedMeasuresFilter
	WaitGroup *sync.WaitGroup
	Result    []AggregatedMeasure
	Err       error
}

type AggregatedInstantTask struct {
	In        AggregatedInstantMeasuresIn
	WaitGroup *sync.WaitGroup
	Result    []AggregatedMeasure
	Err       error
}

type MetricReadingsTask struct {
	In        MetricMeasuresFilter
	WaitGroup *sync.WaitGroup
	Result    MetricReadings
	Err       error
}

type CumulativeTotalTask struct {
	In        getCumulativeTotalFilter
	WaitGroup *sync.WaitGroup
	Result    *RangeTotal
	Err       error
}