Compare commits
4 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
836a69f914 | ||
|
|
5ee97416dd | ||
|
|
ecb1f3b2cf | ||
|
|
5203fb8543 |
@@ -7,6 +7,7 @@ import (
|
|||||||
"github.com/tecamino/tecamino-dbm/models"
|
"github.com/tecamino/tecamino-dbm/models"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// DBM cli arguments
|
||||||
type Args struct {
|
type Args struct {
|
||||||
Port models.Port
|
Port models.Port
|
||||||
Cert cert.Cert
|
Cert cert.Cert
|
||||||
@@ -15,6 +16,7 @@ type Args struct {
|
|||||||
Debug bool
|
Debug bool
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// initialte cli arguments
|
||||||
func Init() *Args {
|
func Init() *Args {
|
||||||
|
|
||||||
a := Args{
|
a := Args{
|
||||||
|
|||||||
@@ -13,14 +13,15 @@ func (d *DBMHandler) SaveData(c *gin.Context) {
|
|||||||
s := time.Now()
|
s := time.Now()
|
||||||
if err := d.SaveDb(); err != nil {
|
if err := d.SaveDb(); err != nil {
|
||||||
r := json_dataModels.NewResponse()
|
r := json_dataModels.NewResponse()
|
||||||
r.SendError(err.Error())
|
r.SetError()
|
||||||
|
r.SetMessage(err.Error())
|
||||||
c.JSON(http.StatusBadRequest, r)
|
c.JSON(http.StatusBadRequest, r)
|
||||||
|
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
r := json_dataModels.NewResponse()
|
r := json_dataModels.NewResponse()
|
||||||
r.SendMessage(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)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -5,29 +5,30 @@ import (
|
|||||||
"encoding/json"
|
"encoding/json"
|
||||||
"fmt"
|
"fmt"
|
||||||
"os"
|
"os"
|
||||||
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"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,
|
||||||
@@ -41,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
|
||||||
@@ -66,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
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -86,7 +95,13 @@ 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
|
||||||
|
//System datapoints are used for internal purposes and should not be saved in the database
|
||||||
|
if strings.Contains(dp.Path, "System:") {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
b, er := json.Marshal(dp)
|
b, er := json.Marshal(dp)
|
||||||
if er != nil {
|
if er != nil {
|
||||||
return er
|
return er
|
||||||
@@ -103,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)
|
|
||||||
}
|
|
||||||
|
|||||||
18
dbm/get.go
18
dbm/get.go
@@ -5,23 +5,25 @@ import (
|
|||||||
json_dataModels "github.com/tecamino/tecamino-json_data/models"
|
json_dataModels "github.com/tecamino/tecamino-json_data/models"
|
||||||
)
|
)
|
||||||
|
|
||||||
func (d *DBMHandler) Get(gets []json_dataModels.Get, id, id2 string) {
|
func (d *DBMHandler) Get(req *json_dataModels.Request, id string) {
|
||||||
if gets == nil {
|
if req == nil {
|
||||||
|
return
|
||||||
|
} else if len(req.Get) == 0 {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
d.RLock()
|
d.RLock()
|
||||||
defer d.RUnlock()
|
defer d.RUnlock()
|
||||||
|
|
||||||
r := json_data.NewResponse()
|
resp := json_data.NewResponse()
|
||||||
r.Id = id2
|
resp.Id = req.Id
|
||||||
for _, get := range gets {
|
for _, get := range req.Get {
|
||||||
var depth uint = 1
|
var depth uint = 1
|
||||||
if get.Query != nil {
|
if get.Query != nil {
|
||||||
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) {
|
||||||
r.AddGet(json_dataModels.Get{
|
resp.AddGet(json_dataModels.Get{
|
||||||
Uuid: dp.Uuid,
|
Uuid: dp.Uuid,
|
||||||
Path: dp.Path,
|
Path: dp.Path,
|
||||||
Type: dp.Type,
|
Type: dp.Type,
|
||||||
@@ -31,7 +33,7 @@ func (d *DBMHandler) Get(gets []json_dataModels.Get, id, id2 string) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if err := d.Conns.SendResponse(id, r); err != nil {
|
if err := d.Conns.SendResponse(id, resp); err != nil {
|
||||||
d.Log.Error("get.Get", err.Error())
|
d.Log.Error("get.Get", err.Error())
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -9,11 +9,13 @@ 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 {
|
||||||
r := json_data.NewResponse()
|
r := json_data.NewResponse()
|
||||||
r.SendError(err.Error())
|
r.SetError()
|
||||||
|
r.SetMessage(err.Error())
|
||||||
c.JSON(http.StatusBadRequest, r)
|
c.JSON(http.StatusBadRequest, r)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
@@ -26,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,
|
||||||
@@ -41,17 +42,17 @@ 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.SendError(err.Error())
|
r.SetError()
|
||||||
|
r.SetMessage(err.Error())
|
||||||
c.JSON(http.StatusBadRequest, r)
|
c.JSON(http.StatusBadRequest, r)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
c.JSON(200, respond)
|
c.JSON(200, respond)
|
||||||
return
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (d *DBMHandler) Delete(c *gin.Context) {
|
func (d *DBMHandler) Delete(c *gin.Context) {
|
||||||
@@ -59,7 +60,8 @@ func (d *DBMHandler) Delete(c *gin.Context) {
|
|||||||
payload, err := json_data.ParseRequest(c.Request.Body)
|
payload, err := json_data.ParseRequest(c.Request.Body)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
r := json_data.NewResponse()
|
r := json_data.NewResponse()
|
||||||
r.SendError(err.Error())
|
r.SetError()
|
||||||
|
r.SetMessage(err.Error())
|
||||||
c.JSON(http.StatusBadRequest, r)
|
c.JSON(http.StatusBadRequest, r)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
@@ -68,15 +70,15 @@ 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.SendError(err.Error())
|
r.SetError()
|
||||||
|
r.SetMessage(err.Error())
|
||||||
c.JSON(http.StatusBadRequest, r)
|
c.JSON(http.StatusBadRequest, r)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
c.JSON(200, response)
|
c.JSON(200, response)
|
||||||
return
|
|
||||||
}
|
}
|
||||||
|
|||||||
10
dbm/set.go
10
dbm/set.go
@@ -4,15 +4,17 @@ import (
|
|||||||
json_dataModels "github.com/tecamino/tecamino-json_data/models"
|
json_dataModels "github.com/tecamino/tecamino-json_data/models"
|
||||||
)
|
)
|
||||||
|
|
||||||
func (d *DBMHandler) Set(sets []json_dataModels.Set) {
|
func (d *DBMHandler) Set(req *json_dataModels.Request) {
|
||||||
if sets == nil {
|
if req == nil {
|
||||||
|
return
|
||||||
|
} else if len(req.Set) == 0 {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
d.RLock()
|
d.RLock()
|
||||||
defer d.RUnlock()
|
defer d.RUnlock()
|
||||||
|
|
||||||
for _, set := range sets {
|
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)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -4,25 +4,28 @@ import (
|
|||||||
json_dataModels "github.com/tecamino/tecamino-json_data/models"
|
json_dataModels "github.com/tecamino/tecamino-json_data/models"
|
||||||
)
|
)
|
||||||
|
|
||||||
func (d *DBMHandler) Subscribe(subs []json_dataModels.Subscribe, id, id2 string) {
|
func (d *DBMHandler) Subscribe(req *json_dataModels.Request, id string) {
|
||||||
if subs == nil {
|
if req == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if len(req.Subscribe) == 0 {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
d.RLock()
|
d.RLock()
|
||||||
defer d.RUnlock()
|
defer d.RUnlock()
|
||||||
|
|
||||||
r := json_dataModels.NewResponse()
|
resp := json_dataModels.NewResponse()
|
||||||
r.Id = id2
|
resp.Id = req.Id
|
||||||
|
|
||||||
for _, sub := range subs {
|
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
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
dp.AddSubscribtion(id, sub)
|
dp.AddSubscribtion(id, sub)
|
||||||
r.AddSubscription(json_dataModels.Subscribe{
|
resp.AddSubscription(json_dataModels.Subscription{
|
||||||
Uuid: dp.Uuid,
|
Uuid: dp.Uuid,
|
||||||
Path: dp.Path,
|
Path: dp.Path,
|
||||||
Value: dp.Value,
|
Value: dp.Value,
|
||||||
@@ -32,34 +35,37 @@ func (d *DBMHandler) Subscribe(subs []json_dataModels.Subscribe, id, id2 string)
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if err := d.Conns.SendResponse(id, r); err != nil {
|
if err := d.Conns.SendResponse(id, resp); err != nil {
|
||||||
d.Log.Error("subscribe.Subscribe", err.Error())
|
d.Log.Error("subscribe.Subscribe", err.Error())
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func (d *DBMHandler) Unsubscribe(subs []json_dataModels.Subscribe, id string) {
|
func (d *DBMHandler) Unsubscribe(req *json_dataModels.Request, id string) {
|
||||||
if subs == nil {
|
if req == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if len(req.Unsubscribe) == 0 {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
d.RLock()
|
d.RLock()
|
||||||
defer d.RUnlock()
|
defer d.RUnlock()
|
||||||
|
|
||||||
r := json_dataModels.NewResponse()
|
resp := json_dataModels.NewResponse()
|
||||||
|
resp.Id = req.Id
|
||||||
|
|
||||||
for _, sub := range subs {
|
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
|
||||||
}
|
}
|
||||||
dp.RemoveSubscribtion(id)
|
dp.RemoveSubscribtion(id)
|
||||||
r.AddUnsubscription(json_dataModels.Subscribe{
|
resp.AddUnsubscription(json_dataModels.Subscription{
|
||||||
Uuid: dp.Uuid,
|
Uuid: dp.Uuid,
|
||||||
Path: dp.Path,
|
Path: dp.Path,
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
if err := d.Conns.SendResponse(id, resp); err != nil {
|
||||||
if err := d.Conns.SendResponse(id, r); err != nil {
|
|
||||||
d.Log.Error("subscribe.Unsubscribe", err.Error())
|
d.Log.Error("subscribe.Unsubscribe", err.Error())
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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
|
|
||||||
}
|
|
||||||
|
|||||||
@@ -45,20 +45,20 @@ func (d *DBMHandler) WebSocket(c *gin.Context) {
|
|||||||
|
|
||||||
// Sets
|
// Sets
|
||||||
|
|
||||||
d.Get(request.Get, id, request.Id)
|
d.Get(request, id)
|
||||||
// Sets
|
// Sets
|
||||||
d.Set(request.Set)
|
d.Set(request)
|
||||||
|
|
||||||
// Subscribe
|
// Subscribe
|
||||||
d.Subscribe(request.Subscribe, id, request.Id)
|
d.Subscribe(request, id)
|
||||||
|
|
||||||
// Unsubscribe
|
// Unsubscribe
|
||||||
d.Unsubscribe(request.Unsubscribe, id)
|
d.Unsubscribe(request, id)
|
||||||
|
|
||||||
request.Get = make([]json_dataModels.Get, 0)
|
request.Get = make([]json_dataModels.Get, 0)
|
||||||
request.Set = make([]json_dataModels.Set, 0)
|
request.Set = make([]json_dataModels.Set, 0)
|
||||||
request.Subscribe = make([]json_dataModels.Subscribe, 0)
|
request.Subscribe = make([]json_dataModels.Subscription, 0)
|
||||||
request.Unsubscribe = make([]json_dataModels.Subscribe, 0)
|
request.Unsubscribe = make([]json_dataModels.Subscription, 0)
|
||||||
request = nil
|
request = nil
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -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
2
go.mod
@@ -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.10
|
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
4
go.sum
@@ -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.10 h1:I5xvJ8eRxX0QbMTuSHlAA16FQ8uE49OiCSsQ7Xjircc=
|
github.com/tecamino/tecamino-json_data v0.0.13 h1:hugbmCgXXh0F7YQAEbbJYHkSdq1caejD7SajDiMs42I=
|
||||||
github.com/tecamino/tecamino-json_data v0.0.10/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=
|
||||||
|
|||||||
6
main.go
6
main.go
@@ -13,10 +13,12 @@ func main() {
|
|||||||
// cli arguments
|
// cli arguments
|
||||||
a := args.Init()
|
a := args.Init()
|
||||||
|
|
||||||
|
// initiate new database manger
|
||||||
dbmHandler, err := dbm.NewDbmHandler(a)
|
dbmHandler, err := dbm.NewDbmHandler(a)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
panic(err)
|
panic(err)
|
||||||
}
|
}
|
||||||
|
//save database after exeutabe ends
|
||||||
defer dbmHandler.SaveDb()
|
defer dbmHandler.SaveDb()
|
||||||
|
|
||||||
//initialize new server
|
//initialize new server
|
||||||
@@ -29,20 +31,20 @@ func main() {
|
|||||||
s.Routes.GET("/saveData", dbmHandler.SaveData)
|
s.Routes.GET("/saveData", dbmHandler.SaveData)
|
||||||
s.Routes.POST("/json_data", dbmHandler.Json_Data)
|
s.Routes.POST("/json_data", dbmHandler.Json_Data)
|
||||||
s.Routes.DELETE("/json_data", dbmHandler.Delete)
|
s.Routes.DELETE("/json_data", dbmHandler.Delete)
|
||||||
|
|
||||||
s.Routes.GET("/", func(c *gin.Context) {
|
s.Routes.GET("/", func(c *gin.Context) {
|
||||||
c.String(200, "DBM WebSocket Server is running!")
|
c.String(200, "DBM WebSocket Server is running!")
|
||||||
})
|
})
|
||||||
|
|
||||||
|
// start http server
|
||||||
go func() {
|
go func() {
|
||||||
dbmHandler.Log.Info("main", fmt.Sprintf("http listen on %d", a.Port.Http))
|
dbmHandler.Log.Info("main", fmt.Sprintf("http listen on %d", a.Port.Http))
|
||||||
// start http server
|
|
||||||
if err := s.ServeHttp(a.Port.Http); err != nil {
|
if err := s.ServeHttp(a.Port.Http); err != nil {
|
||||||
dbmHandler.Log.Error("main", "error http server "+err.Error())
|
dbmHandler.Log.Error("main", "error http server "+err.Error())
|
||||||
panic(err)
|
panic(err)
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
|
|
||||||
|
// start https server
|
||||||
dbmHandler.Log.Info("main", fmt.Sprintf("https listen on %d", a.Port.Https))
|
dbmHandler.Log.Info("main", fmt.Sprintf("https listen on %d", a.Port.Https))
|
||||||
if err := s.ServeHttps(a.Port.Https, a.Cert); err != nil {
|
if err := s.ServeHttps(a.Port.Https, a.Cert); err != nil {
|
||||||
dbmHandler.Log.Error("main", "error http server "+err.Error())
|
dbmHandler.Log.Error("main", "error http server "+err.Error())
|
||||||
|
|||||||
@@ -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 {
|
||||||
@@ -340,7 +355,7 @@ func (d *Datapoint) QueryDatapoints(depth uint, path string) (dps Datapoints) {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
func (d *Datapoint) AddSubscribtion(id string, sub json_dataModels.Subscribe) {
|
func (d *Datapoint) AddSubscribtion(id string, sub json_dataModels.Subscription) {
|
||||||
if d.Subscriptions == nil {
|
if d.Subscriptions == nil {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|||||||
190
models/dbm.go
Normal file
190
models/dbm.go
Normal 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
5
models/uuids.go
Normal file
@@ -0,0 +1,5 @@
|
|||||||
|
package models
|
||||||
|
|
||||||
|
import "github.com/google/uuid"
|
||||||
|
|
||||||
|
type Uuids map[uuid.UUID]*Datapoint
|
||||||
@@ -12,11 +12,13 @@ import (
|
|||||||
json_dataModels "github.com/tecamino/tecamino-json_data/models"
|
json_dataModels "github.com/tecamino/tecamino-json_data/models"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// serves as connection handler of websocket
|
||||||
type Connections struct {
|
type Connections struct {
|
||||||
sync.RWMutex
|
sync.RWMutex
|
||||||
Clients Clients
|
Clients Clients
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// initaiates new conections with client map
|
||||||
func NewConnections() *Connections {
|
func NewConnections() *Connections {
|
||||||
return &Connections{
|
return &Connections{
|
||||||
Clients: NewClients(),
|
Clients: NewClients(),
|
||||||
@@ -36,10 +38,11 @@ func (c *Connections) DisconnectWsConnection(id string, code websocket.StatusCod
|
|||||||
c.Clients.DisconnectWsConnection(id, code, reason)
|
c.Clients.DisconnectWsConnection(id, code, reason)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// sends json response to client
|
||||||
func (c *Connections) SendResponse(id string, r *json_dataModels.Response) error {
|
func (c *Connections) SendResponse(id string, r *json_dataModels.Response) error {
|
||||||
client, ok := c.Clients[id]
|
client, ok := c.Clients[id]
|
||||||
if !ok {
|
if !ok {
|
||||||
return fmt.Errorf("client not found for id " + id)
|
return fmt.Errorf("client not found for id %s", id)
|
||||||
|
|
||||||
}
|
}
|
||||||
b, err := json.Marshal(r)
|
b, err := json.Marshal(r)
|
||||||
|
|||||||
@@ -9,23 +9,28 @@ import (
|
|||||||
"github.com/tecamino/tecamino-logger/logging"
|
"github.com/tecamino/tecamino-logger/logging"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// server model for database manager websocket
|
||||||
type Server struct {
|
type Server struct {
|
||||||
Routes *gin.Engine
|
Routes *gin.Engine
|
||||||
sync.RWMutex
|
sync.RWMutex
|
||||||
Logger *logging.Logger
|
Logger *logging.Logger
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// initalizes new dbm server
|
||||||
func NewServer() *Server {
|
func NewServer() *Server {
|
||||||
return &Server{
|
return &Server{
|
||||||
Routes: gin.Default(),
|
Routes: gin.Default(),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// serve dbm as http
|
||||||
func (s *Server) ServeHttp(port uint) error {
|
func (s *Server) ServeHttp(port uint) error {
|
||||||
return s.Routes.Run(fmt.Sprintf(":%d", port))
|
return s.Routes.Run(fmt.Sprintf(":%d", port))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// serve dbm as http
|
||||||
func (s *Server) ServeHttps(port uint, cert cert.Cert) error {
|
func (s *Server) ServeHttps(port uint, cert cert.Cert) error {
|
||||||
|
// generate self signed tls certificate
|
||||||
if err := cert.GenerateSelfSignedCert(); err != nil {
|
if err := cert.GenerateSelfSignedCert(); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -5,6 +5,7 @@ import (
|
|||||||
"strings"
|
"strings"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// return any input type to float32
|
||||||
func Float32From(v any) float32 {
|
func Float32From(v any) float32 {
|
||||||
switch val := v.(type) {
|
switch val := v.(type) {
|
||||||
case bool:
|
case bool:
|
||||||
@@ -44,6 +45,7 @@ func Float32From(v any) float32 {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// return any input type to float64
|
||||||
func Float64From(v any) float64 {
|
func Float64From(v any) float64 {
|
||||||
switch val := v.(type) {
|
switch val := v.(type) {
|
||||||
case bool:
|
case bool:
|
||||||
@@ -83,6 +85,7 @@ func Float64From(v any) float64 {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// return any input type to int8
|
||||||
func Int8From(v any) int8 {
|
func Int8From(v any) int8 {
|
||||||
switch val := v.(type) {
|
switch val := v.(type) {
|
||||||
case bool:
|
case bool:
|
||||||
@@ -122,6 +125,7 @@ func Int8From(v any) int8 {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// return any input type to int16
|
||||||
func Int16From(v any) int16 {
|
func Int16From(v any) int16 {
|
||||||
switch val := v.(type) {
|
switch val := v.(type) {
|
||||||
case bool:
|
case bool:
|
||||||
@@ -161,6 +165,7 @@ func Int16From(v any) int16 {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// return any input type to int32
|
||||||
func Int32From(v any) int32 {
|
func Int32From(v any) int32 {
|
||||||
switch val := v.(type) {
|
switch val := v.(type) {
|
||||||
case bool:
|
case bool:
|
||||||
@@ -200,6 +205,7 @@ func Int32From(v any) int32 {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// return any input type to int64
|
||||||
func Int64From(v any) int64 {
|
func Int64From(v any) int64 {
|
||||||
switch val := v.(type) {
|
switch val := v.(type) {
|
||||||
case bool:
|
case bool:
|
||||||
@@ -239,6 +245,7 @@ func Int64From(v any) int64 {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// return any input type to int
|
||||||
func Uint8From(v any) uint8 {
|
func Uint8From(v any) uint8 {
|
||||||
switch val := v.(type) {
|
switch val := v.(type) {
|
||||||
case bool:
|
case bool:
|
||||||
@@ -278,6 +285,7 @@ func Uint8From(v any) uint8 {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// return any input type to uint16
|
||||||
func Uint16From(v any) uint16 {
|
func Uint16From(v any) uint16 {
|
||||||
switch val := v.(type) {
|
switch val := v.(type) {
|
||||||
case bool:
|
case bool:
|
||||||
@@ -317,6 +325,7 @@ func Uint16From(v any) uint16 {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// return any input type to uint32
|
||||||
func Uint32From(v any) uint32 {
|
func Uint32From(v any) uint32 {
|
||||||
switch val := v.(type) {
|
switch val := v.(type) {
|
||||||
case bool:
|
case bool:
|
||||||
@@ -356,6 +365,7 @@ func Uint32From(v any) uint32 {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// return any input type to uint64
|
||||||
func Uint64From(v any) uint64 {
|
func Uint64From(v any) uint64 {
|
||||||
switch val := v.(type) {
|
switch val := v.(type) {
|
||||||
case bool:
|
case bool:
|
||||||
@@ -395,6 +405,7 @@ func Uint64From(v any) uint64 {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// return any input type to bool
|
||||||
func BoolFrom(v any) bool {
|
func BoolFrom(v any) bool {
|
||||||
switch val := v.(type) {
|
switch val := v.(type) {
|
||||||
case bool:
|
case bool:
|
||||||
|
|||||||
Reference in New Issue
Block a user