major change of websocket dbmHandler structure

2025-04-23 21:53:01 +02:00
parent a1f947e24a
commit 0a137c9d86
35 changed files with 1676 additions and 424 deletions
--- a/dbm/dbmHandler.go
+++ b/dbm/dbmHandler.go
@@ -0,0 +1,197 @@
+package dbm
+
+import (
+	"bufio"
+	"context"
+	"encoding/json"
+	"fmt"
+	"os"
+	"runtime"
+	"sync"
+	"time"
+
+	"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
+	sync.RWMutex
+	Log *logging.Logger
+	arg *args.Args
+}
+
+// initialze new Database Manager
+// it will call cli arguments
+func NewDbmHandler(a *args.Args) (*DBMHandler, error) {
+
+	logger, err := logging.NewLogger("dbmServer.log", &logging.Config{
+		MaxSize:     1,
+		MaxBackup:   3,
+		MaxAge:      28,
+		Debug:       a.Debug,
+		TerminalOut: true,
+	})
+
+	if err != nil {
+		return nil, err
+	}
+	logger.Info("main", "start dma")
+
+	// Initialize dtabase manager handler
+	dmaHandler := DBMHandler{
+		arg:      a,
+		filePath: fmt.Sprintf("%s/%s.dma", a.RootDir, a.DMAFile),
+		Log:      logger,
+		Clients:  serverModels.NewClients(),
+	}
+
+	// initialize system datapoint and periodically update it
+	if err := dmaHandler.AddSystemDps(); err != nil {
+		return nil, err
+	}
+
+	// check if dtabase file exists to load data
+	if _, err := os.Stat(dmaHandler.filePath); err == nil {
+
+		f, err := os.Open(dmaHandler.filePath)
+		if err != nil {
+			return nil, err
+		}
+		defer f.Close()
+
+		// read in dtaabase file content
+		scanner := bufio.NewScanner(f)
+
+		for scanner.Scan() {
+			dp := models.Datapoint{}
+			err = json.Unmarshal(scanner.Bytes(), &dp)
+			if err != nil {
+				return nil, err
+			}
+			dmaHandler.ImportDatapoints(&dp)
+		}
+	}
+
+	return &dmaHandler, nil
+}
+
+func (d *DBMHandler) SaveDb() (err error) {
+	f, err := os.OpenFile(d.filePath, os.O_CREATE|os.O_RDWR|os.O_TRUNC, 0666)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+
+	for _, dp := range d.DB.GetAllDatapoints(0) {
+		b, er := json.Marshal(dp)
+		if er != nil {
+			return er
+		}
+
+		_, err = f.Write(b)
+		if err != nil {
+			return
+		}
+		_, err = f.Write([]byte("\n"))
+		if err != nil {
+			return
+		}
+	}
+	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
+	}
+	dp := d.QueryDatapoints(1, "System:Datapoints")
+	d.UpdateDatapointValue("System:Datapoints", dp[0].GetValueUint64()+1)
+	return nil
+}
+
+func (d *DBMHandler) ImportDatapoints(dps ...*models.Datapoint) error {
+	for _, dp := range dps {
+		err := d.DB.ImportDatapoint(dp, dp.Path)
+		if err != nil {
+			return err
+		}
+		dp := d.QueryDatapoints(1, "System:Datapoints")
+		d.UpdateDatapointValue("System:Datapoints", dp[0].GetValueUint64()+1)
+	}
+	return nil
+}
+
+func (d *DBMHandler) UpdateDatapointValue(path string, value any) error {
+	return d.DB.UpdateDatapointValue(value, path)
+}
+
+func (d *DBMHandler) RemoveDatapoint(path string) error {
+	if err := d.DB.RemoveDatapoint(path); err != nil {
+		return err
+	}
+	dp := d.QueryDatapoints(1, "System:Datapoints")
+	d.UpdateDatapointValue("System:Datapoints", dp[0].GetValueUint64()+1)
+	return nil
+}
+
+func (d *DBMHandler) QueryDatapoints(depth int, 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
+}
--- a/dbm/publish.go
+++ b/dbm/publish.go
@@ -0,0 +1,34 @@
+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
+}
--- a/dbm/subscribe.go
+++ b/dbm/subscribe.go
@@ -0,0 +1,46 @@
+package dbm
+
+import (
+	"github.com/coder/websocket/wsjson"
+	"github.com/zuadi/tecamino-dbm/models"
+	"golang.org/x/net/context"
+)
+
+func (d *DBMHandler) Subscribe(ctx context.Context, sub *models.Subscribe, id string) {
+	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 {
+				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())
+		}
+	}
+}
+
+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)
+	}
+	return nil
+}
--- a/dbm/webSocket.go
+++ b/dbm/webSocket.go
@@ -0,0 +1,91 @@
+package dbm
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/coder/websocket"
+	"github.com/coder/websocket/wsjson"
+	"github.com/gin-gonic/gin"
+	"github.com/zuadi/tecamino-dbm/models"
+)
+
+const (
+	OnCreate = "onCreate"
+	OnChange = "onChange"
+	OnDelete = "onDelete"
+)
+
+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"
+
+	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)
+	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")
+
+	//Read loop
+	for {
+		request, err := d.readJsonData(ctx, id)
+		if err != nil {
+			break
+		}
+
+		// Subscribe
+
+		if request.Subscribe != nil {
+			go func() {
+				for _, sub := range *request.Subscribe {
+					d.Subscribe(ctx, &sub, 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
+				}
+			}
+		}
+
+		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)
+	if err != nil {
+		code := websocket.CloseStatus(err)
+
+		switch code {
+		case websocket.StatusNormalClosure,
+			websocket.StatusGoingAway,
+			websocket.StatusNoStatusRcvd:
+			d.Log.Info("webSocket.readJsonData", fmt.Sprintf("WebSocket closed: %v (code: %v)\n", err, code))
+			return nil, err
+		default:
+			d.Log.Error("webSocket.readJsonData", fmt.Sprintf("WebSocket read error: %v (code: %v)\n", err, code))
+			return nil, err
+		}
+	}
+	return &request, nil
+}