7 Commits

Author SHA1 Message Date
Adrian Zuercher
c37dd87a37 new respond to send all create change delete different bug fixes 2025-07-23 09:10:56 +02:00
Adrian Zuercher
8be5c80a22 add broadcast to all client and modify lingpong handler 2025-07-23 09:09:14 +02:00
Adrian Zuercher
5e1e4b9daf make new model for stringSlices allowOrigin helper 2025-07-13 20:01:13 +02:00
Adrian Zuercher
0f06128ce8 new send datapoint type and one fix typo 2025-07-11 17:47:12 +02:00
Adrian Zuercher
be07dc8749 add new argument for remote port 2025-07-11 17:46:32 +02:00
Adrian Zuercher
e75d7c8b03 add automatic localhost and local ip to allow origin 2025-06-29 21:18:15 +02:00
Adrian Zürcher
59c7705ca1 fix concurrent write new sendPong channel 2025-06-22 08:58:14 +02:00
15 changed files with 185 additions and 150 deletions

View File

@@ -2,7 +2,6 @@ package args
import ( import (
"flag" "flag"
"strings"
"github.com/tecamino/tecamino-dbm/cert" "github.com/tecamino/tecamino-dbm/cert"
"github.com/tecamino/tecamino-dbm/models" "github.com/tecamino/tecamino-dbm/models"
@@ -19,21 +18,10 @@ type Args struct {
Debug bool Debug bool
} }
type StringSlice []string
func (s *StringSlice) String() string {
return strings.Join(*s, ",")
}
func (s *StringSlice) Set(value string) error {
*s = append(*s, value)
return nil
}
// initialte cli arguments // initialte cli arguments
func Init() *Args { func Init() *Args {
var allowOrigins StringSlice var allowOrigins models.StringSlice
flag.Var(&allowOrigins, "allowOrigin", "Allowed origin (can repeat this flag)") flag.Var(&allowOrigins, "allowOrigin", "Allowed origin (can repeat this flag)")
@@ -43,6 +31,7 @@ func Init() *Args {
keyFile := flag.String("keyFile", "./cert/key.pem", "path of keyfile") keyFile := flag.String("keyFile", "./cert/key.pem", "path of keyfile")
portHttp := flag.Uint("http-port", 8100, "json server communication for http/ws") portHttp := flag.Uint("http-port", 8100, "json server communication for http/ws")
portHttps := flag.Uint("https-port", 8101, "json server communication for http/wss") portHttps := flag.Uint("https-port", 8101, "json server communication for http/wss")
remotePort := flag.Uint("remotePort", 9500, "remote Port of gui user interface")
rootDir := flag.String("workingDir", "./", "working directory") rootDir := flag.String("workingDir", "./", "working directory")
dbmFile := flag.String("dbm", "/test/test", "dbm file name") dbmFile := flag.String("dbm", "/test/test", "dbm file name")
debug := flag.Bool("debug", false, "debug flag") debug := flag.Bool("debug", false, "debug flag")
@@ -58,6 +47,7 @@ func Init() *Args {
Port: models.Port{ Port: models.Port{
Http: *portHttp, Http: *portHttp,
Https: *portHttps, Https: *portHttps,
Remote: *remotePort,
}, },
RootDir: *rootDir, RootDir: *rootDir,
DBMFile: *dbmFile, DBMFile: *dbmFile,
@@ -65,7 +55,7 @@ func Init() *Args {
} }
if len(allowOrigins) == 0 { if len(allowOrigins) == 0 {
allowOrigins = StringSlice{"http://localhost:9500"} allowOrigins = models.StringSlice{"http://localhost:9500"}
} }
a.AllowOrigins = allowOrigins a.AllowOrigins = allowOrigins

View File

@@ -35,6 +35,7 @@ func (d *DBMHandler) Json_Data(c *gin.Context) {
Path: res.Path, Path: res.Path,
Type: res.Type, Type: res.Type,
Value: res.Value, Value: res.Value,
Drivers: &res.Drivers,
HasChild: res.Datapoints != nil, HasChild: res.Datapoints != nil,
Rights: res.ReadWrite, Rights: res.ReadWrite,
}) })
@@ -66,9 +67,7 @@ func (d *DBMHandler) Delete(c *gin.Context) {
} }
response := json_data.NewResponse() response := json_data.NewResponse()
if payload.Set != nil { if payload.Set != nil {
response.Set, err = d.DBM.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()

View File

@@ -49,6 +49,7 @@ func (d *DBMHandler) Subscribe(req *json_dataModels.Request, id string) {
Path: dp.Path, Path: dp.Path,
Value: dp.Value, Value: dp.Value,
HasChild: dp.Datapoints != nil, HasChild: dp.Datapoints != nil,
Rights: dp.ReadWrite.GetRights(),
Driver: sub.Driver, Driver: sub.Driver,
Drivers: &dp.Drivers, Drivers: &dp.Drivers,
}) })

2
go.mod
View File

@@ -7,7 +7,7 @@ require (
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/gorilla/websocket v1.5.3 github.com/gorilla/websocket v1.5.3
github.com/tecamino/tecamino-json_data v0.0.16 github.com/tecamino/tecamino-json_data v0.0.19
github.com/tecamino/tecamino-logger v0.2.0 github.com/tecamino/tecamino-logger v0.2.0
) )

4
go.sum
View File

@@ -71,8 +71,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.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA= github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
github.com/tecamino/tecamino-json_data v0.0.16 h1:aZFxnhm4g6WMDPoqy4HosUk7vl0DB0iIcVs8bbT4MzU= github.com/tecamino/tecamino-json_data v0.0.19 h1:jaEiY38Rdur1ZKJ5/5H/7ZL5vn1NdUnM/0bQj2vByKc=
github.com/tecamino/tecamino-json_data v0.0.16/go.mod h1:LLlyD7Wwqplb2BP4PeO86EokEcTRidlW5MwgPd1T2JY= github.com/tecamino/tecamino-json_data v0.0.19/go.mod h1:LLlyD7Wwqplb2BP4PeO86EokEcTRidlW5MwgPd1T2JY=
github.com/tecamino/tecamino-logger v0.2.0 h1:NPH/Gg9qRhmVoW8b39i1eXu/LEftHc74nyISpcRG+XU= github.com/tecamino/tecamino-logger v0.2.0 h1:NPH/Gg9qRhmVoW8b39i1eXu/LEftHc74nyISpcRG+XU=
github.com/tecamino/tecamino-logger v0.2.0/go.mod h1:0M1E9Uei/qw3e3WA1x3lBo1eP3H5oeYE7GjYrMahnj8= github.com/tecamino/tecamino-logger v0.2.0/go.mod h1:0M1E9Uei/qw3e3WA1x3lBo1eP3H5oeYE7GjYrMahnj8=
github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS4MhqMhdFk5YI= github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS4MhqMhdFk5YI=

View File

@@ -23,7 +23,7 @@ func main() {
//initialize new server //initialize new server
dbmHandler.Log.Debug("main", "initialize new server instance") dbmHandler.Log.Debug("main", "initialize new server instance")
s := ws.NewServer(a.AllowOrigins) s := ws.NewServer(a.AllowOrigins, a.Port.Remote)
//set routes //set routes
dbmHandler.Log.Debug("main", "setting routes") dbmHandler.Log.Debug("main", "setting routes")

View File

@@ -55,8 +55,8 @@ func (d *Datapoint) Set(path string, set json_dataModels.Set) (bool, error) {
changed = true changed = true
d.Value = d.Type.ConvertValue(set.Value) d.Value = d.Type.ConvertValue(set.Value)
} }
} }
if set.Rights != "" { if set.Rights != "" {
changed = true changed = true
d.ReadWrite = set.Rights.GetRights() d.ReadWrite = set.Rights.GetRights()
@@ -108,7 +108,8 @@ func (d *Datapoint) CreateDatapoints(sets ...json_dataModels.Set) (created []jso
if i == len(parts)-1 { if i == len(parts)-1 {
// Leaf node: create or update datapoint // Leaf node: create or update datapoint
if existing, ok := current.Datapoints[part]; ok { if existing, ok := current.Datapoints[part]; ok {
publish, err := existing.Set("", dp)
_, err := existing.Set("", dp)
if err != nil { if err != nil {
return nil, nil, err return nil, nil, err
} }
@@ -122,18 +123,18 @@ func (d *Datapoint) CreateDatapoints(sets ...json_dataModels.Set) (created []jso
Updated: true, Updated: true,
}) })
if publish {
existing.Publish(OnChange) existing.Publish(OnChange)
}
} else { } else {
ndp := Datapoint{ ndp := Datapoint{
Uuid: uuid.New(), Uuid: uuid.New(),
CreateDateTime: time.Now().UnixMilli(), CreateDateTime: time.Now().UnixMilli(),
Subscriptions: InitSubscribtion(), Subscriptions: InitSubscribtion(),
} }
// Create new // Create new
current.Datapoints[part] = &ndp current.Datapoints[part] = &ndp
publish, err := ndp.Set(strings.Join(parts, ":"), dp) _, err := ndp.Set(strings.Join(parts, ":"), dp)
if err != nil { if err != nil {
return nil, nil, err return nil, nil, err
} }
@@ -145,10 +146,9 @@ func (d *Datapoint) CreateDatapoints(sets ...json_dataModels.Set) (created []jso
Rights: ndp.ReadWrite, Rights: ndp.ReadWrite,
Driver: dp.Driver, Driver: dp.Driver,
}) })
if publish { ndp.Publish(OnCreate)
current.Publish(OnChange)
} //add uuid to flat map for faster lookup
//add uuid to flat map for faster lookuo
uuids[ndp.Uuid] = &ndp uuids[ndp.Uuid] = &ndp
} }
} }
@@ -178,6 +178,8 @@ func (d *Datapoint) CreateDatapoints(sets ...json_dataModels.Set) (created []jso
newDp.ReadWrite = dp.Rights.GetRights() newDp.ReadWrite = dp.Rights.GetRights()
} }
newDp.Publish(OnCreate)
current.Datapoints[part] = newDp current.Datapoints[part] = newDp
current = newDp current = newDp
@@ -214,7 +216,7 @@ func (d *Datapoint) ImportDatapoint(dp *Datapoint, path string) (uuids Uuids, er
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 //add uuid to flat map for faster lookup
uuids[dp.Uuid] = dp uuids[dp.Uuid] = dp
dp.Publish(OnChange) dp.Publish(OnChange)
} }
@@ -270,47 +272,64 @@ func (d *Datapoint) UpdateValue(conns *ws.ClientHandler, value any) error {
return nil return nil
} }
func (d *Datapoint) RemoveDatapoint(conns *ws.ClientHandler, set json_dataModels.Set) (json_dataModels.Set, error) { func (d *Datapoint) RemoveDatapoint(conns *ws.ClientHandler, set json_dataModels.Set) (sets []json_dataModels.Set, err error) {
parts := strings.Split(set.Path, ":") parts := strings.Split(set.Path, ":")
if len(parts) < 1 { if len(parts) < 1 {
return json_dataModels.Set{}, fmt.Errorf("invalid path: '%s'", set.Path) return sets, fmt.Errorf("invalid path: '%s'", set.Path)
} }
current := d current := d
for i := 0; i < len(parts)-1; i++ { for i := 0; i < len(parts)-1; i++ {
next, ok := current.Datapoints[parts[i]] next, ok := current.Datapoints[parts[i]]
if !ok { if !ok {
return json_dataModels.Set{}, fmt.Errorf("path not found: '%s'", strings.Join(parts[:i+1], ":")) return sets, fmt.Errorf("path not found: '%s'", strings.Join(parts[:i+1], ":"))
} }
current = next current = next
} }
toDelete := parts[len(parts)-1] toDelete := parts[len(parts)-1]
if dp, ok := current.Datapoints[toDelete]; ok { if dp, ok := current.Datapoints[toDelete]; ok {
sets = append(sets, removeChildren(dp)...)
dp.Publish(OnDelete) dp.Publish(OnDelete)
sets = append(sets, json_dataModels.Set{
Uuid: dp.Uuid,
Path: dp.Path,
})
delete(current.Datapoints, toDelete) delete(current.Datapoints, toDelete)
return json_dataModels.Set{ return sets, nil
Uuid: set.Uuid,
Path: set.Path,
}, nil
} }
return json_dataModels.Set{}, fmt.Errorf("datapoint '%s' not found", set.Path) return sets, fmt.Errorf("datapoint '%s' not found", set.Path)
}
// removes all children and grandchlidren of datapoint
func removeChildren(dp *Datapoint) (sets []json_dataModels.Set) {
for name, d := range dp.Datapoints {
sets = append(sets, removeChildren(d)...)
d.Publish(OnDelete)
sets = append(sets, json_dataModels.Set{
Uuid: d.Uuid,
Path: d.Path,
})
delete(d.Datapoints, name)
}
return sets
} }
func (d *Datapoint) GetAllDatapoints(depth uint) (dps Datapoints) { func (d *Datapoint) GetAllDatapoints(depth uint) (dps Datapoints) {
var dfs func(dp *Datapoint, currentDepth uint) var dfs func(dp *Datapoint, currentDepth uint)
dfs = func(dp *Datapoint, currentDepth uint) { dfs = func(dp *Datapoint, currentDepth uint) {
if depth == 1 { switch depth {
return case 0:
} else if depth == 0 {
// Return all descendants // Return all descendants
for _, child := range dp.Datapoints { for _, child := range dp.Datapoints {
dps = append(dps, child) dps = append(dps, child)
dfs(child, currentDepth+1) dfs(child, currentDepth+1)
} }
return return
case 1:
return
} }
if currentDepth == depth-1 { if currentDepth == depth-1 {
@@ -379,12 +398,12 @@ func (d *Datapoint) RemoveSubscribtion(client *wsModels.Client) {
} }
func (d *Datapoint) Publish(eventType string) error { func (d *Datapoint) Publish(eventType string) error {
for client := range d.Subscriptions {
r := json_data.NewResponse() r := json_data.NewResponse()
r.AddPublish(json_dataModels.Publish{ r.AddPublish(json_dataModels.Publish{
Event: eventType, Event: eventType,
Uuid: d.Uuid, Uuid: d.Uuid,
Path: d.Path, Path: d.Path,
Type: d.Type,
Value: d.Value, Value: d.Value,
}) })
@@ -392,8 +411,13 @@ func (d *Datapoint) Publish(eventType string) error {
if err != nil { if err != nil {
return err return err
} }
switch eventType {
case OnCreate, OnDelete:
ws.SendBroadcast(b)
default:
for client := range d.Subscriptions {
client.SendResponse(b) client.SendResponse(b)
} }
}
return nil return nil
} }

View File

@@ -87,18 +87,20 @@ func (d *DBM) UpdateDatapointValue(value any, uid uuid.UUID, path ...string) err
return d.Datapoints.UpdateDatapointValue(value, path[0]) return d.Datapoints.UpdateDatapointValue(value, path[0])
} }
func (d *DBM) RemoveDatapoint(sets ...json_dataModels.Set) ([]json_dataModels.Set, error) { func (d *DBM) RemoveDatapoint(sets ...json_dataModels.Set) (lsRemoved []json_dataModels.Set, err error) {
var lsRemoved []json_dataModels.Set
for _, set := range sets { for _, set := range sets {
removed, err := d.Datapoints.RemoveDatapoint(d.Conns, set) if set.Path == "" {
if dp, ok := d.Uuids[set.Uuid]; ok {
set.Path = dp.Path
}
}
lsRemoved, err = d.Datapoints.RemoveDatapoint(d.Conns, set)
if err != nil { if err != nil {
return lsRemoved, err return
} }
lsRemoved = append(lsRemoved, removed) d.ModifyCountedDatapoints(uint64(len(lsRemoved)), true)
d.ModifyCountedDatapoints(1, true)
} }
return
return lsRemoved, nil
} }
func (d *DBM) QueryDatapoints(depth uint, uid uuid.UUID, key ...string) []*Datapoint { func (d *DBM) QueryDatapoints(depth uint, uid uuid.UUID, key ...string) []*Datapoint {
@@ -159,8 +161,6 @@ func (d *DBM) GoSystemTime() error {
func (d *DBM) GoSystemMemory() error { func (d *DBM) GoSystemMemory() error {
path := "System:UsedMemory" path := "System:UsedMemory"
var m runtime.MemStats
var mOld uint64
typ := json_dataModels.STR typ := json_dataModels.STR
rights := json_dataModels.Read rights := json_dataModels.Read
@@ -172,14 +172,12 @@ func (d *DBM) GoSystemMemory() error {
go func() { go func() {
for { for {
var m runtime.MemStats
runtime.ReadMemStats(&m) runtime.ReadMemStats(&m)
if m.Sys != mOld { mem := fmt.Sprintf("%.2f MB", float64(m.Alloc)/1024/1024)
mem := fmt.Sprintf("%.2f MB", float64(m.Sys)/1024/1024)
if er := d.UpdateDatapointValue(mem, uuid.Nil, path); er != nil { if er := d.UpdateDatapointValue(mem, uuid.Nil, path); er != nil {
d.Log.Error("dmb.Handler.AddSystemDps.UpdateDatapointValue", er.Error()) d.Log.Error("dmb.Handler.AddSystemDps.UpdateDatapointValue", er.Error())
} }
mOld = m.Sys
}
time.Sleep(time.Second) time.Sleep(time.Second)
} }
}() }()

View File

@@ -3,4 +3,5 @@ package models
type Port struct { type Port struct {
Http uint Http uint
Https uint Https uint
Remote uint
} }

14
models/stringSlices.go Normal file
View File

@@ -0,0 +1,14 @@
package models
import "strings"
type StringSlice []string
func (s *StringSlice) String() string {
return strings.Join(*s, ",")
}
func (s *StringSlice) Set(value string) error {
*s = append(*s, value)
return nil
}

View File

@@ -142,7 +142,7 @@ func TestUpdateDps(t *testing.T) {
func TestServer(t *testing.T) { func TestServer(t *testing.T) {
fmt.Println("start") fmt.Println("start")
server := ws.NewServer([]string{".*"}) server := ws.NewServer([]string{".*"}, 9500)
t.Fatal(server.ServeHttp("http://localhost", 8100)) t.Fatal(server.ServeHttp("http://localhost", 8100))
} }

View File

@@ -1,5 +1,10 @@
package utils package utils
import (
"fmt"
"net"
)
func ListofA2ZZ() (list []string) { func ListofA2ZZ() (list []string) {
for i := 'A'; i <= 'Z'; i++ { for i := 'A'; i <= 'Z'; i++ {
list = append(list, string(i)) list = append(list, string(i))
@@ -9,12 +14,21 @@ func ListofA2ZZ() (list []string) {
list = append(list, string(i)+string(j)) list = append(list, string(i)+string(j))
} }
} }
// for i := 'A'; i <= 'Z'; i++ {
// for j := 'A'; j <= 'Z'; j++ {
// for k := 'A'; k <= 'Z'; k++ {
// list = append(list, string(i)+string(j)+string(k))
// }
// }
// }
return return
} }
func GetLocalIP() (string, error) {
addrs, err := net.InterfaceAddrs()
if err != nil {
return "", err
}
for _, addr := range addrs {
if ipNet, ok := addr.(*net.IPNet); ok && !ipNet.IP.IsLoopback() {
if ipNet.IP.To4() != nil {
return ipNet.IP.String(), nil
}
}
}
return "", fmt.Errorf("no local IP address found")
}

View File

@@ -17,6 +17,12 @@ type ClientHandler struct {
Clients models.Clients Clients models.Clients
} }
func SendBroadcast(msg []byte) {
for _, c := range models.Broadcast {
c.SendResponse(msg)
}
}
// initaiates new conections with client map // initaiates new conections with client map
func NewConnectionHandler() *ClientHandler { func NewConnectionHandler() *ClientHandler {
return &ClientHandler{ return &ClientHandler{
@@ -41,7 +47,6 @@ func (cH *ClientHandler) ConnectNewClient(id string, c *gin.Context) (client *mo
cH.Lock() cH.Lock()
cH.Clients[id] = client cH.Clients[id] = client
cH.Unlock() cH.Unlock()
return client, nil return client, nil
} }

View File

@@ -1,10 +1,10 @@
package models package models
import ( import (
"encoding/json"
"fmt" "fmt"
"log" "log"
"net/http" "net/http"
"slices"
"time" "time"
"github.com/gin-gonic/gin" "github.com/gin-gonic/gin"
@@ -13,6 +13,8 @@ import (
var Origins []string = []string{"*"} var Origins []string = []string{"*"}
var Broadcast Clients = make(Clients)
const ( const (
// Time allowed to write a message to the peer. // Time allowed to write a message to the peer.
writeWait = 10 * time.Second writeWait = 10 * time.Second
@@ -27,7 +29,7 @@ const (
type Client struct { type Client struct {
Id string Id string
Connected bool `json:"connected"` Connected bool `json:"connected"`
Conn *websocket.Conn `json:"-"` conn *websocket.Conn `json:"-"`
OnOpen func() OnOpen func()
OnMessage func(data []byte) OnMessage func(data []byte)
OnClose func(code int, reason string) OnClose func(code int, reason string)
@@ -42,17 +44,10 @@ var upgrader = websocket.Upgrader{
CheckOrigin: func(r *http.Request) bool { CheckOrigin: func(r *http.Request) bool {
if len(Origins) == 0 { if len(Origins) == 0 {
return false return false
} } else if Origins[0] == "*" {
if Origins[0] == "*" {
return true return true
} }
origin := r.Header.Get("Origin") return slices.Contains(Origins, r.Header.Get("Origin"))
for _, o := range Origins {
if o == origin {
return true
}
}
return false
}, },
EnableCompression: false, EnableCompression: false,
} }
@@ -66,21 +61,26 @@ func ConnectNewClient(id string, c *gin.Context) (*Client, error) {
client := &Client{ client := &Client{
Id: id, Id: id,
Connected: true, Connected: true,
Conn: conn, conn: conn,
send: make(chan []byte), send: make(chan []byte, 512),
unregister: make(chan []byte), unregister: make(chan []byte, 256),
} }
Broadcast[client.Id] = client
conn.SetPingHandler(func(appData string) error { conn.SetPingHandler(func(appData string) error {
if client.OnPing != nil { if client.OnPing != nil {
client.OnPing() client.OnPing()
} }
conn.SetWriteDeadline(time.Now().Add(writeWait)) conn.SetWriteDeadline(time.Now().Add(writeWait))
conn.SetReadDeadline(time.Now().Add(writeWait)) conn.SetReadDeadline(time.Now().Add(writeWait))
return conn.WriteMessage(websocket.PongMessage, []byte(appData)) if err := client.conn.WriteControl(websocket.PongMessage, []byte(appData), time.Now().Add(pongWait)); err != nil {
client.OnError(err)
}
return nil
}) })
conn.SetPongHandler(func(string) error { conn.SetPongHandler(func(appData string) error {
conn.SetReadDeadline(time.Now().Add(pongWait)) conn.SetReadDeadline(time.Now().Add(pongWait))
if client.OnPong != nil { if client.OnPong != nil {
client.OnPong() client.OnPong()
@@ -91,6 +91,7 @@ func ConnectNewClient(id string, c *gin.Context) (*Client, error) {
// Start reading messages from client // Start reading messages from client
go client.Read() go client.Read()
go client.Write() go client.Write()
go client.PingInterval(pingPeriod)
return client, nil return client, nil
} }
@@ -99,9 +100,10 @@ func (c *Client) Read() {
c.OnOpen() c.OnOpen()
} }
c.Conn.SetReadDeadline(time.Now().Add(writeWait)) c.conn.SetReadDeadline(time.Now().Add(writeWait))
for c.Connected { for c.Connected {
msgType, msg, err := c.Conn.ReadMessage() msgType, msg, err := c.conn.ReadMessage()
if err != nil { if err != nil {
c.handleError(fmt.Errorf("read error (id:%s): %w", c.Id, err)) c.handleError(fmt.Errorf("read error (id:%s): %w", c.Id, err))
return return
@@ -111,13 +113,11 @@ func (c *Client) Read() {
c.Close(websocket.CloseNormalClosure, "Client closed") c.Close(websocket.CloseNormalClosure, "Client closed")
return return
case websocket.TextMessage: case websocket.TextMessage:
if isPing := c.handleJsonPing(msg); !isPing {
if c.OnMessage != nil { if c.OnMessage != nil {
c.OnMessage(msg) c.OnMessage(msg)
} else { } else {
log.Printf("Received message but no handler set (id:%s): %s", c.Id, string(msg)) log.Printf("Received message but no handler set (id:%s): %s", c.Id, string(msg))
} }
}
default: default:
log.Printf("Unhandled message type %d (id:%s)", msgType, c.Id) log.Printf("Unhandled message type %d (id:%s)", msgType, c.Id)
} }
@@ -125,44 +125,31 @@ func (c *Client) Read() {
} }
func (c *Client) Write() { func (c *Client) Write() {
ticker := time.NewTicker(pingPeriod) defer c.conn.Close()
defer func() {
ticker.Stop()
c.Conn.Close()
}()
for {
for {
select { select {
case message, ok := <-c.send: case message, ok := <-c.send:
c.Conn.SetWriteDeadline(time.Now().Add(writeWait)) c.conn.SetWriteDeadline(time.Now().Add(writeWait))
if !ok { if !ok {
// The hub closed the channel. // The hub closed the channel.
if err := c.Conn.WriteMessage(websocket.CloseMessage, []byte{}); err != nil { if err := c.conn.WriteMessage(websocket.CloseMessage, []byte("ping")); err != nil {
c.handleError(err) c.handleError(err)
return return
} }
c.handleError(fmt.Errorf("server %s closed channel", c.Id)) c.handleError(fmt.Errorf("server %s closed channel", c.Id))
return return
} else { } else {
if err := c.Conn.WriteMessage(websocket.TextMessage, message); err != nil { if err := c.conn.WriteMessage(websocket.TextMessage, message); err != nil {
c.handleError(err) c.handleError(err)
return return
} }
} }
case <-ticker.C:
c.Conn.SetWriteDeadline(time.Now().Add(writeWait))
if err := c.Conn.WriteMessage(websocket.PingMessage, nil); err != nil {
c.handleError(err)
return
}
if c.OnPing != nil {
c.OnPing()
}
case message := <-c.unregister: case message := <-c.unregister:
c.Conn.WriteMessage(websocket.CloseMessage, message) c.conn.WriteMessage(websocket.CloseMessage, message)
c.Connected = false c.Connected = false
close(c.send) close(c.send)
delete(Broadcast, c.Id)
close(c.unregister) close(c.unregister)
return return
} }
@@ -170,31 +157,20 @@ func (c *Client) Write() {
} }
} }
func (c *Client) handleJsonPing(msg []byte) (isPing bool) { func (c *Client) PingInterval(interval time.Duration) {
var wsMsg WSMessage ticker := time.NewTicker(interval)
err := json.Unmarshal(msg, &wsMsg) defer ticker.Stop()
if err == nil && wsMsg.IsPing() { for range ticker.C {
c.Conn.SetReadDeadline(time.Now().Add(writeWait))
// Respond with pong JSON
select {
case c.send <- GetPongByteSlice():
default:
// optional: log or handle if send buffer is full
c.handleError(fmt.Errorf("failed to queue pong message"))
return
}
if err != nil {
c.handleError(fmt.Errorf("write pong error: %w", err))
return
}
if c.OnPing != nil { if c.OnPing != nil {
c.OnPing() c.OnPing()
} }
isPing = true
}
if err := c.conn.WriteControl(websocket.PingMessage, []byte("ping"), time.Now().Add(pongWait)); err != nil {
c.OnError(err)
return return
}
}
} }
func (c *Client) SendResponse(data []byte) { func (c *Client) SendResponse(data []byte) {
@@ -206,6 +182,7 @@ func (c *Client) SendResponse(data []byte) {
func (c *Client) Close(code int, reason string) error { func (c *Client) Close(code int, reason string) error {
closeMsg := websocket.FormatCloseMessage(code, reason) closeMsg := websocket.FormatCloseMessage(code, reason)
select { select {
case c.unregister <- closeMsg: // Attempt to send case c.unregister <- closeMsg: // Attempt to send
default: // If the channel is full, this runs default: // If the channel is full, this runs
@@ -221,6 +198,7 @@ func (c *Client) handleError(err error) {
if c.OnError != nil { if c.OnError != nil {
c.OnError(err) c.OnError(err)
} }
if err := c.Close(websocket.CloseInternalServerErr, err.Error()); err != nil { if err := c.Close(websocket.CloseInternalServerErr, err.Error()); err != nil {
if c.OnError != nil { if c.OnError != nil {
c.OnError(err) c.OnError(err)

View File

@@ -2,12 +2,14 @@ package websocket
import ( import (
"fmt" "fmt"
"log"
"sync" "sync"
"time" "time"
"github.com/gin-contrib/cors" "github.com/gin-contrib/cors"
"github.com/gin-gonic/gin" "github.com/gin-gonic/gin"
"github.com/tecamino/tecamino-dbm/cert" "github.com/tecamino/tecamino-dbm/cert"
"github.com/tecamino/tecamino-dbm/utils"
"github.com/tecamino/tecamino-logger/logging" "github.com/tecamino/tecamino-logger/logging"
) )
@@ -19,8 +21,17 @@ type Server struct {
} }
// initalizes new dbm server // initalizes new dbm server
func NewServer(allowOrigins []string) *Server { func NewServer(allowOrigins []string, port uint) *Server {
r := gin.Default() r := gin.Default()
allowOrigins = append(allowOrigins, fmt.Sprintf("http://localhost:%d", port))
localIP, err := utils.GetLocalIP()
if err != nil {
log.Printf("get local ip : %s", err.Error())
} else {
allowOrigins = append(allowOrigins, fmt.Sprintf("http://%s:%d", localIP, port))
}
r.Use(cors.New(cors.Config{ r.Use(cors.New(cors.Config{
AllowOrigins: allowOrigins, AllowOrigins: allowOrigins,
AllowMethods: []string{"GET", "POST", "PUT", "DELETE", "OPTIONS"}, AllowMethods: []string{"GET", "POST", "PUT", "DELETE", "OPTIONS"},