94 lines
1.9 KiB
Go
94 lines
1.9 KiB
Go
|
package daemon
|
||
|
|
|
||
|
|
import (
|
||
|
|
"fmt"
|
||
|
|
"net/url"
|
||
|
|
"sync"
|
||
|
|
|
||
|
|
"github.com/go-flucky/flucky/pkg/storage"
|
||
|
|
"github.com/go-flucky/flucky/pkg/storage/logfile"
|
||
|
|
"github.com/go-flucky/flucky/pkg/types"
|
||
|
|
)
|
||
|
|
|
||
|
|
type cacheStore struct {
|
||
|
|
compression bool
|
||
|
|
cache []*types.MeasuredValue
|
||
|
|
mux *sync.Mutex
|
||
|
|
round float64
|
||
|
|
URL *url.URL
|
||
|
|
}
|
||
|
|
|
||
|
|
func (cs *cacheStore) Add(measuredValue *types.MeasuredValue) {
|
||
|
|
cs.mux.Lock()
|
||
|
|
defer cs.mux.Unlock()
|
||
|
|
cs.cache = append(cs.cache, measuredValue)
|
||
|
|
}
|
||
|
|
|
||
|
|
func (cs *cacheStore) Flush(measuredValue *types.MeasuredValue) *cacheStore {
|
||
|
|
cs.mux.Lock()
|
||
|
|
defer cs.mux.Unlock()
|
||
|
|
cs.cache = make([]*types.MeasuredValue, 0)
|
||
|
|
return cs
|
||
|
|
}
|
||
|
|
|
||
|
|
func (cs *cacheStore) Get(id string) *types.MeasuredValue {
|
||
|
|
cs.mux.Lock()
|
||
|
|
defer cs.mux.Unlock()
|
||
|
|
for _, measuredValue := range cs.cache {
|
||
|
|
if measuredValue.ID == id {
|
||
|
|
return measuredValue
|
||
|
|
}
|
||
|
|
}
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
|
||
|
|
func (cs *cacheStore) Size() int {
|
||
|
|
cs.mux.Lock()
|
||
|
|
defer cs.mux.Unlock()
|
||
|
|
return len(cs.cache)
|
||
|
|
}
|
||
|
|
|
||
|
|
func (cs *cacheStore) WriteToEndpoint() error {
|
||
|
|
cs.mux.Lock()
|
||
|
|
defer cs.mux.Unlock()
|
||
|
|
defer func() { cs.cache = make([]*types.MeasuredValue, 0) }()
|
||
|
|
switch cs.URL.Scheme {
|
||
|
|
case "file":
|
||
|
|
return cs.logfile()
|
||
|
|
case "postgres":
|
||
|
|
return fmt.Errorf("Not yet implemented to store values into a postgres database")
|
||
|
|
}
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
|
||
|
|
func (cs *cacheStore) logfile() error {
|
||
|
|
|
||
|
|
newMeasuredValues := make([]*types.MeasuredValue, 0)
|
||
|
|
for _, measuredValue := range cs.cache {
|
||
|
|
newMeasuredValues = append(newMeasuredValues, measuredValue)
|
||
|
|
}
|
||
|
|
|
||
|
|
if cs.round != 0 {
|
||
|
|
storage.Round(newMeasuredValues, cs.round)
|
||
|
|
}
|
||
|
|
|
||
|
|
measuredLogfile := logfile.New(cs.URL.Path)
|
||
|
|
|
||
|
|
measuredValues, err := measuredLogfile.Read()
|
||
|
|
if err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
measuredValues = append(measuredValues, newMeasuredValues...)
|
||
|
|
|
||
|
|
if cs.compression {
|
||
|
|
measuredValues = storage.Compression(measuredValues)
|
||
|
|
}
|
||
|
|
|
||
|
|
err = measuredLogfile.Write(measuredValues)
|
||
|
|
if err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
|
||
|
|
return nil
|
||
|
|
}
|