hoangvv/NextZen-UserService
1
0
Fork 0
mirror of https://github.com/KaySar12/NextZen-UserService.git synced 2025-10-06 19:59:42 +07:00
Code Issues Packages Projects Releases Wiki Activity

add message_bus service dependency (#25)

Browse Source
This commit is contained in:
link
2022-12-01 17:17:44 +08:00
committed by GitHub
parent f6ad3aa89a
commit 98a35cb107
5 changed files with 102 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
route/event_listen.go
+41 -36
View File
@@ -5,6 +5,7 @@ import (
"fmt"
"log"
"strings"
"time"
"github.com/IceWhaleTech/CasaOS-Common/external"
"github.com/IceWhaleTech/CasaOS-Common/utils/logger"
@@ -17,49 +18,53 @@ import (
)
func EventListen() {
for i := 0; i < 100; i++ {
messageBusUrl, err := external.GetMessageBusAddress(config.CommonInfo.RuntimePath)
if err != nil {
logger.Error("get message bus url error", zap.Any("err", err))
return
}
wsURL := fmt.Sprintf("ws://%s/event/%s", strings.ReplaceAll(messageBusUrl, "http://", ""), "local-storage")
ws, err := websocket.Dial(wsURL, "", "http://localhost")
if err != nil {
logger.Error("connect websocket err", zap.Any("error", err))
}
defer ws.Close()
log.Println("subscribed to", wsURL)
for {
msg := make([]byte, 1024)
n, err := ws.Read(msg)
messageBusUrl, err := external.GetMessageBusAddress(config.CommonInfo.RuntimePath)
if err != nil {
log.Fatalln(err.Error())
logger.Error("get message bus url error", zap.Any("err", err))
return
}
var event message_bus.Event
if err := json.Unmarshal(msg[:n], &event); err != nil {
log.Println(err.Error())
}
propertiesStr, err := json.Marshal(event.Properties)
wsURL := fmt.Sprintf("ws://%s/event/%s", strings.ReplaceAll(messageBusUrl, "http://", ""), "local-storage")
ws, err := websocket.Dial(wsURL, "", "http://localhost")
if err != nil {
logger.Error("connect websocket err", zap.Any("error", err))
time.Sleep(time.Second * 5)
continue
}
model := model.EventModel{
SourceID: event.SourceID,
Name: event.Name,
Properties: string(propertiesStr),
UUID: *event.Uuid,
defer ws.Close()
log.Println("subscribed to", wsURL)
for {
msg := make([]byte, 1024)
n, err := ws.Read(msg)
if err != nil {
log.Fatalln(err.Error())
}
var event message_bus.Event
if err := json.Unmarshal(msg[:n], &event); err != nil {
log.Println(err.Error())
}
propertiesStr, err := json.Marshal(event.Properties)
if err != nil {
continue
}
model := model.EventModel{
SourceID: event.SourceID,
Name: event.Name,
Properties: string(propertiesStr),
UUID: *event.Uuid,
}
service.MyService.Event().CreateEvemt(model)
output, err := json.MarshalIndent(event, "", " ")
if err != nil {
log.Println(err.Error())
}
log.Println(string(output))
}
service.MyService.Event().CreateEvemt(model)
output, err := json.MarshalIndent(event, "", " ")
if err != nil {
log.Println(err.Error())
}
log.Println(string(output))
}
}
route/v2/api_route_event.go
+4
View File
@@ -18,3 +18,7 @@ func (s *UserService) GetEvents(ctx echo.Context, params codegen.GetEventsParams
list := service.MyService.Event().GetEvents()
return ctx.JSON(http.StatusOK, list)
}
func (s *UserService) DeleteEventBySerial(ctx echo.Context, serial codegen.Serial) error {
service.MyService.Event().DeleteEventBySerial(serial)
return ctx.JSON(http.StatusOK, serial)
}