fix double use of name

dbm/dbmHandler.go

@@ -131,8 +131,8 @@ func (d *DBMHandler) ImportDatapoints(dps ...models.Datapoint) error {
 		if err != nil {
 			return err
 		}
-		dp := d.QueryDatapoints(1, "System:Datapoints")
+		dps := d.QueryDatapoints(1, "System:Datapoints")
-		d.UpdateDatapointValue("System:Datapoints", dp[0].GetValueUint64()+1)
+		d.UpdateDatapointValue("System:Datapoints", dps[0].GetValueUint64()+1)
 	}
 	return nil
 }

dbm/get.go

@@ -0,0 +1,37 @@
+package dbm
+
+import (
+	json_data "github.com/tecamino/tecamino-json_data"
+	json_dataModels "github.com/tecamino/tecamino-json_data/models"
+)
+
+func (d *DBMHandler) Get(gets []json_dataModels.Get, id, id2 string) {
+	if gets == nil {
+		return
+	}
+	d.RLock()
+	defer d.RUnlock()
+
+	r := json_data.NewResponse()
+	r.Id = id2
+	for _, get := range gets {
+		var depth uint = 1
+		if get.Query != nil {
+			depth = get.Query.Depth
+		}
+
+		for _, dp := range d.DB.QueryDatapoints(depth, get.Path) {
+			r.AddGet(json_dataModels.Get{
+				Uuid:   dp.Uuid,
+				Path:   dp.Path,
+				Type:   dp.Type,
+				Value:  dp.Value,
+				Rights: dp.ReadWrite,
+			})
+		}
+	}
+
+	if err := d.Conns.SendResponse(id, r); err != nil {
+		d.Log.Error("get.Get", err.Error())
+	}
+}

dbm/subscribe.go

@@ -1,24 +1,18 @@
 package dbm
 
 import (
-	"github.com/coder/websocket/wsjson"
 	json_dataModels "github.com/tecamino/tecamino-json_data/models"
 )
 
-func (d *DBMHandler) Subscribe(subs []json_dataModels.Subscribe, id string) {
+func (d *DBMHandler) Subscribe(subs []json_dataModels.Subscribe, id, id2 string) {
 	if subs == nil {
 		return
 	}
 	d.RLock()
 	defer d.RUnlock()
 
-	client, ok := d.Conns.Clients[id]
+	r := json_dataModels.NewResponse()
-	if !ok {
+	r.Id = id2
-		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) {
@@ -28,7 +22,7 @@ func (d *DBMHandler) Subscribe(subs []json_dataModels.Subscribe, id string) {
 				}
 			}
 			dp.AddSubscribtion(id, sub)
-			response.AddSubscription(json_dataModels.Subscribe{
+			r.AddSubscription(json_dataModels.Subscribe{
 				Uuid:    dp.Uuid,
 				Path:    dp.Path,
 				Value:   dp.Value,
@@ -37,7 +31,8 @@ func (d *DBMHandler) Subscribe(subs []json_dataModels.Subscribe, id string) {
 			})
 		}
 	}
-	if err := wsjson.Write(client.Ctx, client.Conn, response); err != nil {
+
+	if err := d.Conns.SendResponse(id, r); err != nil {
 		d.Log.Error("subscribe.Subscribe", err.Error())
 	}
 }
@@ -49,13 +44,7 @@ func (d *DBMHandler) Unsubscribe(subs []json_dataModels.Subscribe, id string) {
 	d.RLock()
 	defer d.RUnlock()
 
-	client, ok := d.Conns.Clients[id]
+	r := json_dataModels.NewResponse()
-	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) {
@@ -63,13 +52,14 @@ func (d *DBMHandler) Unsubscribe(subs []json_dataModels.Subscribe, id string) {
 				continue
 			}
 			dp.RemoveSubscribtion(id)
-			response.AddUnsubscription(json_dataModels.Subscribe{
+			r.AddUnsubscription(json_dataModels.Subscribe{
 				Uuid: dp.Uuid,
 				Path: dp.Path,
 			})
 		}
 	}
-	if err := wsjson.Write(client.Ctx, client.Conn, response); err != nil {
+
-		d.Log.Error("subscribe.Subscribe", err.Error())
+	if err := d.Conns.SendResponse(id, r); err != nil {
+		d.Log.Error("subscribe.Unsubscribe", err.Error())
 	}
 }

dbm/system.go

@@ -54,7 +54,7 @@ func (d *DBMHandler) GoSystemTime() error {
 		for {
 			t := time.Now().UnixMilli()
 			if tOld != t {
-				if er := d.DB.UpdateDatapointValue(d.Conns, t, path); er != nil {
+				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

dbm/webSocket.go

@@ -1,11 +1,12 @@
 package dbm
 
 import (
+	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
 
 	"github.com/coder/websocket"
-	"github.com/coder/websocket/wsjson"
 	"github.com/gin-gonic/gin"
 	"github.com/tecamino/tecamino-dbm/auth"
 	json_dataModels "github.com/tecamino/tecamino-json_data/models"
@@ -35,31 +36,46 @@ func (d *DBMHandler) WebSocket(c *gin.Context) {
 
 	//Read loop
 	for {
+
 		request, err := d.readJsonData(id)
 		if err != nil {
+			d.Log.Error("websocket.WebSocket", err.Error())
 			break
 		}
 
 		// Sets
-		go d.Set(request.Set)
+
+		d.Get(request.Get, id, request.Id)
+		// Sets
+		d.Set(request.Set)
 
 		// Subscribe
-		go d.Subscribe(request.Subscribe, id)
+		d.Subscribe(request.Subscribe, id, request.Id)
 
 		// Unsubscribe
-		go d.Unsubscribe(request.Unsubscribe, id)
+		d.Unsubscribe(request.Unsubscribe, id)
 
+		request.Get = make([]json_dataModels.Get, 0)
+		request.Set = make([]json_dataModels.Set, 0)
+		request.Subscribe = make([]json_dataModels.Subscribe, 0)
+		request.Unsubscribe = make([]json_dataModels.Subscribe, 0)
+		request = nil
 	}
 }
 
-func (d *DBMHandler) readJsonData(id string) (request json_dataModels.Request, err error) {
+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)
+	_, reader, err := client.Conn.Reader(client.Ctx)
+	if err != nil {
+		return request, err
+	}
+
+	b, err := io.ReadAll(reader)
 	if err != nil {
 		code := websocket.CloseStatus(err)
 
@@ -74,5 +90,10 @@ func (d *DBMHandler) readJsonData(id string) (request json_dataModels.Request, e
 			return
 		}
 	}
+
+	if err := json.Unmarshal(b, &request); err != nil {
+		return request, err
+	}
+
 	return
 }

models/datapoint.go

@@ -0,0 +1,393 @@
+package models
+
+import (
+	"errors"
+	"fmt"
+	"regexp"
+	"strings"
+	"time"
+
+	"github.com/google/uuid"
+	serverModels "github.com/tecamino/tecamino-dbm/server/models"
+	"github.com/tecamino/tecamino-dbm/utils"
+	json_data "github.com/tecamino/tecamino-json_data"
+	json_dataModels "github.com/tecamino/tecamino-json_data/models"
+)
+
+const (
+	OnCreate = "onCreate"
+	OnChange = "onChange"
+	OnDelete = "onDelete"
+)
+
+type Datapoint struct {
+	Datapoints     map[string]*Datapoint   `json:"-"`
+	Uuid           uuid.UUID               `json:"uuid"`
+	Path           string                  `json:"path"`
+	Value          any                     `json:"value,omitempty"`
+	CreateDateTime int64                   `json:"createDateTime,omitempty"`
+	UpdateDateTime int64                   `json:"updateDateTime,omitempty"`
+	Type           json_dataModels.Type    `json:"type"`
+	ReadWrite      json_dataModels.Rights  `json:"readWrite"`
+	Drivers        json_dataModels.Drivers `json:"drivers,omitempty"`
+	Subscriptions  Subscriptions           `json:"-"`
+}
+
+func (d *Datapoint) Set(path string, set json_dataModels.Set) (bool, error) {
+	var changed bool
+	if path != "" {
+		changed = true
+		d.Path = path
+	}
+
+	if set.Type != "" {
+		changed = true
+		d.Type = set.Type
+	}
+
+	if d.Type != "" {
+		if d.Value == nil && set.Value == nil {
+			changed = true
+			d.Value = d.Type.DefaultValue()
+		} else if d.Value != d.Type.ConvertValue(set.Value) {
+			changed = true
+			d.Value = d.Type.ConvertValue(set.Value)
+		}
+
+	}
+	if set.Rights != "" {
+		changed = true
+		d.ReadWrite = set.Rights.GetRights()
+	}
+
+	if changed {
+		d.UpdateDateTime = time.Now().UnixMilli()
+	}
+
+	if set.Driver == nil {
+		return changed, nil
+	}
+	if set.Driver.Type == "" {
+		return changed, errors.New("driver type missing")
+	}
+	if set.Driver.Bus == "" {
+		return changed, errors.New("driver bus name missing")
+	}
+
+	if d.Drivers == nil {
+		d.Drivers = json_dataModels.NewDrivers()
+	}
+	d.Drivers.AddDriver(set.Driver.Type, set.Driver.Bus, set.Driver.Address)
+	d.UpdateDateTime = time.Now().UnixMilli()
+
+	return changed, nil
+}
+
+func (d *Datapoint) GetValueUint64() uint64 {
+	return utils.Uint64From(d.Value)
+}
+
+func (d *Datapoint) CreateDatapoints(conns *serverModels.Connections, sets ...json_dataModels.Set) (created []json_dataModels.Set, err error) {
+	if len(sets) == 0 {
+		return
+	}
+	for _, dp := range sets {
+		parts := strings.Split(dp.Path, ":")
+
+		current := d
+		for i, part := range parts {
+			if current.Datapoints == nil {
+				current.Datapoints = make(map[string]*Datapoint)
+			}
+
+			if i == len(parts)-1 {
+				// Leaf node: create or update datapoint
+				if existing, ok := current.Datapoints[part]; ok {
+					publish, err := existing.Set("", dp)
+					if err != nil {
+						return nil, err
+					}
+					created = append(created, json_dataModels.Set{
+						Uuid:    existing.Uuid,
+						Path:    existing.Path,
+						Type:    existing.Type,
+						Value:   existing.Value,
+						Rights:  existing.ReadWrite,
+						Drivers: &existing.Drivers,
+						Updated: true,
+					})
+
+					if publish {
+						existing.Publish(conns, OnChange)
+					}
+				} else {
+					ndp := Datapoint{
+						Uuid:           uuid.New(),
+						CreateDateTime: time.Now().UnixMilli(),
+						Subscriptions:  InitSubscribtion(),
+					}
+					// Create new
+					current.Datapoints[part] = &ndp
+					publish, err := ndp.Set(strings.Join(parts, ":"), dp)
+					if err != nil {
+						return nil, err
+					}
+					created = append(created, json_dataModels.Set{
+						Uuid:   ndp.Uuid,
+						Path:   ndp.Path,
+						Type:   ndp.Type,
+						Value:  ndp.Value,
+						Rights: ndp.ReadWrite,
+						Driver: dp.Driver,
+					})
+					if publish {
+						current.Publish(conns, OnChange)
+					}
+				}
+			}
+
+			// Traverse or create intermediate datapoints
+			if next, ok := current.Datapoints[part]; ok {
+				current = next
+			} else {
+				newDp := &Datapoint{
+					Uuid:           uuid.New(),
+					Path:           strings.Join(parts[:i+1], ":"),
+					Type:           json_dataModels.NONE,
+					CreateDateTime: time.Now().UnixMilli(),
+					UpdateDateTime: time.Now().UnixMilli(),
+					Subscriptions:  InitSubscribtion(),
+				}
+
+				created = append(created, json_dataModels.Set{
+					Uuid:   newDp.Uuid,
+					Path:   newDp.Path,
+					Type:   newDp.Type,
+					Value:  newDp.Value,
+					Rights: newDp.ReadWrite,
+				})
+
+				if dp.Rights != "" {
+					newDp.ReadWrite = dp.Rights.GetRights()
+				}
+
+				current.Datapoints[part] = newDp
+				current = newDp
+			}
+		}
+	}
+	return
+}
+
+func (d *Datapoint) ImportDatapoint(conns *serverModels.Connections, dp Datapoint, path string) error {
+	parts := strings.Split(dp.Path, ":")
+
+	current := d
+	for i, part := range parts {
+		if current.Datapoints == nil {
+			current.Datapoints = make(map[string]*Datapoint)
+		}
+
+		if i == len(parts)-1 {
+			// Leaf node: import the datapoint
+			if existing, ok := current.Datapoints[part]; ok {
+				existing.Type = dp.Type
+				existing.Value = dp.Type.ConvertValue(dp.Value)
+				existing.ReadWrite = dp.ReadWrite.GetRights()
+				existing.UpdateDateTime = time.Now().UnixMilli()
+				dp.Publish(conns, OnChange)
+			} else {
+				dp.Path = strings.Join(parts, ":")
+				dp.ReadWrite = dp.ReadWrite.GetRights()
+				dp.UpdateDateTime = time.Now().UnixMilli()
+				dp.Subscriptions = InitSubscribtion()
+				current.Datapoints[part] = &dp
+				dp.Publish(conns, OnChange)
+			}
+			return nil
+		}
+
+		// Traverse or create intermediate nodes
+		if next, ok := current.Datapoints[part]; ok {
+			current = next
+		} else {
+			newDp := &Datapoint{
+				Path:           strings.Join(parts[:i+1], ":"),
+				Type:           json_dataModels.NONE,
+				ReadWrite:      dp.ReadWrite.GetRights(),
+				UpdateDateTime: time.Now().UnixMilli(),
+			}
+			newDp.ReadWrite = newDp.ReadWrite.GetRights()
+			current.Datapoints[part] = newDp
+			current = newDp
+		}
+	}
+	return nil
+}
+
+func (d *Datapoint) UpdateDatapointValue(conns *serverModels.Connections, value any, path string) error {
+
+	paths := strings.Split(path, ":")
+
+	current := d
+	for i, part := range paths {
+		dp, ok := current.Datapoints[part]
+		if !ok {
+			return fmt.Errorf("datapoint path not found: %s (at %s)", path, part)
+		}
+		if i == len(paths)-1 {
+			dp.Value = dp.Type.ConvertValue(value)
+			dp.UpdateDateTime = time.Now().UnixMilli()
+			dp.Publish(conns, OnChange)
+			return nil
+		}
+		current = dp
+	}
+	return nil
+}
+
+func (d *Datapoint) UpdateValue(conns *serverModels.Connections, value any) error {
+	d.Value = d.Type.ConvertValue(value)
+	d.UpdateDateTime = time.Now().UnixMilli()
+	d.Publish(conns, OnChange)
+	return nil
+}
+
+func (d *Datapoint) RemoveDatapoint(conns *serverModels.Connections, set json_dataModels.Set) (json_dataModels.Set, error) {
+	parts := strings.Split(set.Path, ":")
+
+	if len(parts) < 1 {
+		return json_dataModels.Set{}, fmt.Errorf("invalid path: '%s'", set.Path)
+	}
+
+	current := d
+	for i := range len(parts) - 1 {
+		next, ok := current.Datapoints[parts[i]]
+		if !ok {
+			return json_dataModels.Set{}, fmt.Errorf("path not found: '%s'", strings.Join(parts[:i+1], ":"))
+		}
+		current = next
+	}
+
+	toDelete := parts[len(parts)-1]
+	if dp, ok := current.Datapoints[toDelete]; ok {
+		dp.Publish(conns, OnDelete)
+		delete(current.Datapoints, toDelete)
+		return json_dataModels.Set{
+			Uuid: set.Uuid,
+			Path: set.Path,
+		}, nil
+	}
+	return json_dataModels.Set{}, fmt.Errorf("datapoint '%s' not found", set.Path)
+}
+
+func (d *Datapoint) GetAllDatapoints(depth uint) (dps Datapoints) {
+
+	var dfs func(dp *Datapoint, currentDepth uint)
+	dfs = func(dp *Datapoint, currentDepth uint) {
+		if depth == 1 {
+			return
+		} else if depth == 0 {
+			// Return all descendants
+			for _, child := range dp.Datapoints {
+				dps = append(dps, child)
+				dfs(child, currentDepth+1)
+			}
+			return
+		}
+
+		if currentDepth == depth-1 {
+			return
+		}
+
+		for _, child := range dp.Datapoints {
+			dps = append(dps, child)
+			dfs(child, currentDepth+1)
+		}
+	}
+	dps = append(dps, d)
+	dfs(d, 0)
+	dps.SortSlice()
+	return
+}
+
+func (d *Datapoint) QueryDatapoints(depth uint, path string) (dps Datapoints) {
+	parts := strings.Split(path, ":")
+
+	var dfs func(current *Datapoint, index int)
+	dfs = func(current *Datapoint, index int) {
+
+		if index == len(parts) {
+			dps = append(dps, current.GetAllDatapoints(depth)...)
+			return
+		}
+
+		pattern := "^" + parts[index] + "$"
+		re, err := regexp.Compile(pattern)
+		if err != nil {
+			return
+		}
+
+		for name, dp := range current.Datapoints {
+			if re.MatchString(name) {
+				dfs(dp, index+1)
+			}
+		}
+	}
+
+	dfs(d, 0)
+	dps.SortSlice()
+	return
+}
+
+func (d *Datapoint) AddSubscribtion(id string, sub json_dataModels.Subscribe) {
+	if d.Subscriptions == nil {
+		return
+	}
+
+	if s, ok := d.Subscriptions[id]; ok {
+		s.OnCreate = sub.OnCreate
+		s.OnChange = sub.OnChange
+		s.OnDelete = sub.OnDelete
+	} else {
+		d.Subscriptions[id] = &Subscription{
+			OnCreate: sub.OnCreate,
+			OnChange: sub.OnChange,
+			OnDelete: sub.OnDelete,
+		}
+	}
+}
+
+func (d *Datapoint) RemoveSubscribtion(id string) {
+	if _, ok := d.Subscriptions[id]; !ok {
+		return
+	}
+	delete(d.Subscriptions, id)
+}
+
+func (d *Datapoint) Publish(conns *serverModels.Connections, eventType string) error {
+	if conns.Clients == nil {
+		return nil
+	}
+	conns.RLock()
+	defer conns.RUnlock()
+
+	for id := range d.Subscriptions {
+		if _, ok := conns.Clients[id]; !ok {
+			delete(d.Subscriptions, id)
+		} else {
+			r := json_data.NewResponse()
+			r.AddUPublish(json_dataModels.Publish{
+				Event: eventType,
+				Uuid:  d.Uuid,
+				Path:  d.Path,
+				Value: d.Value,
+			})
+
+			if err := conns.SendResponse(id, r); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}

models/datapoints.go

@@ -1,392 +1,12 @@
 package models
 
-import (
+import "sort"
-	"errors"
-	"fmt"
-	"regexp"
-	"strings"
-	"time"
 
-	"github.com/coder/websocket/wsjson"
+type Datapoints []*Datapoint
-	"github.com/google/uuid"
-	serverModels "github.com/tecamino/tecamino-dbm/server/models"
-	"github.com/tecamino/tecamino-dbm/utils"
-	json_data "github.com/tecamino/tecamino-json_data"
-	json_dataModels "github.com/tecamino/tecamino-json_data/models"
-)
 
-const (
+func (d *Datapoints) SortSlice() {
-	OnCreate = "onCreate"
+	// Sort by Path before processing
-	OnChange = "onChange"
+	sort.Slice(*d, func(i, j int) bool {
-	OnDelete = "onDelete"
+		return (*d)[i].Path < (*d)[j].Path
-)
-
-type Datapoint struct {
-	Datapoints     map[string]*Datapoint   `json:"-"`
-	Uuid           uuid.UUID               `json:"uuid"`
-	Path           string                  `json:"path"`
-	Value          any                     `json:"value,omitempty"`
-	CreateDateTime int64                   `json:"createDateTime,omitempty"`
-	UpdateDateTime int64                   `json:"updateDateTime,omitempty"`
-	Type           json_dataModels.Type    `json:"type"`
-	ReadWrite      json_dataModels.Rights  `json:"readWrite"`
-	Drivers        json_dataModels.Drivers `json:"drivers,omitempty"`
-	Subscriptions  Subscriptions           `json:"-"`
-}
-
-func (d *Datapoint) Set(path string, set json_dataModels.Set) (bool, error) {
-	var changed bool
-	if path != "" {
-		changed = true
-		d.Path = path
-	}
-
-	if set.Type != "" {
-		changed = true
-		d.Type = set.Type
-	}
-
-	if d.Type != "" {
-		if d.Value == nil && set.Value == nil {
-			changed = true
-			d.Value = d.Type.DefaultValue()
-		} else if d.Value != d.Type.ConvertValue(set.Value) {
-			changed = true
-			d.Value = d.Type.ConvertValue(set.Value)
-		}
-
-	}
-	if set.Rights != "" {
-		changed = true
-		d.ReadWrite = set.Rights.GetRights()
-	}
-
-	if changed {
-		d.UpdateDateTime = time.Now().UnixMilli()
-	}
-
-	if set.Driver == nil {
-		return changed, nil
-	}
-	if set.Driver.Type == "" {
-		return changed, errors.New("driver type missing")
-	}
-	if set.Driver.Bus == "" {
-		return changed, errors.New("driver bus name missing")
-	}
-
-	if d.Drivers == nil {
-		d.Drivers = json_dataModels.NewDrivers()
-	}
-	d.Drivers.AddDriver(set.Driver.Type, set.Driver.Bus, set.Driver.Address)
-	d.UpdateDateTime = time.Now().UnixMilli()
-
-	return changed, nil
-}
-
-func (d *Datapoint) GetValueUint64() uint64 {
-	return utils.Uint64From(d.Value)
-}
-
-func (d *Datapoint) CreateDatapoints(conns *serverModels.Connections, sets ...json_dataModels.Set) (created []json_dataModels.Set, err error) {
-	if len(sets) == 0 {
-		return
-	}
-	for _, dp := range sets {
-		parts := strings.Split(dp.Path, ":")
-
-		current := d
-		for i, part := range parts {
-			if current.Datapoints == nil {
-				current.Datapoints = make(map[string]*Datapoint)
-			}
-
-			if i == len(parts)-1 {
-				// Leaf node: create or update datapoint
-				if existing, ok := current.Datapoints[part]; ok {
-					publish, err := existing.Set("", dp)
-					if err != nil {
-						return nil, err
-					}
-					created = append(created, json_dataModels.Set{
-						Uuid:    existing.Uuid,
-						Path:    existing.Path,
-						Type:    existing.Type,
-						Value:   existing.Value,
-						Rights:  existing.ReadWrite,
-						Drivers: &existing.Drivers,
-						Updated: true,
 	})
-
-					if publish {
-						existing.Publish(conns, OnChange)
-					}
-				} else {
-					ndp := Datapoint{
-						Uuid:           uuid.New(),
-						CreateDateTime: time.Now().UnixMilli(),
-						Subscriptions:  InitSubscribtion(),
-					}
-					// Create new
-					current.Datapoints[part] = &ndp
-					publish, err := ndp.Set(strings.Join(parts, ":"), dp)
-					if err != nil {
-						return nil, err
-					}
-					created = append(created, json_dataModels.Set{
-						Uuid:   ndp.Uuid,
-						Path:   ndp.Path,
-						Type:   ndp.Type,
-						Value:  ndp.Value,
-						Rights: ndp.ReadWrite,
-						Driver: dp.Driver,
-					})
-					if publish {
-						current.Publish(conns, OnChange)
-					}
-				}
-			}
-
-			// Traverse or create intermediate datapoints
-			if next, ok := current.Datapoints[part]; ok {
-				current = next
-			} else {
-				newDp := &Datapoint{
-					Uuid:           uuid.New(),
-					Path:           strings.Join(parts[:i+1], ":"),
-					Type:           json_dataModels.NONE,
-					CreateDateTime: time.Now().UnixMilli(),
-					UpdateDateTime: time.Now().UnixMilli(),
-					Subscriptions:  InitSubscribtion(),
-				}
-
-				created = append(created, json_dataModels.Set{
-					Uuid:   newDp.Uuid,
-					Path:   newDp.Path,
-					Type:   newDp.Type,
-					Value:  newDp.Value,
-					Rights: newDp.ReadWrite,
-				})
-
-				if dp.Rights != "" {
-					newDp.ReadWrite = dp.Rights.GetRights()
-				}
-
-				current.Datapoints[part] = newDp
-				current = newDp
-			}
-		}
-	}
-	return
-}
-
-func (d *Datapoint) ImportDatapoint(conns *serverModels.Connections, dp Datapoint, path string) error {
-	parts := strings.Split(dp.Path, ":")
-
-	current := d
-	for i, part := range parts {
-		if current.Datapoints == nil {
-			current.Datapoints = make(map[string]*Datapoint)
-		}
-
-		if i == len(parts)-1 {
-			// Leaf node: import the datapoint
-			if existing, ok := current.Datapoints[part]; ok {
-				existing.Type = dp.Type
-				existing.Value = current.Type.ConvertValue(dp.Value)
-				existing.ReadWrite = dp.ReadWrite.GetRights()
-				existing.UpdateDateTime = time.Now().UnixMilli()
-				dp.Publish(conns, OnChange)
-			} else {
-				dp.Path = strings.Join(parts, ":")
-				dp.ReadWrite = dp.ReadWrite.GetRights()
-				dp.UpdateDateTime = time.Now().UnixMilli()
-				dp.Subscriptions = InitSubscribtion()
-				current.Datapoints[part] = &dp
-				dp.Publish(conns, OnChange)
-			}
-			return nil
-		}
-
-		// Traverse or create intermediate nodes
-		if next, ok := current.Datapoints[part]; ok {
-			current = next
-		} else {
-			newDp := &Datapoint{
-				Path:           strings.Join(parts[:i+1], ":"),
-				Type:           json_dataModels.NONE,
-				ReadWrite:      dp.ReadWrite.GetRights(),
-				UpdateDateTime: time.Now().UnixMilli(),
-			}
-			newDp.ReadWrite = newDp.ReadWrite.GetRights()
-			current.Datapoints[part] = newDp
-			current = newDp
-		}
-	}
-	return nil
-}
-
-func (d *Datapoint) UpdateDatapointValue(conns *serverModels.Connections, value any, path string) error {
-
-	paths := strings.Split(path, ":")
-
-	current := d
-	for i, part := range paths {
-		dp, ok := current.Datapoints[part]
-		if !ok {
-			return fmt.Errorf("datapoint path not found: %s (at %s)", path, part)
-		}
-		if i == len(paths)-1 {
-			dp.Value = dp.Type.ConvertValue(value)
-			dp.UpdateDateTime = time.Now().UnixMilli()
-			dp.Publish(conns, OnChange)
-			return nil
-		}
-		current = dp
-	}
-	return nil
-}
-
-func (d *Datapoint) UpdateValue(conns *serverModels.Connections, value any) error {
-	d.Value = d.Type.ConvertValue(value)
-	d.UpdateDateTime = time.Now().UnixMilli()
-	d.Publish(conns, OnChange)
-	return nil
-}
-
-func (d *Datapoint) RemoveDatapoint(conns *serverModels.Connections, set json_dataModels.Set) (json_dataModels.Set, error) {
-	parts := strings.Split(set.Path, ":")
-
-	if len(parts) < 1 {
-		return json_dataModels.Set{}, fmt.Errorf("invalid path: '%s'", set.Path)
-	}
-
-	current := d
-	for i := range len(parts) - 1 {
-		next, ok := current.Datapoints[parts[i]]
-		if !ok {
-			return json_dataModels.Set{}, fmt.Errorf("path not found: '%s'", strings.Join(parts[:i+1], ":"))
-		}
-		current = next
-	}
-
-	toDelete := parts[len(parts)-1]
-	if dp, ok := current.Datapoints[toDelete]; ok {
-		dp.Publish(conns, OnDelete)
-		delete(current.Datapoints, toDelete)
-		return json_dataModels.Set{
-			Uuid: set.Uuid,
-			Path: set.Path,
-		}, nil
-	}
-	return json_dataModels.Set{}, fmt.Errorf("datapoint '%s' not found", set.Path)
-}
-
-func (d *Datapoint) GetAllDatapoints(depth uint) (dps []*Datapoint) {
-
-	var dfs func(dp *Datapoint, currentDepth uint)
-	dfs = func(dp *Datapoint, currentDepth uint) {
-		if depth == 1 {
-			return
-		} else if depth == 0 {
-			// Return all descendants
-			for _, child := range dp.Datapoints {
-				dps = append(dps, child)
-				dfs(child, currentDepth+1)
-			}
-			return
-		}
-
-		if currentDepth == depth-1 {
-			return
-		}
-
-		for _, child := range dp.Datapoints {
-			dps = append(dps, child)
-			dfs(child, currentDepth+1)
-		}
-	}
-	dps = append(dps, d)
-	dfs(d, 0)
-	return
-}
-
-func (d *Datapoint) QueryDatapoints(depth uint, path string) (dps []*Datapoint) {
-	parts := strings.Split(path, ":")
-
-	var dfs func(current *Datapoint, index int)
-	dfs = func(current *Datapoint, index int) {
-
-		if index == len(parts) {
-			dps = append(dps, current.GetAllDatapoints(depth)...)
-			return
-		}
-
-		pattern := "^" + parts[index] + "$"
-		re, err := regexp.Compile(pattern)
-		if err != nil {
-			return
-		}
-
-		for name, dp := range current.Datapoints {
-			if re.MatchString(name) {
-				dfs(dp, index+1)
-			}
-		}
-	}
-
-	dfs(d, 0)
-	return
-}
-
-func (d *Datapoint) AddSubscribtion(id string, sub json_dataModels.Subscribe) {
-	if d.Subscriptions == nil {
-		return
-	}
-
-	if s, ok := d.Subscriptions[id]; ok {
-		s.OnCreate = sub.OnCreate
-		s.OnChange = sub.OnChange
-		s.OnDelete = sub.OnDelete
-	} else {
-		d.Subscriptions[id] = &Subscription{
-			OnCreate: sub.OnCreate,
-			OnChange: sub.OnChange,
-			OnDelete: sub.OnDelete,
-		}
-	}
-}
-
-func (d *Datapoint) RemoveSubscribtion(id string) {
-	if _, ok := d.Subscriptions[id]; !ok {
-		return
-	}
-	delete(d.Subscriptions, id)
-}
-
-func (d *Datapoint) Publish(conns *serverModels.Connections, eventType string) error {
-	if conns.Clients == nil {
-		return nil
-	}
-	conns.RLock()
-	defer conns.RUnlock()
-
-	for id := range d.Subscriptions {
-		if client, ok := conns.Clients[id]; !ok {
-			delete(d.Subscriptions, id)
-		} else {
-			r := json_data.NewResponse()
-			r.AddUPublish(json_dataModels.Publish{
-				Event: eventType,
-				Uuid:  d.Uuid,
-				Path:  d.Path,
-				Value: d.Value,
-			})
-			err := wsjson.Write(client.Ctx, client.Conn, r)
-			if err != nil {
-				return err
-			}
-		}
-	}
-	return nil
 }

server/models/connections.go

@@ -1,10 +1,15 @@
 package models
 
 import (
+	"context"
+	"encoding/json"
+	"fmt"
 	"sync"
+	"time"
 
 	"github.com/coder/websocket"
 	"github.com/gin-gonic/gin"
+	json_dataModels "github.com/tecamino/tecamino-json_data/models"
 )
 
 type Connections struct {
@@ -30,3 +35,30 @@ func (c *Connections) RemoveClient(id string) {
 func (c *Connections) DisconnectWsConnection(id string, code websocket.StatusCode, reason string) {
 	c.Clients.DisconnectWsConnection(id, code, reason)
 }
+
+func (c *Connections) SendResponse(id string, r *json_dataModels.Response) error {
+	client, ok := c.Clients[id]
+	if !ok {
+		return fmt.Errorf("client not found for id " + id)
+
+	}
+	b, err := json.Marshal(r)
+	if err != nil {
+		return err
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	w, err := client.Conn.Writer(ctx, websocket.MessageText)
+	if err != nil {
+		return err
+	}
+	defer w.Close()
+
+	_, err = w.Write(b)
+	if err != nil {
+		return err
+	}
+	return nil
+}