169 lines
3.5 KiB
Go
169 lines
3.5 KiB
Go
package mqe
|
|
|
|
import (
|
|
"database/sql"
|
|
"fmt"
|
|
"time"
|
|
)
|
|
|
|
type periodProducer interface {
|
|
Next() (time.Time, PeriodBound, PeriodBound, bool, error)
|
|
Last() (PeriodBound, bool, error)
|
|
}
|
|
|
|
type PeriodBound struct {
|
|
Time int64
|
|
Value float64
|
|
}
|
|
|
|
type PeriodProducer struct {
|
|
metricID int64
|
|
parsePeriodLayout string
|
|
location *time.Location
|
|
until int64
|
|
corrections []_f64Correction
|
|
rows *sql.Rows
|
|
tx *sql.Tx
|
|
}
|
|
|
|
type PeriodProducerOptions struct {
|
|
MetricID int64
|
|
ParsePeriodLayout string
|
|
Location *time.Location
|
|
Until int64
|
|
Corrections []_f64Correction
|
|
Rows *sql.Rows
|
|
Tx *sql.Tx
|
|
}
|
|
|
|
func NewPeriodProducer(opt PeriodProducerOptions) *PeriodProducer {
|
|
if opt.Location == nil {
|
|
panic("Location option is required")
|
|
}
|
|
|
|
if opt.MetricID == 0 {
|
|
panic("MetricID option is required")
|
|
}
|
|
|
|
if opt.ParsePeriodLayout == "" {
|
|
panic("ParsePeriodLayout option is required")
|
|
}
|
|
|
|
if opt.Until == 0 {
|
|
panic("Until option is required")
|
|
}
|
|
|
|
if opt.Rows == nil {
|
|
panic("Rows option is required")
|
|
}
|
|
|
|
if opt.Tx == nil {
|
|
panic("Tx option is required")
|
|
}
|
|
|
|
s := new(PeriodProducer)
|
|
s.metricID = opt.MetricID
|
|
s.parsePeriodLayout = opt.ParsePeriodLayout
|
|
s.location = opt.Location
|
|
s.until = opt.Until
|
|
s.corrections = opt.Corrections
|
|
s.rows = opt.Rows
|
|
s.tx = opt.Tx
|
|
return s
|
|
}
|
|
|
|
func (s *PeriodProducer) Next() (period time.Time, first PeriodBound, last PeriodBound, found bool, err error) {
|
|
var frameStr string
|
|
|
|
hasNext := s.rows.Next()
|
|
if hasNext {
|
|
err = s.rows.Scan(&frameStr, &first.Time, &first.Value, &last.Time, &last.Value)
|
|
if err != nil {
|
|
return
|
|
}
|
|
|
|
first.Value = applyCorrectionsToMeasure(s.corrections, first.Time, first.Value)
|
|
last.Value = applyCorrectionsToMeasure(s.corrections, last.Time, last.Value)
|
|
|
|
period, err = time.ParseInLocation(s.parsePeriodLayout, frameStr, s.location)
|
|
if err != nil {
|
|
err = fmt.Errorf("time.ParseInLocation: %s; layout=%q; str=%q",
|
|
err, s.parsePeriodLayout, frameStr)
|
|
return
|
|
}
|
|
|
|
found = true
|
|
} else {
|
|
err = s.rows.Err()
|
|
}
|
|
return
|
|
}
|
|
|
|
func (s *PeriodProducer) Last() (next PeriodBound, found bool, err error) {
|
|
// Находим следующее показание метрики.
|
|
err = s.tx.QueryRow(`
|
|
SELECT tm, value, true
|
|
FROM f64
|
|
WHERE metricID=? AND tm > ?
|
|
ORDER BY tm ASC
|
|
LIMIT 1`,
|
|
s.metricID, s.until).Scan(&next.Time, &next.Value, &found)
|
|
|
|
if err != nil {
|
|
if err != sql.ErrNoRows {
|
|
return
|
|
}
|
|
// Нет записей
|
|
err = nil
|
|
}
|
|
|
|
next.Value = applyCorrectionsToMeasure(s.corrections, next.Time, next.Value)
|
|
return
|
|
}
|
|
|
|
// ДЛЯ ТЕСТИРОВАНИЯ
|
|
|
|
type stubPeriod struct {
|
|
Period time.Time
|
|
First PeriodBound
|
|
Last PeriodBound
|
|
}
|
|
|
|
type StubPeriodProducerOptions struct {
|
|
Last *PeriodBound
|
|
Periods []stubPeriod
|
|
}
|
|
|
|
func NewStubPeriodProducer(opt StubPeriodProducerOptions) *StubPeriodProducer {
|
|
s := new(StubPeriodProducer)
|
|
s.last = opt.Last
|
|
s.periods = opt.Periods
|
|
return s
|
|
}
|
|
|
|
type StubPeriodProducer struct {
|
|
last *PeriodBound // для отдельного запроса
|
|
periods []stubPeriod
|
|
idx int
|
|
}
|
|
|
|
func (s *StubPeriodProducer) Next() (period time.Time, first PeriodBound, last PeriodBound, found bool, err error) {
|
|
if s.idx < len(s.periods) {
|
|
p := s.periods[s.idx]
|
|
s.idx++
|
|
period = p.Period
|
|
first = p.First
|
|
last = p.Last
|
|
found = true
|
|
}
|
|
return
|
|
}
|
|
|
|
func (s *StubPeriodProducer) Last() (next PeriodBound, found bool, err error) {
|
|
if s.last != nil {
|
|
next = *s.last
|
|
found = true
|
|
}
|
|
return
|
|
}
|