paadi/tecamino-dbm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

implement new json_data model

Browse Source
This commit is contained in:
Adrian Zürcher
2025-04-29 08:31:06 +02:00
parent 0a137c9d86
commit 49d8d03d8a
25 changed files with 609 additions and 572 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
dbm/db.go Normal file
View File

@@ -0,0 +1,25 @@
package dbm
import (
"fmt"
"net/http"
"time"
"github.com/gin-gonic/gin"
json_dataModels "github.com/tecamino/tecamino-json_data/models"
)
func (d *DBMHandler) SaveData(c *gin.Context) {
s := time.Now()
if err := d.SaveDb(); err != nil {
r := json_dataModels.NewResponse()
r.SendError(err.Error())
c.JSON(http.StatusBadRequest, r)
return
}
r := json_dataModels.NewResponse()
r.SendMessage(fmt.Sprintf("DBM %ddatapints saved in: %v", d.GetNumbersOfDatapoints(), time.Since(s)))
c.JSON(http.StatusOK, r)
}

124
dbm/dbmHandler.go
View File

@@ -2,24 +2,23 @@ package dbm
import (
"bufio"
"context"
"encoding/json"
"fmt"
"os"
"runtime"
"sync"
"time"
"github.com/tecamino/tecamino-dbm/args"
"github.com/tecamino/tecamino-dbm/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/zuadi/tecamino-dbm/args"
"github.com/zuadi/tecamino-dbm/models"
serverModels "github.com/zuadi/tecamino-dbm/server/models"
)
type DBMHandler struct {
filePath string
DB models.Datapoint
Clients serverModels.Clients
Conns *serverModels.Connections
sync.RWMutex
Log *logging.Logger
arg *args.Args
@@ -40,14 +39,14 @@ func NewDbmHandler(a *args.Args) (*DBMHandler, error) {
if err != nil {
return nil, err
}
logger.Info("main", "start dma")
logger.Info("main", "start dma handler")
// Initialize dtabase manager handler
dmaHandler := DBMHandler{
arg: a,
filePath: fmt.Sprintf("%s/%s.dma", a.RootDir, a.DMAFile),
Log: logger,
Clients: serverModels.NewClients(),
Conns: serverModels.NewConnections(),
}
// initialize system datapoint and periodically update it
@@ -56,6 +55,7 @@ func NewDbmHandler(a *args.Args) (*DBMHandler, error) {
}
// check if dtabase file exists to load data
s := time.Now()
if _, err := os.Stat(dmaHandler.filePath); err == nil {
f, err := os.Open(dmaHandler.filePath)
@@ -69,14 +69,13 @@ func NewDbmHandler(a *args.Args) (*DBMHandler, error) {
for scanner.Scan() {
dp := models.Datapoint{}
err = json.Unmarshal(scanner.Bytes(), &dp)
if err != nil {
if err = json.Unmarshal(scanner.Bytes(), &dp); err != nil {
return nil, err
}
dmaHandler.ImportDatapoints(&dp)
dmaHandler.ImportDatapoints(dp)
}
}
dmaHandler.Log.Info("dmbHandler.NewDmbHandler", fmt.Sprintf("%d datapoint imported in %v", dmaHandler.GetNumbersOfDatapoints(), time.Since(s)))
return &dmaHandler, nil
}
@@ -105,18 +104,30 @@ func (d *DBMHandler) SaveDb() (err error) {
return
}
func (d *DBMHandler) CreateDatapoint(typ models.Type, value any, right models.Rights, path string) error {
if err := d.DB.CreateDatapoint(typ, value, right, path); err != nil {
return err
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()+1)
return nil
d.UpdateDatapointValue("System:Datapoints", dp[0].GetValueUint64()+ndp)
return dps, nil
}
func (d *DBMHandler) ImportDatapoints(dps ...*models.Datapoint) error {
func (d *DBMHandler) ImportDatapoints(dps ...models.Datapoint) error {
for _, dp := range dps {
err := d.DB.ImportDatapoint(dp, dp.Path)
err := d.DB.ImportDatapoint(d.Conns, dp, dp.Path)
if err != nil {
return err
}
@@ -127,71 +138,24 @@ func (d *DBMHandler) ImportDatapoints(dps ...*models.Datapoint) error {
}
func (d *DBMHandler) UpdateDatapointValue(path string, value any) error {
return d.DB.UpdateDatapointValue(value, path)
return d.DB.UpdateDatapointValue(d.Conns, value, path)
}
func (d *DBMHandler) RemoveDatapoint(path string) error {
if err := d.DB.RemoveDatapoint(path); err != nil {
return err
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)
}
dp := d.QueryDatapoints(1, "System:Datapoints")
d.UpdateDatapointValue("System:Datapoints", dp[0].GetValueUint64()+1)
return nil
return lsRemoved, nil
}
func (d *DBMHandler) QueryDatapoints(depth int, key string) []*models.Datapoint {
func (d *DBMHandler) QueryDatapoints(depth uint, key string) []*models.Datapoint {
return d.DB.QueryDatapoints(depth, key)
}
func (d *DBMHandler) AddSystemDps() (err error) {
tim := "System:Time"
memory := "System:UsedMemory"
var m runtime.MemStats
var mOld uint64
var tOld int64
err = d.DB.CreateDatapoint(models.LOU, 0, models.Read, "System:Datapoints")
if err != nil {
d.Log.Error("dmb.Handler.AddSystemDps.CreateDatapoint", err.Error())
return
}
err = d.DB.CreateDatapoint(models.STR, nil, models.Read, tim)
if err != nil {
d.Log.Error("dmb.Handler.AddSystemDps.CreateDatapoint", err.Error())
return
}
err = d.DB.CreateDatapoint(models.STR, nil, models.Read, memory)
if err != nil {
d.Log.Error("dmb.Handler.AddSystemDps.CreateDatapoint", err.Error())
return
}
go func() {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
defer cancel()
for {
t := time.Now().UnixMilli()
if tOld != t {
if er := d.DB.UpdateDatapointValue(t, tim); er != nil {
d.Log.Error("dmb.Handler.AddSystemDps.UpdateDatapointValue", er.Error())
}
d.Publish(ctx, OnChange, tim, t)
tOld = t
}
runtime.ReadMemStats(&m)
if m.Sys != mOld {
mem := fmt.Sprintf("%.2f MB", float64(m.Sys)/1024/1024)
if er := d.DB.UpdateDatapointValue(mem, memory); er != nil {
d.Log.Error("dmb.Handler.AddSystemDps.UpdateDatapointValue", er.Error())
}
d.Publish(ctx, OnChange, memory, mem)
mOld = m.Sys
}
time.Sleep(1 * time.Second)
}
}()
return
}

82
dbm/json_data.go Normal file
View File

@@ -0,0 +1,82 @@
package dbm
import (
"net/http"
"github.com/gin-gonic/gin"
json_data "github.com/tecamino/tecamino-json_data"
json_dataModels "github.com/tecamino/tecamino-json_data/models"
)
func (d *DBMHandler) Json_Data(c *gin.Context) {
var err error
payload, err := json_data.ParseRequest(c.Request.Body)
if err != nil {
r := json_data.NewResponse()
r.SendError(err.Error())
c.JSON(http.StatusBadRequest, r)
return
}
respond := json_dataModels.NewResponse()
if payload.Get != nil {
var depth uint = 1
for _, get := range payload.Get {
if get.Query != nil {
depth = get.Query.Depth
}
for _, res := range d.QueryDatapoints(depth, get.Path) {
respond.AddGet(json_dataModels.Get{
Uuid: res.Uuid,
Path: res.Path,
Type: res.Type,
Value: res.Value,
Rights: res.ReadWrite,
Drivers: &res.Drivers,
})
}
}
}
if payload.Set != nil {
respond.Set, err = d.CreateDatapoints(payload.Set...)
if err != nil {
r := json_data.NewResponse()
r.SendError(err.Error())
c.JSON(http.StatusBadRequest, r)
return
}
}
c.JSON(200, respond)
return
}
func (d *DBMHandler) Delete(c *gin.Context) {
var err error
payload, err := json_data.ParseRequest(c.Request.Body)
if err != nil {
r := json_data.NewResponse()
r.SendError(err.Error())
c.JSON(http.StatusBadRequest, r)
return
}
response := json_data.NewResponse()
if payload.Set != nil {
response.Set, err = d.RemoveDatapoint(payload.Set...)
if err != nil {
r := json_data.NewResponse()
r.SendError(err.Error())
c.JSON(http.StatusBadRequest, r)
return
}
}
c.JSON(200, response)
return
}

34
dbm/publish.go
View File

@@ -1,34 +0,0 @@
package dbm
import (
"context"
"github.com/coder/websocket/wsjson"
"github.com/zuadi/tecamino-dbm/models"
)
func (d *DBMHandler) Publish(ctx context.Context, eventType, path string, value any) error {
d.RLock()
defer d.RUnlock()
for _, dp := range d.DB.QueryDatapoints(1, path) {
for id, pub := range dp.Subscribtions {
if client, ok := d.Clients[id]; !ok {
delete(dp.Subscribtions, id)
} else {
if pub.OnChange {
err := wsjson.Write(ctx, client.Conn, models.JsonResponse{
Event: OnChange,
Path: dp.Path,
Value: value,
})
if err != nil {
d.Log.Error("publish.Publish", err.Error())
return err
}
}
}
}
}
return nil
}

91
dbm/subscribe.go
View File

@@ -2,45 +2,74 @@ package dbm
import (
"github.com/coder/websocket/wsjson"
"github.com/zuadi/tecamino-dbm/models"
"golang.org/x/net/context"
json_dataModels "github.com/tecamino/tecamino-json_data/models"
)
func (d *DBMHandler) Subscribe(ctx context.Context, sub *models.Subscribe, id string) {
func (d *DBMHandler) Subscribe(subs []json_dataModels.Subscribe, id string) {
if subs == nil {
return
}
d.RLock()
defer d.RUnlock()
for _, dp := range d.DB.QueryDatapoints(sub.Depth, sub.Path) {
if sub.Driver != nil {
if dp.Drivers == nil {
continue
} else if _, ok := dp.Drivers[*sub.Driver]; !ok {
client, ok := d.Conns.Clients[id]
if !ok {
d.Log.Error("subscribe.Subscribe", "client not found for id "+id)
return
}
response := json_dataModels.NewResponse()
for _, sub := range subs {
for _, dp := range d.DB.QueryDatapoints(sub.Depth, sub.Path) {
if sub.Driver != "" {
if dp.Drivers == nil || dp.Drivers[sub.Driver] == nil {
continue
}
}
dp.AddSubscribtion(id, sub)
response.AddSubscription(json_dataModels.Subscribe{
Uuid: dp.Uuid,
Path: dp.Path,
Value: dp.Value,
Driver: sub.Driver,
Drivers: &dp.Drivers,
})
}
}
if err := wsjson.Write(client.Ctx, client.Conn, response); err != nil {
d.Log.Error("subscribe.Subscribe", err.Error())
}
}
func (d *DBMHandler) Unsubscribe(subs []json_dataModels.Subscribe, id string) {
if subs == nil {
return
}
d.RLock()
defer d.RUnlock()
client, ok := d.Conns.Clients[id]
if !ok {
d.Log.Error("subscribe.Subscribe", "client not found for id "+id)
return
}
response := json_dataModels.NewResponse()
for _, sub := range subs {
for _, dp := range d.DB.QueryDatapoints(sub.Depth, sub.Path) {
if _, ok := dp.Subscriptions[id]; !ok {
continue
}
}
dp.AddSubscribtion(id, sub)
err := wsjson.Write(ctx, d.Clients[id].Conn, models.JsonResponse{
Event: OnCreate,
Path: dp.Path,
Value: dp.Value,
})
if err != nil {
d.Log.Error("subscribe.Subscribe", err.Error())
dp.RemoveSubscribtion(id)
response.AddUnsubscription(json_dataModels.Subscribe{
Uuid: dp.Uuid,
Path: dp.Path,
})
}
}
}
func (d *DBMHandler) Unsubscribe(ctx context.Context, sub *models.Subscribe, id string) error {
d.RLock()
defer d.RUnlock()
for _, dp := range d.DB.QueryDatapoints(sub.Depth, sub.Path) {
if _, ok := dp.Subscribtions[id]; !ok {
continue
}
dp.RemoveSubscribtion(id)
if err := wsjson.Write(client.Ctx, client.Conn, response); err != nil {
d.Log.Error("subscribe.Subscribe", err.Error())
}
return nil
}

97
dbm/system.go Normal file
View File

@@ -0,0 +1,97 @@
package dbm
import (
"fmt"
"runtime"
"time"
"github.com/tecamino/tecamino-dbm/utils"
json_dataModels "github.com/tecamino/tecamino-json_data/models"
)
func (d *DBMHandler) AddSystemDps() (err error) {
path := "System:Datapoints"
typ := json_dataModels.LOU
rights := json_dataModels.Read
_, err = d.DB.CreateDatapoints(d.Conns, json_dataModels.Set{Path: path, Value: 0, Type: typ, Rights: rights})
if err != nil {
d.Log.Error("dmb.Handler.AddSystemDps", err.Error())
return err
}
dp := d.QueryDatapoints(1, path)
d.UpdateDatapointValue(path, dp[0].GetValueUint64()+1)
if err = d.GoSystemTime(); err != nil {
return err
}
if err = d.GoSystemMemory(); err != nil {
return err
}
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, t, 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
}

66
dbm/webSocket.go
View File

@@ -1,13 +1,14 @@
package dbm
import (
"context"
"errors"
"fmt"
"github.com/coder/websocket"
"github.com/coder/websocket/wsjson"
"github.com/gin-gonic/gin"
"github.com/zuadi/tecamino-dbm/models"
"github.com/tecamino/tecamino-dbm/auth"
json_dataModels "github.com/tecamino/tecamino-json_data/models"
)
const (
@@ -17,62 +18,45 @@ const (
)
func (d *DBMHandler) WebSocket(c *gin.Context) {
// id, err := auth.GetIDFromAuth(c)
// if err != nil {
// d.Log.Error("dbmHandler.webSocket.Websocket", "error GetIDFromAuth: "+err.Error())
// return
// }
id := "test"
id, err := auth.GetIDFromQuery(c)
if err != nil {
d.Log.Error("dbmHandler.webSocket.Websocket", "error GetIDFromQuery: "+err.Error())
return
}
d.Log.Debug("dbmHandler.webSocket.Websocket", "authorization id token: "+id)
ctx, cancel := context.WithCancel(c.Request.Context())
defer cancel()
err := d.Clients.ConnectRecievingWsConnection(id, c)
defer d.Clients.RemoveClient(id)
err = d.Conns.ConnectRecievingWsConnection(id, c)
defer d.Conns.RemoveClient(id)
if err != nil {
d.Log.Error("dbmHandler.webSocket.Websocket", "error connecting recieving websocket conection: "+err.Error())
return
}
defer d.Clients.DisconnectWsConnection(id, websocket.StatusInternalError, "Internal error")
defer d.Conns.DisconnectWsConnection(id, websocket.StatusInternalError, "Internal error")
//Read loop
for {
request, err := d.readJsonData(ctx, id)
request, err := d.readJsonData(id)
if err != nil {
break
}
// Subscribe
if request.Subscribe != nil {
go func() {
for _, sub := range *request.Subscribe {
d.Subscribe(ctx, &sub, id)
}
}()
}
go d.Subscribe(request.Subscribe, id)
// Unsubscribe
if request.Unsubscribe != nil {
for _, unsub := range *request.Unsubscribe {
err = d.Unsubscribe(ctx, &unsub, id)
if err != nil {
d.Log.Error("dbmHandler.WebSocket unsubscribe", err)
break
}
}
}
go d.Unsubscribe(request.Unsubscribe, id)
if err != nil {
break
}
}
}
func (d *DBMHandler) readJsonData(ctx context.Context, id string) (*models.JsonData, error) {
var request models.JsonData
err := wsjson.Read(ctx, d.Clients[id].Conn, &request)
func (d *DBMHandler) readJsonData(id string) (request json_dataModels.Request, err error) {
client, ok := d.Conns.Clients[id]
if !ok {
return request, errors.New("client id not found")
}
err = wsjson.Read(client.Ctx, client.Conn, &request)
if err != nil {
code := websocket.CloseStatus(err)
@@ -81,11 +65,11 @@ func (d *DBMHandler) readJsonData(ctx context.Context, id string) (*models.JsonD
websocket.StatusGoingAway,
websocket.StatusNoStatusRcvd:
d.Log.Info("webSocket.readJsonData", fmt.Sprintf("WebSocket closed: %v (code: %v)\n", err, code))
return nil, err
return
default:
d.Log.Error("webSocket.readJsonData", fmt.Sprintf("WebSocket read error: %v (code: %v)\n", err, code))
return nil, err
return
}
}
return &request, nil
return
}