add dbm new model and fixed json_data

This commit is contained in:
Adrian Zürcher
2025-05-12 17:12:25 +02:00
parent 5ee97416dd
commit 836a69f914
14 changed files with 260 additions and 167 deletions

View File

@@ -21,7 +21,7 @@ func (d *DBMHandler) SaveData(c *gin.Context) {
} }
r := json_dataModels.NewResponse() r := json_dataModels.NewResponse()
r.SetMessage(fmt.Sprintf("DBM %d datapoints saved in: %v", d.GetNumbersOfDatapoints(), time.Since(s))) r.SetMessage(fmt.Sprintf("DBM %d datapoints saved in: %v", d.DBM.GetNumbersOfDatapoints(), time.Since(s)))
d.Log.Info("db.SaveData", fmt.Sprintf("DBM %d datapoints saved in: %v", d.GetNumbersOfDatapoints(), time.Since(s))) d.Log.Info("db.SaveData", fmt.Sprintf("DBM %d datapoints saved in: %v", d.DBM.GetNumbersOfDatapoints(), time.Since(s)))
c.JSON(http.StatusOK, r) c.JSON(http.StatusOK, r)
} }

View File

@@ -12,23 +12,23 @@ import (
"github.com/tecamino/tecamino-dbm/args" "github.com/tecamino/tecamino-dbm/args"
"github.com/tecamino/tecamino-dbm/models" "github.com/tecamino/tecamino-dbm/models"
serverModels "github.com/tecamino/tecamino-dbm/server/models" serverModels "github.com/tecamino/tecamino-dbm/server/models"
json_dataModels "github.com/tecamino/tecamino-json_data/models"
"github.com/tecamino/tecamino-logger/logging" "github.com/tecamino/tecamino-logger/logging"
) )
type DBMHandler struct { type DBMHandler struct {
filePath string DBM *models.DBM
DB models.Datapoint
Conns *serverModels.Connections Conns *serverModels.Connections
sync.RWMutex sync.RWMutex
Log *logging.Logger Log *logging.Logger
arg *args.Args arg *args.Args
filePath string
} }
// initialze new Database Manager // initialze new Database Manager
// it will call cli arguments // it will call cli arguments
func NewDbmHandler(a *args.Args) (*DBMHandler, error) { func NewDbmHandler(a *args.Args) (*DBMHandler, error) {
//initialize new logger
logger, err := logging.NewLogger("dbmServer.log", &logging.Config{ logger, err := logging.NewLogger("dbmServer.log", &logging.Config{
MaxSize: 1, MaxSize: 1,
MaxBackup: 3, MaxBackup: 3,
@@ -42,12 +42,16 @@ func NewDbmHandler(a *args.Args) (*DBMHandler, error) {
} }
logger.Info("main", "start dma handler") logger.Info("main", "start dma handler")
//initialize connection map
conns := serverModels.NewConnections()
// Initialize dtabase manager handler // Initialize dtabase manager handler
dmaHandler := DBMHandler{ dmaHandler := DBMHandler{
arg: a, arg: a,
filePath: fmt.Sprintf("%s/%s.dbm", a.RootDir, a.DBMFile), filePath: fmt.Sprintf("%s/%s.dbm", a.RootDir, a.DBMFile),
DBM: models.NewDBM(conns, logger),
Log: logger, Log: logger,
Conns: serverModels.NewConnections(), Conns: conns,
} }
// initialize system datapoint and periodically update it // initialize system datapoint and periodically update it
@@ -67,16 +71,20 @@ func NewDbmHandler(a *args.Args) (*DBMHandler, error) {
// read in dtaabase file content // read in dtaabase file content
scanner := bufio.NewScanner(f) scanner := bufio.NewScanner(f)
var line int
for scanner.Scan() { for scanner.Scan() {
line++
dp := models.Datapoint{} dp := models.Datapoint{}
if err = json.Unmarshal(scanner.Bytes(), &dp); err != nil { if err = json.Unmarshal(scanner.Bytes(), &dp); err != nil {
dmaHandler.Log.Error("dmbHandler.NewDmbHandler", "error in line "+fmt.Sprint(line)+" "+scanner.Text())
dmaHandler.Log.Error("dmbHandler.NewDmbHandler", err.Error())
return nil, err return nil, err
} }
dmaHandler.ImportDatapoints(dp) dmaHandler.DBM.ImportDatapoints(dp)
} }
} }
dmaHandler.Log.Info("dmbHandler.NewDmbHandler", fmt.Sprintf("%d datapoint imported in %v", dmaHandler.GetNumbersOfDatapoints(), time.Since(s))) dmaHandler.Log.Info("dmbHandler.NewDmbHandler", fmt.Sprintf("%d datapoint imported in %v", dmaHandler.DBM.GetNumbersOfDatapoints(), time.Since(s)))
return &dmaHandler, nil return &dmaHandler, nil
} }
@@ -87,7 +95,7 @@ func (d *DBMHandler) SaveDb() (err error) {
} }
defer f.Close() defer f.Close()
for _, dp := range d.DB.GetAllDatapoints(0) { for _, dp := range d.DBM.GetAllDatapoints(0) {
//exclude System datapoints from saving //exclude System datapoints from saving
//System datapoints are used for internal purposes and should not be saved in the database //System datapoints are used for internal purposes and should not be saved in the database
if strings.Contains(dp.Path, "System:") { if strings.Contains(dp.Path, "System:") {
@@ -110,59 +118,3 @@ func (d *DBMHandler) SaveDb() (err error) {
} }
return return
} }
func (d *DBMHandler) CreateDatapoints(sets ...json_dataModels.Set) ([]json_dataModels.Set, error) {
if len(sets) == 0 {
return nil, nil
}
dps, err := d.DB.CreateDatapoints(d.Conns, sets...)
if err != nil {
return nil, err
}
var ndp uint64
for _, dp := range dps {
if !dp.Updated {
ndp++
}
}
dp := d.QueryDatapoints(1, "System:Datapoints")
d.UpdateDatapointValue("System:Datapoints", dp[0].GetValueUint64()+ndp)
return dps, nil
}
func (d *DBMHandler) ImportDatapoints(dps ...models.Datapoint) error {
for _, dp := range dps {
err := d.DB.ImportDatapoint(d.Conns, dp, dp.Path)
if err != nil {
return err
}
dps := d.QueryDatapoints(1, "System:Datapoints")
d.UpdateDatapointValue("System:Datapoints", dps[0].GetValueUint64()+1)
}
return nil
}
func (d *DBMHandler) UpdateDatapointValue(path string, value any) error {
return d.DB.UpdateDatapointValue(d.Conns, value, path)
}
func (d *DBMHandler) RemoveDatapoint(sets ...json_dataModels.Set) ([]json_dataModels.Set, error) {
var lsRemoved []json_dataModels.Set
for _, set := range sets {
removed, err := d.DB.RemoveDatapoint(d.Conns, set)
if err != nil {
return lsRemoved, err
}
lsRemoved = append(lsRemoved, removed)
dp := d.QueryDatapoints(1, "System:Datapoints")
d.UpdateDatapointValue("System:Datapoints", dp[0].GetValueUint64()-1)
}
return lsRemoved, nil
}
func (d *DBMHandler) QueryDatapoints(depth uint, key string) []*models.Datapoint {
return d.DB.QueryDatapoints(depth, key)
}

View File

@@ -22,7 +22,7 @@ func (d *DBMHandler) Get(req *json_dataModels.Request, id string) {
depth = get.Query.Depth depth = get.Query.Depth
} }
for _, dp := range d.DB.QueryDatapoints(depth, get.Path) { for _, dp := range d.DBM.QueryDatapoints(depth, get.Uuid, get.Path) {
resp.AddGet(json_dataModels.Get{ resp.AddGet(json_dataModels.Get{
Uuid: dp.Uuid, Uuid: dp.Uuid,
Path: dp.Path, Path: dp.Path,

View File

@@ -9,6 +9,7 @@ import (
) )
func (d *DBMHandler) Json_Data(c *gin.Context) { func (d *DBMHandler) Json_Data(c *gin.Context) {
var err error var err error
payload, err := json_data.ParseRequest(c.Request.Body) payload, err := json_data.ParseRequest(c.Request.Body)
if err != nil { if err != nil {
@@ -27,8 +28,7 @@ func (d *DBMHandler) Json_Data(c *gin.Context) {
if get.Query != nil { if get.Query != nil {
depth = get.Query.Depth depth = get.Query.Depth
} }
for _, res := range d.DBM.QueryDatapoints(depth, get.Uuid, get.Path) {
for _, res := range d.QueryDatapoints(depth, get.Path) {
respond.AddGet(json_dataModels.Get{ respond.AddGet(json_dataModels.Get{
Uuid: res.Uuid, Uuid: res.Uuid,
Path: res.Path, Path: res.Path,
@@ -42,7 +42,7 @@ func (d *DBMHandler) Json_Data(c *gin.Context) {
} }
if payload.Set != nil { if payload.Set != nil {
respond.Set, err = d.CreateDatapoints(payload.Set...) respond.Set, err = d.DBM.CreateDatapoints(payload.Set...)
if err != nil { if err != nil {
r := json_data.NewResponse() r := json_data.NewResponse()
r.SetError() r.SetError()
@@ -70,7 +70,7 @@ func (d *DBMHandler) Delete(c *gin.Context) {
if payload.Set != nil { if payload.Set != nil {
response.Set, err = d.RemoveDatapoint(payload.Set...) response.Set, err = d.DBM.RemoveDatapoint(payload.Set...)
if err != nil { if err != nil {
r := json_data.NewResponse() r := json_data.NewResponse()
r.SetError() r.SetError()

View File

@@ -14,7 +14,7 @@ func (d *DBMHandler) Set(req *json_dataModels.Request) {
defer d.RUnlock() defer d.RUnlock()
for _, set := range req.Set { for _, set := range req.Set {
for _, dp := range d.DB.QueryDatapoints(1, set.Path) { for _, dp := range d.DBM.QueryDatapoints(1, set.Uuid, set.Path) {
dp.UpdateValue(d.Conns, set.Value) dp.UpdateValue(d.Conns, set.Value)
} }
} }

View File

@@ -18,7 +18,7 @@ func (d *DBMHandler) Subscribe(req *json_dataModels.Request, id string) {
resp.Id = req.Id resp.Id = req.Id
for _, sub := range req.Subscribe { for _, sub := range req.Subscribe {
for _, dp := range d.DB.QueryDatapoints(sub.Depth, sub.Path) { for _, dp := range d.DBM.QueryDatapoints(sub.Depth, sub.Uuid, sub.Path) {
if sub.Driver != "" { if sub.Driver != "" {
if dp.Drivers == nil || dp.Drivers[sub.Driver] == nil { if dp.Drivers == nil || dp.Drivers[sub.Driver] == nil {
continue continue
@@ -51,9 +51,10 @@ func (d *DBMHandler) Unsubscribe(req *json_dataModels.Request, id string) {
defer d.RUnlock() defer d.RUnlock()
resp := json_dataModels.NewResponse() resp := json_dataModels.NewResponse()
resp.Id = req.Id
for _, sub := range req.Unsubscribe { for _, sub := range req.Unsubscribe {
for _, dp := range d.DB.QueryDatapoints(sub.Depth, sub.Path) { for _, dp := range d.DBM.QueryDatapoints(sub.Depth, sub.Uuid, sub.Path) {
if _, ok := dp.Subscriptions[id]; !ok { if _, ok := dp.Subscriptions[id]; !ok {
continue continue
} }
@@ -64,7 +65,6 @@ func (d *DBMHandler) Unsubscribe(req *json_dataModels.Request, id string) {
}) })
} }
} }
if err := d.Conns.SendResponse(id, resp); err != nil { if err := d.Conns.SendResponse(id, resp); err != nil {
d.Log.Error("subscribe.Unsubscribe", err.Error()) d.Log.Error("subscribe.Unsubscribe", err.Error())
} }

View File

@@ -1,11 +1,8 @@
package dbm package dbm
import ( import (
"fmt" "github.com/google/uuid"
"runtime" "github.com/tecamino/tecamino-dbm/models"
"time"
"github.com/tecamino/tecamino-dbm/utils"
json_dataModels "github.com/tecamino/tecamino-json_data/models" json_dataModels "github.com/tecamino/tecamino-json_data/models"
) )
@@ -14,84 +11,19 @@ func (d *DBMHandler) AddSystemDps() (err error) {
typ := json_dataModels.LOU typ := json_dataModels.LOU
rights := json_dataModels.Read rights := json_dataModels.Read
_, err = d.DB.CreateDatapoints(d.Conns, json_dataModels.Set{Path: path, Value: 0, Type: typ, Rights: rights}) _, err = d.DBM.CreateDatapoints(json_dataModels.Set{Path: path, Value: 0, Type: typ, Rights: rights})
if err != nil { if err != nil {
d.Log.Error("dmb.Handler.AddSystemDps", err.Error()) d.Log.Error("dmb.Handler.AddSystemDps", err.Error())
return err return err
} }
dp := d.QueryDatapoints(1, path) models.SystemDatapoints = d.DBM.QueryDatapoints(1, uuid.Nil, path)[0].Uuid
d.UpdateDatapointValue(path, dp[0].GetValueUint64()+1)
if err = d.GoSystemTime(); err != nil { if err = d.DBM.GoSystemTime(); err != nil {
return err return err
} }
if err = d.GoSystemMemory(); err != nil { if err = d.DBM.GoSystemMemory(); err != nil {
return err return err
} }
return return
} }
func (d *DBMHandler) GetNumbersOfDatapoints() uint64 {
return utils.Uint64From(d.DB.Datapoints["System"].Datapoints["Datapoints"].Value)
}
func (d *DBMHandler) GoSystemTime() error {
path := "System:Time"
var tOld int64
typ := json_dataModels.STR
rights := json_dataModels.Read
_, err := d.DB.CreateDatapoints(d.Conns, json_dataModels.Set{Path: path, Type: typ, Rights: rights})
if err != nil {
d.Log.Error("system.GoSystemTime", err.Error())
return err
}
dp := d.QueryDatapoints(1, "System:Datapoints")
d.UpdateDatapointValue("System:Datapoints", dp[0].GetValueUint64()+1)
go func() {
for {
t := time.Now().UnixMilli()
if tOld != t {
if er := d.DB.UpdateDatapointValue(d.Conns, time.UnixMilli(t).Format("2006-01-02 15:04:05"), path); er != nil {
d.Log.Error("dmb.Handler.AddSystemDps.UpdateDatapointValue", er.Error())
}
tOld = t
}
time.Sleep(time.Second)
}
}()
return nil
}
func (d *DBMHandler) GoSystemMemory() error {
path := "System:UsedMemory"
var m runtime.MemStats
var mOld uint64
typ := json_dataModels.STR
rights := json_dataModels.Read
_, err := d.DB.CreateDatapoints(d.Conns, json_dataModels.Set{Path: path, Type: typ, Rights: rights})
if err != nil {
d.Log.Error("system.GoSystemMemory", err.Error())
return err
}
dp := d.QueryDatapoints(1, "System:Datapoints")
d.UpdateDatapointValue("System:Datapoints", dp[0].GetValueUint64()+1)
go func() {
for {
runtime.ReadMemStats(&m)
if m.Sys != mOld {
mem := fmt.Sprintf("%.2f MB", float64(m.Sys)/1024/1024)
if er := d.DB.UpdateDatapointValue(d.Conns, mem, path); er != nil {
d.Log.Error("dmb.Handler.AddSystemDps.UpdateDatapointValue", er.Error())
}
mOld = m.Sys
}
time.Sleep(time.Second)
}
}()
return nil
}

View File

@@ -72,10 +72,12 @@ func (d *DBMHandler) readJsonData(id string) (request *json_dataModels.Request,
_, reader, err := client.Conn.Reader(client.Ctx) _, reader, err := client.Conn.Reader(client.Ctx)
if err != nil { if err != nil {
return request, err d.Log.Info("webSocket.readJsonData", fmt.Sprintf("WebSocket reader: %v\n", err))
return request, nil
} }
b, err := io.ReadAll(reader) b, err := io.ReadAll(reader)
if err != nil { if err != nil {
code := websocket.CloseStatus(err) code := websocket.CloseStatus(err)

2
go.mod
View File

@@ -6,7 +6,7 @@ require (
github.com/coder/websocket v1.8.13 github.com/coder/websocket v1.8.13
github.com/gin-gonic/gin v1.10.0 github.com/gin-gonic/gin v1.10.0
github.com/google/uuid v1.6.0 github.com/google/uuid v1.6.0
github.com/tecamino/tecamino-json_data v0.0.12 github.com/tecamino/tecamino-json_data v0.0.13
github.com/tecamino/tecamino-logger v0.2.0 github.com/tecamino/tecamino-logger v0.2.0
) )

4
go.sum
View File

@@ -63,8 +63,8 @@ github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o
github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo= github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg= github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
github.com/tecamino/tecamino-json_data v0.0.12 h1:S4Y+WcfQNrin7P73ZI+4eJWh62IwJVhriRsPGGM8N34= github.com/tecamino/tecamino-json_data v0.0.13 h1:hugbmCgXXh0F7YQAEbbJYHkSdq1caejD7SajDiMs42I=
github.com/tecamino/tecamino-json_data v0.0.12/go.mod h1:LLlyD7Wwqplb2BP4PeO86EokEcTRidlW5MwgPd1T2JY= github.com/tecamino/tecamino-json_data v0.0.13/go.mod h1:LLlyD7Wwqplb2BP4PeO86EokEcTRidlW5MwgPd1T2JY=
github.com/tecamino/tecamino-logger v0.2.0 h1:NPH/Gg9qRhmVoW8b39i1eXu/LEftHc74nyISpcRG+XU= github.com/tecamino/tecamino-logger v0.2.0 h1:NPH/Gg9qRhmVoW8b39i1eXu/LEftHc74nyISpcRG+XU=
github.com/tecamino/tecamino-logger v0.2.0/go.mod h1:0M1E9Uei/qw3e3WA1x3lBo1eP3H5oeYE7GjYrMahnj8= github.com/tecamino/tecamino-logger v0.2.0/go.mod h1:0M1E9Uei/qw3e3WA1x3lBo1eP3H5oeYE7GjYrMahnj8=
github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS4MhqMhdFk5YI= github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS4MhqMhdFk5YI=

View File

@@ -87,10 +87,13 @@ func (d *Datapoint) GetValueUint64() uint64 {
return utils.Uint64From(d.Value) return utils.Uint64From(d.Value)
} }
func (d *Datapoint) CreateDatapoints(conns *serverModels.Connections, sets ...json_dataModels.Set) (created []json_dataModels.Set, err error) { func (d *Datapoint) CreateDatapoints(conns *serverModels.Connections, sets ...json_dataModels.Set) (created []json_dataModels.Set, uuids Uuids, err error) {
if len(sets) == 0 { if len(sets) == 0 {
return return
} }
uuids = make(Uuids, 1)
for _, dp := range sets { for _, dp := range sets {
parts := strings.Split(dp.Path, ":") parts := strings.Split(dp.Path, ":")
@@ -105,7 +108,7 @@ func (d *Datapoint) CreateDatapoints(conns *serverModels.Connections, sets ...js
if existing, ok := current.Datapoints[part]; ok { if existing, ok := current.Datapoints[part]; ok {
publish, err := existing.Set("", dp) publish, err := existing.Set("", dp)
if err != nil { if err != nil {
return nil, err return nil, nil, err
} }
created = append(created, json_dataModels.Set{ created = append(created, json_dataModels.Set{
Uuid: existing.Uuid, Uuid: existing.Uuid,
@@ -130,7 +133,7 @@ func (d *Datapoint) CreateDatapoints(conns *serverModels.Connections, sets ...js
current.Datapoints[part] = &ndp current.Datapoints[part] = &ndp
publish, err := ndp.Set(strings.Join(parts, ":"), dp) publish, err := ndp.Set(strings.Join(parts, ":"), dp)
if err != nil { if err != nil {
return nil, err return nil, nil, err
} }
created = append(created, json_dataModels.Set{ created = append(created, json_dataModels.Set{
Uuid: ndp.Uuid, Uuid: ndp.Uuid,
@@ -143,6 +146,8 @@ func (d *Datapoint) CreateDatapoints(conns *serverModels.Connections, sets ...js
if publish { if publish {
current.Publish(conns, OnChange) current.Publish(conns, OnChange)
} }
//add uuid to flat map for faster lookuo
uuids[ndp.Uuid] = &ndp
} }
} }
@@ -173,15 +178,20 @@ func (d *Datapoint) CreateDatapoints(conns *serverModels.Connections, sets ...js
current.Datapoints[part] = newDp current.Datapoints[part] = newDp
current = newDp current = newDp
//add uuid to flat map for faster lookuo
uuids[newDp.Uuid] = newDp
} }
} }
} }
return return
} }
func (d *Datapoint) ImportDatapoint(conns *serverModels.Connections, dp Datapoint, path string) error { func (d *Datapoint) ImportDatapoint(conns *serverModels.Connections, dp Datapoint, path string) (uuids Uuids, err error) {
parts := strings.Split(dp.Path, ":") parts := strings.Split(dp.Path, ":")
uuids = make(Uuids, 1)
current := d current := d
for i, part := range parts { for i, part := range parts {
if current.Datapoints == nil { if current.Datapoints == nil {
@@ -202,9 +212,12 @@ func (d *Datapoint) ImportDatapoint(conns *serverModels.Connections, dp Datapoin
dp.UpdateDateTime = time.Now().UnixMilli() dp.UpdateDateTime = time.Now().UnixMilli()
dp.Subscriptions = InitSubscribtion() dp.Subscriptions = InitSubscribtion()
current.Datapoints[part] = &dp current.Datapoints[part] = &dp
//add uuid to flat map for faster lookuo
uuids[dp.Uuid] = &dp
dp.Publish(conns, OnChange) dp.Publish(conns, OnChange)
} }
return nil
return uuids, nil
} }
// Traverse or create intermediate nodes // Traverse or create intermediate nodes
@@ -220,9 +233,11 @@ func (d *Datapoint) ImportDatapoint(conns *serverModels.Connections, dp Datapoin
newDp.ReadWrite = newDp.ReadWrite.GetRights() newDp.ReadWrite = newDp.ReadWrite.GetRights()
current.Datapoints[part] = newDp current.Datapoints[part] = newDp
current = newDp current = newDp
//add uuid to flat map for faster lookuo
uuids[newDp.Uuid] = newDp
} }
} }
return nil return uuids, nil
} }
func (d *Datapoint) UpdateDatapointValue(conns *serverModels.Connections, value any, path string) error { func (d *Datapoint) UpdateDatapointValue(conns *serverModels.Connections, value any, path string) error {

190
models/dbm.go Normal file
View File

@@ -0,0 +1,190 @@
package models
import (
"fmt"
"runtime"
"time"
"maps"
"github.com/google/uuid"
serverModels "github.com/tecamino/tecamino-dbm/server/models"
"github.com/tecamino/tecamino-dbm/utils"
json_dataModels "github.com/tecamino/tecamino-json_data/models"
"github.com/tecamino/tecamino-logger/logging"
)
type DBM struct {
Datapoints Datapoint
Uuids Uuids
Conns *serverModels.Connections
Log *logging.Logger
}
var SystemDatapoints uuid.UUID
func NewDBM(conns *serverModels.Connections, log *logging.Logger) *DBM {
return &DBM{
Datapoints: Datapoint{},
Uuids: make(Uuids),
Conns: conns,
Log: log,
}
}
func (d *DBM) CreateDatapoints(sets ...json_dataModels.Set) ([]json_dataModels.Set, error) {
if len(sets) == 0 {
return nil, nil
}
dps, uuids, err := d.Datapoints.CreateDatapoints(d.Conns, sets...)
//save uuid in seperate map for fast look up
maps.Copy(d.Uuids, uuids)
if err != nil {
return nil, err
}
var ndp uint64
for _, dp := range dps {
if !dp.Updated {
ndp++
}
}
d.ModifyCountedDatapoints(ndp, false)
return dps, nil
}
func (d *DBM) ImportDatapoints(dps ...Datapoint) error {
for _, dp := range dps {
uuids, err := d.Datapoints.ImportDatapoint(d.Conns, dp, dp.Path)
if err != nil {
return err
}
maps.Copy(d.Uuids, uuids)
d.ModifyCountedDatapoints(1, false)
}
return nil
}
func (d *DBM) UpdateDatapointValue(value any, uid uuid.UUID, path ...string) error {
if uid != uuid.Nil {
if _, ok := d.Uuids[uid]; !ok {
return fmt.Errorf("uuid %s not found", uid.String())
}
dp := d.Uuids[uid]
dp.Value = dp.Type.ConvertValue(value)
dp.UpdateDateTime = time.Now().UnixMilli()
dp.Publish(d.Conns, OnChange)
}
if len(path) > 1 {
return fmt.Errorf("only one path allowed")
}
return d.Datapoints.UpdateDatapointValue(d.Conns, value, path[0])
}
func (d *DBM) RemoveDatapoint(sets ...json_dataModels.Set) ([]json_dataModels.Set, error) {
var lsRemoved []json_dataModels.Set
for _, set := range sets {
removed, err := d.Datapoints.RemoveDatapoint(d.Conns, set)
if err != nil {
return lsRemoved, err
}
lsRemoved = append(lsRemoved, removed)
d.ModifyCountedDatapoints(1, true)
}
return lsRemoved, nil
}
func (d *DBM) QueryDatapoints(depth uint, uid uuid.UUID, key ...string) []*Datapoint {
if uid != uuid.Nil {
if _, ok := d.Uuids[uid]; !ok {
return nil
}
dp := d.Uuids[uid]
dps := []*Datapoint{dp}
return append(dps, dp.QueryDatapoints(depth, key[0])...)
}
return d.Datapoints.QueryDatapoints(depth, key[0])
}
func (d *DBM) GetAllDatapoints(depth uint) (dps Datapoints) {
return d.Datapoints.GetAllDatapoints(0)
}
func (d *DBM) GetNumbersOfDatapoints() uint64 {
return utils.Uint64From(d.Datapoints.Datapoints["System"].Datapoints["Datapoints"].Value)
}
func (d *DBM) ModifyCountedDatapoints(count uint64, countDown bool) {
dp := d.QueryDatapoints(1, SystemDatapoints, "System:Datapoints")
amount := dp[0].GetValueUint64()
if countDown {
amount -= count
} else {
amount += count
}
d.UpdateDatapointValue(amount, SystemDatapoints, "System:Datapoints")
}
func (d *DBM) GoSystemTime() error {
path := "System:Time"
var tOld int64
typ := json_dataModels.STR
rights := json_dataModels.Read
_, err := d.CreateDatapoints(json_dataModels.Set{Path: path, Type: typ, Rights: rights})
if err != nil {
d.Log.Error("system.GoSystemTime", err.Error())
return err
}
go func() {
for {
t := time.Now().UnixMilli()
if tOld != t {
if er := d.UpdateDatapointValue(time.UnixMilli(t).Format("2006-01-02 15:04:05"), uuid.Nil, path); er != nil {
d.Log.Error("dmb.Handler.AddSystemDps.UpdateDatapointValue", er.Error())
}
tOld = t
}
time.Sleep(time.Second)
}
}()
return nil
}
func (d *DBM) GoSystemMemory() error {
path := "System:UsedMemory"
var m runtime.MemStats
var mOld uint64
typ := json_dataModels.STR
rights := json_dataModels.Read
_, err := d.CreateDatapoints(json_dataModels.Set{Path: path, Type: typ, Rights: rights})
if err != nil {
d.Log.Error("system.GoSystemMemory", err.Error())
return err
}
go func() {
for {
runtime.ReadMemStats(&m)
if m.Sys != mOld {
mem := fmt.Sprintf("%.2f MB", float64(m.Sys)/1024/1024)
if er := d.UpdateDatapointValue(mem, uuid.Nil, path); er != nil {
d.Log.Error("dmb.Handler.AddSystemDps.UpdateDatapointValue", er.Error())
}
mOld = m.Sys
}
time.Sleep(time.Second)
}
}()
return nil
}

5
models/uuids.go Normal file
View File

@@ -0,0 +1,5 @@
package models
import "github.com/google/uuid"
type Uuids map[uuid.UUID]*Datapoint

View File

@@ -7,6 +7,7 @@ import (
"testing" "testing"
"time" "time"
"github.com/google/uuid"
"github.com/tecamino/tecamino-dbm/args" "github.com/tecamino/tecamino-dbm/args"
"github.com/tecamino/tecamino-dbm/cert" "github.com/tecamino/tecamino-dbm/cert"
"github.com/tecamino/tecamino-dbm/dbm" "github.com/tecamino/tecamino-dbm/dbm"
@@ -87,11 +88,7 @@ func TestQuery(t *testing.T) {
panic(err) panic(err)
} }
// for i, o := range dmaHandler.QueryDatapoints(".*002.*") { for i, o := range dmaHandler.DBM.QueryDatapoints(1, uuid.Nil, "Test:A:000") {
// fmt.Println(600, i, o)
// }
for i, o := range dmaHandler.QueryDatapoints(1, "Test:A:000") {
fmt.Println(600, i, o) fmt.Println(600, i, o)
} }