paulbsd/ipbl
1
0
Fork 0
Code Issues Pull Requests Projects Releases 29 Wiki Activity

more stable ipbl websocket feat

Browse Source
This commit is contained in:
Paul 2023-04-09 15:05:31 +02:00
parent f5ed3ee455
commit 14d72693dd
3 changed files with 41 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
src/ws/init.go
View File

@ -11,10 +11,10 @@ import (
var WebSocketChannel = make(chan models.APIEvent, 100) var WebSocketChannel = make(chan models.APIEvent, 100)
var WebSocketChannelDone = make(chan bool) var WebSocketChannelDone = make(chan bool)
var LISTENERS map[string]ConnectionInfo var LISTENERS map[string]*ConnectionInfo
func Init(cfg *config.Config) { func Init(cfg *config.Config) {
LISTENERS = make(map[string]ConnectionInfo) LISTENERS = make(map[string]*ConnectionInfo)
} }
func WelcomeAgents(ws *websocket.Conn, welcome WSWelcome, t string) { func WelcomeAgents(ws *websocket.Conn, welcome WSWelcome, t string) {
@ -26,13 +26,13 @@ func WelcomeAgents(ws *websocket.Conn, welcome WSWelcome, t string) {
ConnectionPS: ws, ConnectionPS: ws,
InitDate: time.Now(), InitDate: time.Now(),
} }
LISTENERS[welcome.Hostname] = connectinfo LISTENERS[welcome.Hostname] = &connectinfo
} else if t == "rr" { } else if t == "rr" {
connectinfo := ConnectionInfo{ connectinfo := ConnectionInfo{
ConnectionRR: ws, ConnectionRR: ws,
InitDate: time.Now(), InitDate: time.Now(),
} }
LISTENERS[welcome.Hostname] = connectinfo LISTENERS[welcome.Hostname] = &connectinfo
} }
} else { } else {
if t == "ps" { if t == "ps" {

13
src/ws/pubsub.go
View File

@ -3,6 +3,7 @@ package ws
import ( import (
"encoding/json" "encoding/json"
"log" "log"
"time"
"git.paulbsd.com/paulbsd/ipbl/src/config" "git.paulbsd.com/paulbsd/ipbl/src/config"
"github.com/labstack/echo/v4" "github.com/labstack/echo/v4"
@ -19,7 +20,6 @@ func HandleWSPS(c echo.Context, cfg *config.Config) (err error) {
err := websocket.Message.Receive(ws, &msg) err := websocket.Message.Receive(ws, &msg)
if err != nil { if err != nil {
log.Println(err) log.Println(err)
return
} }
err = json.Unmarshal(msg, &welcome) err = json.Unmarshal(msg, &welcome)
@ -29,6 +29,15 @@ func HandleWSPS(c echo.Context, cfg *config.Config) (err error) {
WelcomeAgents(ws, welcome, "ps") WelcomeAgents(ws, welcome, "ps")
} }
err = websocket.Message.Receive(ws, "OK")
if err != nil {
log.Println(err)
}
for {
time.Sleep(50 * time.Millisecond)
}
/*for { /*for {
var apievent, ok = <-WebSocketChannel var apievent, ok = <-WebSocketChannel
if ok { if ok {
@ -39,7 +48,7 @@ func HandleWSPS(c echo.Context, cfg *config.Config) (err error) {
} }
err = websocket.Message.Send(ws, data) err = websocket.Message.Send(ws, data)
if err != nil { if err != nil {
gcConn(ws, "ps") //gcConn(ws, "ps")
log.Println(LISTENERS) log.Println(LISTENERS)
return return
} }

44
src/ws/reqrep.go
View File

@ -20,7 +20,6 @@ func HandleWSRR(c echo.Context, cfg *config.Config) error {
err := websocket.Message.Receive(ws, &msg) err := websocket.Message.Receive(ws, &msg)
if err != nil { if err != nil {
log.Println(err) log.Println(err)
return
} }
err = json.Unmarshal(msg, &welcome) err = json.Unmarshal(msg, &welcome)
@ -35,12 +34,10 @@ func HandleWSRR(c echo.Context, cfg *config.Config) error {
var msg []byte var msg []byte
err := websocket.Message.Receive(ws, &msg) err := websocket.Message.Receive(ws, &msg)
log.Println(string(msg))
if err != nil { if err != nil {
log.Println(err) log.Println(err)
return return
} }
websocket.Message.Send(ws, "OK")
err = json.Unmarshal(msg, &apievent) err = json.Unmarshal(msg, &apievent)
if err != nil { if err != nil {
@ -53,14 +50,14 @@ func HandleWSRR(c echo.Context, cfg *config.Config) error {
switch apievent.MsgType { switch apievent.MsgType {
case "bootstrap": case "bootstrap":
for _, value := range LISTENERS { for index, value := range LISTENERS {
err = websocket.Message.Send(value.ConnectionPS, msg) if index != apievent.Hostname {
if err != nil { err = websocket.Message.Send(value.ConnectionPS, msg)
log.Println(err) if err != nil {
log.Println(err)
}
} }
} }
log.Println(LISTENERS)
case "add": case "add":
session := cfg.Db.NewSession() session := cfg.Db.NewSession()
event.APIParse(session, apievent) event.APIParse(session, apievent)
@ -77,22 +74,33 @@ func HandleWSRR(c echo.Context, cfg *config.Config) error {
log.Println(err) log.Println(err)
} }
} }
log.Printf("websocket: Inserted event") log.Printf("ws: Inserted event")
case "file": case "init":
apievent.MsgType = "ws" for _, value := range LISTENERS {
err = websocket.Message.Send(value.ConnectionPS, msg)
if err != nil {
log.Println(err)
}
}
default: default:
log.Println("test")
} }
WebSocketChannel <- apievent
err = websocket.Message.Send(ws, "OK") err = websocket.Message.Send(ws, "OK")
if err != nil { if err != nil {
gcConn(ws, "rr") log.Println(err)
log.Println(LISTENERS)
return
} }
/*
WebSocketChannel <- apievent
err = websocket.Message.Send(ws, "OK")
if err != nil {
//gcConn(ws, "rr")
log.Println(LISTENERS)
return
}
*/
} }
}).ServeHTTP(c.Response(), c.Request()) }).ServeHTTP(c.Response(), c.Request())
return nil return nil