Send notification changes to message bus (#861)

cmd/migration-tool/main.go

@@ -17,10 +17,10 @@ import (
 
 	interfaces "github.com/IceWhaleTech/CasaOS-Common"
 	"github.com/IceWhaleTech/CasaOS-Common/utils/systemctl"
+	"github.com/IceWhaleTech/CasaOS/common"
 	"github.com/IceWhaleTech/CasaOS/pkg/config"
 	"github.com/IceWhaleTech/CasaOS/pkg/sqlite"
 	"github.com/IceWhaleTech/CasaOS/service"
-	"github.com/IceWhaleTech/CasaOS/types"
 	"gorm.io/gorm"
 )
 
@@ -47,7 +47,7 @@ func init() {
 	flag.Parse()
 
 	if *versionFlag {
-		fmt.Println("v" + types.CURRENTVERSION)
+		fmt.Println("v" + common.VERSION)
 		os.Exit(0)
 	}
 
@@ -86,7 +86,7 @@ func init() {
 	sqliteDB = sqlite.GetDb(dbFlag)
 	// gredis.GetRedisConn(config.RedisInfo),
 
-	service.MyService = service.NewService(sqliteDB, "", nil)
+	service.MyService = service.NewService(sqliteDB, "")
 }
 
 func main() {

common/constants.go

@@ -0,0 +1,7 @@
+package common
+
+const (
+	SERVICENAME = "casaos"
+	VERSION     = "0.4.2"
+	BODY        = " "
+)

common/message.go

@@ -0,0 +1,23 @@
+package common
+
+import (
+	"fmt"
+
+	"github.com/IceWhaleTech/CasaOS/codegen/message_bus"
+)
+
+var (
+	// devtype -> action -> event
+	EventTypes map[string]map[string]message_bus.EventType
+
+	PropertyNameLookupMaps = map[string]map[string]string{
+		"system": {
+			fmt.Sprintf("%s:%s", SERVICENAME, "utilization"): "ID_BUS",
+		},
+	}
+
+	ActionPastTense = map[string]string{
+		"add":    "added",
+		"remove": "removed",
+	}
+)

go.mod

@@ -31,6 +31,7 @@ require (
 	github.com/patrickmn/go-cache v2.1.0+incompatible
 	github.com/pkg/errors v0.9.1
 	github.com/robfig/cron v1.2.0
+	github.com/samber/lo v1.37.0
 	github.com/satori/go.uuid v1.2.0
 	github.com/shirou/gopsutil/v3 v3.22.11
 	github.com/sirupsen/logrus v1.9.0
@@ -45,6 +46,7 @@ require (
 
 require (
 	github.com/andybalholm/brotli v1.0.1 // indirect
+	github.com/apapsch/go-jsonmerge/v2 v2.0.0 // indirect
 	github.com/benbjohnson/clock v1.3.0 // indirect
 	github.com/coreos/go-systemd/v22 v22.3.2 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
@@ -107,6 +109,7 @@ require (
 	github.com/yusufpapurcu/wmi v1.2.2 // indirect
 	go.uber.org/atomic v1.7.0 // indirect
 	go.uber.org/multierr v1.6.0 // indirect
+	golang.org/x/exp v0.0.0-20220303212507-bbda1eaf7a17 // indirect
 	golang.org/x/image v0.0.0-20191009234506-e7c1f5e7dbb8 // indirect
 	golang.org/x/net v0.5.0 // indirect
 	golang.org/x/sys v0.4.0 // indirect

go.sum

@@ -3,12 +3,16 @@ github.com/Curtis-Milo/nat-type-identifier-go v0.0.0-20220215191915-18d42168c63d
 github.com/Curtis-Milo/nat-type-identifier-go v0.0.0-20220215191915-18d42168c63d/go.mod h1:lW9x+yEjqKdPbE3+cf2fGPJXCw/hChX3Omi9QHTLFsQ=
 github.com/IceWhaleTech/CasaOS-Common v0.4.2-alpha1 h1:4Z61swpEC/OUGLlzww1v/D916j7yqYnfBevdVD/dgjI=
 github.com/IceWhaleTech/CasaOS-Common v0.4.2-alpha1/go.mod h1:xcemiRsXcs1zrmQxYMyExDjZ7UHYwkJqYE71IDIV0xA=
+github.com/RaveNoX/go-jsoncommentstrip v1.0.0/go.mod h1:78ihd09MekBnJnxpICcwzCMzGrKSKYe4AqU6PDYYpjk=
 github.com/Xhofe/go-cache v0.0.0-20220723083548-714439c8af9a h1:RenIAa2q4H8UcS/cqmwdT1WCWIAH5aumP8m8RpbqVsE=
 github.com/Xhofe/go-cache v0.0.0-20220723083548-714439c8af9a/go.mod h1:sSBbaOg90XwWKtpT56kVujF0bIeVITnPlssLclogS04=
 github.com/andybalholm/brotli v1.0.1 h1:KqhlKozYbRtJvsPrrEeXcO+N2l6NYT5A2QAFmSULpEc=
 github.com/andybalholm/brotli v1.0.1/go.mod h1:loMXtMfwqflxFJPmdbJO0a3KNoPuLBgiu3qAvBg8x/Y=
+github.com/apapsch/go-jsonmerge/v2 v2.0.0 h1:axGnT1gRIfimI7gJifB699GoE/oq+F2MU7Dml6nw9rQ=
+github.com/apapsch/go-jsonmerge/v2 v2.0.0/go.mod h1:lvDnEdqiQrp0O42VQGgmlKpxL1AP2+08jFMw88y4klk=
 github.com/benbjohnson/clock v1.3.0 h1:ip6w0uFQkncKQ979AypyG0ER7mqUSBdKLOgAle/AT8A=
 github.com/benbjohnson/clock v1.3.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
+github.com/bmatcuk/doublestar v1.1.1/go.mod h1:UD6OnuiIn0yFxxA2le/rnRU1G4RaI4UvFv1sNto9p6w=
 github.com/chzyer/logex v1.2.0/go.mod h1:9+9sk7u7pGNWYMkh0hdiL++6OeibzJccyQU4p4MedaY=
 github.com/chzyer/readline v1.5.0/go.mod h1:x22KAscuvRqlLoK9CsoYsmxoXZMMFVyOl86cAH8qUic=
 github.com/chzyer/test v0.0.0-20210722231415-061457976a23/go.mod h1:Q3SI9o4m/ZMnBNeIyt5eFwwo7qiLfzFZmjNmxjkiQlU=
@@ -153,6 +157,7 @@ github.com/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8Hm
 github.com/josharian/intern v1.0.0/go.mod h1:5DoeVV0s6jJacbCEi61lwdGj/aVlrQvzHFFd8Hwg//Y=
 github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM=
 github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
+github.com/juju/gnuflag v0.0.0-20171113085948-2ce1bb71843d/go.mod h1:2PavIy+JPciBPrBUjwbNvtwB6RQlve+hkpll6QSNmOE=
 github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51/go.mod h1:CzGEWj7cYgsdH8dAjBGEr58BoE7ScuLd+fwFZ44+/x8=
 github.com/klauspost/compress v1.4.1/go.mod h1:RyIbtBH6LamlWaDj8nUwkbUhJ87Yi3uG0guNDohfE1A=
 github.com/klauspost/compress v1.11.4/go.mod h1:aoV0uJVorq1K+umq18yTdKaF57EivdYsUV+/s2qKfXs=
@@ -227,12 +232,15 @@ github.com/robfig/cron v1.2.0/go.mod h1:JGuDeoQd7Z6yL4zQhZ3OPEVHB7fL6Ka6skscFHfm
 github.com/rogpeppe/go-internal v1.6.1/go.mod h1:xXDCJY+GAPziupqXw64V24skbSoqbTEfhy4qGm1nDQc=
 github.com/rogpeppe/go-internal v1.8.0 h1:FCbCCtXNOY3UtUuHUYaghJg4y7Fd14rXifAYUAtL9R8=
 github.com/rogpeppe/go-internal v1.8.0/go.mod h1:WmiCO8CzOY8rg0OYDC4/i/2WRWAB6poM+XZ2dLUbcbE=
+github.com/samber/lo v1.37.0 h1:XjVcB8g6tgUp8rsPsJ2CvhClfImrpL04YpQHXeHPhRw=
+github.com/samber/lo v1.37.0/go.mod h1:9vaz2O4o8oOnK23pd2TrXufcbdbJIa3b6cstBWKpopA=
 github.com/satori/go.uuid v1.2.0 h1:0uYX9dsZ2yD7q2RtLRtPSdGDWzjeM3TbMJP9utgA0ww=
 github.com/satori/go.uuid v1.2.0/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdhQKdks0=
 github.com/shirou/gopsutil/v3 v3.22.11 h1:kxsPKS+Eeo+VnEQ2XCaGJepeP6KY53QoRTETx3+1ndM=
 github.com/shirou/gopsutil/v3 v3.22.11/go.mod h1:xl0EeL4vXJ+hQMAGN8B9VFpxukEMA0XdevQOe5MZ1oY=
 github.com/sirupsen/logrus v1.9.0 h1:trlNQbNUG3OdDrDil03MCb1H2o9nJ1x4/5LYw7byDE0=
 github.com/sirupsen/logrus v1.9.0/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
+github.com/spkg/bom v0.0.0-20160624110644-59b7046e48ad/go.mod h1:qLr4V1qq6nMqFKkMo8ZTx3f+BZEkzsRUY10Xsm2mwU0=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
 github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
@@ -286,6 +294,8 @@ golang.org/x/crypto v0.0.0-20210711020723-a769d52b0f97/go.mod h1:GvvjBRRGRdwPK5y
 golang.org/x/crypto v0.0.0-20211215153901-e495a2d5b3d3/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
 golang.org/x/crypto v0.5.0 h1:U/0M97KRkSFvyD/3FSmdP5W5swImpNgle/EHFhOsQPE=
 golang.org/x/crypto v0.5.0/go.mod h1:NK/OQwhpMQP3MwtdjgLlYHnH9ebylxKWv3e0fK+mkQU=
+golang.org/x/exp v0.0.0-20220303212507-bbda1eaf7a17 h1:3MTrJm4PyNL9NBqvYDSj3DHl46qQakyfqfWo4jgfaEM=
+golang.org/x/exp v0.0.0-20220303212507-bbda1eaf7a17/go.mod h1:lgLbSvA5ygNOMpwM/9anMpWVlVJ7Z+cHWq/eFuinpGE=
 golang.org/x/image v0.0.0-20191009234506-e7c1f5e7dbb8 h1:hVwzHzIUGRjiF7EcUjqNxk3NCfkPxbDKRdnNE1Rpg0U=
 golang.org/x/image v0.0.0-20191009234506-e7c1f5e7dbb8/go.mod h1:FeLwcggjj3mMvU+oOTbSwawSJRM1uh48EjtB4UJZlP0=
 golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=

main.go

@@ -1,8 +1,9 @@
 //go:generate bash -c "mkdir -p codegen && go run github.com/deepmap/oapi-codegen/cmd/oapi-codegen@v1.12.4 -generate types,server,spec -package codegen api/casaos/openapi.yaml > codegen/casaos_api.go"
-
+//go:generate bash -c "mkdir -p codegen/message_bus && go run github.com/deepmap/oapi-codegen/cmd/oapi-codegen@v1.12.4 -generate types,client -package message_bus https://raw.githubusercontent.com/IceWhaleTech/CasaOS-MessageBus/main/api/message_bus/openapi.yaml > codegen/message_bus/api.go"
 package main
 
 import (
+	"context"
 	_ "embed"
 	"flag"
 	"fmt"
@@ -14,10 +15,11 @@ import (
 	"github.com/IceWhaleTech/CasaOS-Common/model"
 	"github.com/IceWhaleTech/CasaOS-Common/utils/constants"
 	"github.com/IceWhaleTech/CasaOS-Common/utils/logger"
-	"github.com/gin-gonic/gin"
 
 	util_http "github.com/IceWhaleTech/CasaOS-Common/utils/http"
 
+	"github.com/IceWhaleTech/CasaOS/codegen/message_bus"
+	"github.com/IceWhaleTech/CasaOS/common"
 	"github.com/IceWhaleTech/CasaOS/pkg/cache"
 	"github.com/IceWhaleTech/CasaOS/pkg/config"
 	"github.com/IceWhaleTech/CasaOS/pkg/sqlite"
@@ -25,7 +27,6 @@ import (
 	"github.com/IceWhaleTech/CasaOS/pkg/utils/file"
 	"github.com/IceWhaleTech/CasaOS/route"
 	"github.com/IceWhaleTech/CasaOS/service"
-	"github.com/IceWhaleTech/CasaOS/types"
 	"github.com/coreos/go-systemd/daemon"
 	"go.uber.org/zap"
 
@@ -56,7 +57,7 @@ var (
 func init() {
 	flag.Parse()
 	if *versionFlag {
-		fmt.Println("v" + types.CURRENTVERSION)
+		fmt.Println("v" + common.VERSION)
 		return
 	}
 
@@ -73,7 +74,7 @@ func init() {
 	sqliteDB = sqlite.GetDb(*dbFlag)
 	// gredis.GetRedisConn(config.RedisInfo),
 
-	service.MyService = service.NewService(sqliteDB, config.CommonInfo.RuntimePath, route.SocketIo())
+	service.MyService = service.NewService(sqliteDB, config.CommonInfo.RuntimePath)
 
 	service.Cache = cache.Init()
 
@@ -99,10 +100,6 @@ func main() {
 	}
 
 	v1Router := route.InitV1Router()
-	defer service.SocketServer.Close()
-
-	v1Router.GET("/v1/socketio/*any", gin.WrapH(service.SocketServer))
-	v1Router.POST("/v1/socketio/*any", gin.WrapH(service.SocketServer))
 
 	v2Router := route.InitV2Router()
 	v2DocRouter := route.InitV2DocRouter(_docHTML, _docYAML)
@@ -148,7 +145,6 @@ func main() {
 		"/v1/image",
 		"/v1/samba",
 		"/v1/notify",
-		"/v1/socketio",
 		"/v1/driver",
 		"/v1/cloud",
 		"/v1/recover",
@@ -166,6 +162,21 @@ func main() {
 			panic(err)
 		}
 	}
+	var events []message_bus.EventType
+	events = append(events, message_bus.EventType{Name: "casaos:system:utilization", SourceID: common.SERVICENAME, PropertyTypeList: []message_bus.PropertyType{}})
+	events = append(events, message_bus.EventType{Name: "casaos:file:recover", SourceID: common.SERVICENAME, PropertyTypeList: []message_bus.PropertyType{}})
+	events = append(events, message_bus.EventType{Name: "casaos:file:operate", SourceID: common.SERVICENAME, PropertyTypeList: []message_bus.PropertyType{}})
+	// register at message bus
+
+	response, err := service.MyService.MessageBus().RegisterEventTypesWithResponse(context.Background(), events)
+	if err != nil {
+		logger.Error("error when trying to register one or more event types - some event type will not be discoverable", zap.Error(err))
+	}
+
+	if response != nil && response.StatusCode() != http.StatusOK {
+		logger.Error("error when trying to register one or more event types - some event type will not be discoverable", zap.String("status", response.Status()), zap.String("body", string(response.Body)))
+	}
+
 	go func() {
 		time.Sleep(time.Second * 2)
 		// v0.3.6

pkg/utils/version/version.go

@@ -14,15 +14,15 @@ import (
 	"strconv"
 	"strings"
 
+	"github.com/IceWhaleTech/CasaOS/common"
 	"github.com/IceWhaleTech/CasaOS/model"
-	"github.com/IceWhaleTech/CasaOS/types"
 )
 
 func IsNeedUpdate(version model.Version) (bool, model.Version) {
 
 	v1 := strings.Split(version.Version, ".")
 
-	v2 := strings.Split(types.CURRENTVERSION, ".")
+	v2 := strings.Split(common.VERSION, ".")
 
 	for len(v1) < len(v2) {
 		v1 = append(v1, "0")

route/init.go

@@ -18,13 +18,13 @@ import (
 	"time"
 
 	"github.com/IceWhaleTech/CasaOS-Common/utils/logger"
+	"github.com/IceWhaleTech/CasaOS/common"
 	"github.com/IceWhaleTech/CasaOS/model"
 	"github.com/IceWhaleTech/CasaOS/pkg/config"
 	"github.com/IceWhaleTech/CasaOS/pkg/samba"
 	"github.com/IceWhaleTech/CasaOS/pkg/utils/encryption"
 	"github.com/IceWhaleTech/CasaOS/pkg/utils/file"
 	"github.com/IceWhaleTech/CasaOS/service"
-	"github.com/IceWhaleTech/CasaOS/types"
 	"go.uber.org/zap"
 )
 
@@ -50,7 +50,7 @@ func InitInfo() {
 		logger.Error("GetMacAddress", zap.String("error", err.Error()))
 	}
 	mb.Hash = encryption.GetMD5ByStr(mac)
-	mb.Version = types.CURRENTVERSION
+	mb.Version = common.VERSION
 	os.Remove(config.AppInfo.DBPath + "/baseinfo.conf")
 	by, err := json.Marshal(mb)
 	if err != nil {

route/periodical.go

@@ -69,7 +69,7 @@ func SendAllHardwareStatusBySocket() {
 	for k, v := range systemTempMap {
 		body[k] = v
 	}
-	service.MyService.Notify().SendNotify("sys_hardware_status", body)
+	service.MyService.Notify().SendNotify("casaos:system:utilization", body)
 }
 
 // func MonitoryUSB() {

route/socket.go

@@ -1,62 +0,0 @@
-/*
- * @Author: LinkLeong link@icewhale.com
- * @Date: 2022-05-23 17:18:56
- * @LastEditors: LinkLeong
- * @LastEditTime: 2022-06-09 21:48:10
- * @FilePath: /CasaOS/route/socket.go
- * @Description:
- * @Website: https://www.casaos.io
- * Copyright (c) 2022 by icewhale, All Rights Reserved.
- */
-package route
-
-import (
-	"github.com/IceWhaleTech/CasaOS-Common/utils/logger"
-	"github.com/IceWhaleTech/CasaOS/service"
-	socketio "github.com/googollee/go-socket.io"
-	"go.uber.org/zap"
-)
-
-func SocketIo() *socketio.Server {
-	server := socketio.NewServer(nil)
-	server.OnConnect("/", func(s socketio.Conn) error {
-		s.SetContext("")
-		logger.Info("connected", zap.Any("id", s.ID()))
-		s.Join("public")
-		service.ClientCount += 1
-		return nil
-	})
-
-	server.OnEvent("/", "notice", func(s socketio.Conn, msg string) {
-		logger.Info("notice", zap.Any("msg", msg))
-		s.Emit("reply", "have "+msg)
-	})
-
-	// server.OnEvent("/chat", "msg", func(s socketio.Conn, msg string) string {
-	// 	s.SetContext(msg)
-	// 	return "recv " + msg
-	// })
-
-	// server.OnEvent("/", "bye", func(s socketio.Conn) string {
-	// 	last := s.Context().(string)
-	// 	s.Emit("bye", last)
-	// 	s.Close()
-	// 	return last
-	// })
-
-	server.OnError("/", func(s socketio.Conn, e error) {
-		logger.Error("meet error", zap.Any("error", e))
-	})
-
-	server.OnDisconnect("/", func(s socketio.Conn, reason string) {
-		service.ClientCount -= 1
-		logger.Info("closed", zap.Any("reason", reason))
-	})
-
-	go func() {
-		if err := server.Serve(); err != nil {
-			logger.Error("error when trying to  listen socketio ", zap.Any("error", err))
-		}
-	}()
-	return server
-}

route/v1.go

@@ -155,14 +155,10 @@ func InitV1Router() *gin.Engine {
 			v1NotifyGroup.POST("/:path", v1.PostNotifyMessage)
 			// merge to system
 			v1NotifyGroup.POST("/system_status", v1.PostSystemStatusNotify)
-			v1NotifyGroup.POST("/install_app", v1.PostInstallAppNotify)
+			//v1NotifyGroup.POST("/install_app", v1.PostInstallAppNotify)
-			v1NotifyGroup.POST("/uninstall_app", v1.PostUninstallAppNotify)
+			//v1NotifyGroup.POST("/uninstall_app", v1.PostUninstallAppNotify)
 		}
 	}
 
-	// socketio
-	// v1Group.GET("/socketio/*any", gin.WrapH(service.SocketServer))
-	// v1Group.POST("/socketio/*any", gin.WrapH(service.SocketServer))
-
 	return r
 }

route/v1/notify.go

@@ -11,14 +11,14 @@ import (
 )
 
 func PostNotifyMessage(c *gin.Context) {
-	path := c.Param("path")
+	name := c.Param("name")
 	message := make(map[string]interface{})
 	if err := c.ShouldBind(&message); err != nil {
 		c.JSON(http.StatusBadRequest, model.Result{Success: common_err.INVALID_PARAMS, Message: err.Error()})
 		return
 	}
 
-	service.MyService.Notify().SendNotify(path, message)
+	service.MyService.Notify().SendNotify(name, message)
 	c.JSON(common_err.SUCCESS, model.Result{Success: common_err.SUCCESS, Message: common_err.GetMsg(common_err.SUCCESS)})
 }
 
@@ -40,7 +40,7 @@ func PostInstallAppNotify(c *gin.Context) {
 		return
 	}
 
-	service.MyService.Notify().SendInstallAppBySocket(app)
+	//service.MyService.Notify().SendInstallAppBySocket(app)
 	c.JSON(common_err.SUCCESS, model.Result{Success: common_err.SUCCESS, Message: common_err.GetMsg(common_err.SUCCESS)})
 }
 
@@ -51,6 +51,6 @@ func PostUninstallAppNotify(c *gin.Context) {
 		return
 	}
 
-	service.MyService.Notify().SendUninstallAppBySocket(app)
+	//service.MyService.Notify().SendUninstallAppBySocket(app)
 	c.JSON(common_err.SUCCESS, model.Result{Success: common_err.SUCCESS, Message: common_err.GetMsg(common_err.SUCCESS)})
 }

route/v1/recover.go

@@ -23,7 +23,7 @@ func GetRecoverStorage(c *gin.Context) {
 			c.String(200, `<p>Code cannot be empty</p><script>window.close()</script>`)
 			notify["status"] = "fail"
 			notify["message"] = "Code cannot be empty"
-			service.MyService.Notify().SendNotify("recover_status", notify)
+			service.MyService.Notify().SendNotify("casaos:file:recover", notify)
 			return
 		}
 
@@ -38,7 +38,7 @@ func GetRecoverStorage(c *gin.Context) {
 			c.String(200, `<p>Initialization failure:`+err.Error()+`</p><script>window.close()</script>`)
 			notify["status"] = "fail"
 			notify["message"] = "Initialization failure"
-			service.MyService.Notify().SendNotify("recover_status", notify)
+			service.MyService.Notify().SendNotify("casaos:file:recover", notify)
 			return
 		}
 
@@ -47,7 +47,7 @@ func GetRecoverStorage(c *gin.Context) {
 			c.String(200, `<p>Failed to get user information:`+err.Error()+`</p><script>window.close()</script>`)
 			notify["status"] = "fail"
 			notify["message"] = "Failed to get user information"
-			service.MyService.Notify().SendNotify("recover_status", notify)
+			service.MyService.Notify().SendNotify("casaos:file:recover", notify)
 			return
 		}
 		if len(username) > 0 {
@@ -61,7 +61,7 @@ func GetRecoverStorage(c *gin.Context) {
 			service.MyService.Storage().CheckAndMountByName(username)
 			notify["status"] = "warn"
 			notify["message"] = "The same configuration has been added"
-			service.MyService.Notify().SendNotify("recover_status", notify)
+			service.MyService.Notify().SendNotify("casaos:file:recover", notify)
 			return
 		}
 		dmap := make(map[string]string)
@@ -75,7 +75,7 @@ func GetRecoverStorage(c *gin.Context) {
 		notify := make(map[string]interface{})
 		notify["status"] = "success"
 		notify["message"] = "Success"
-		service.MyService.Notify().SendNotify("recover_status", notify)
+		service.MyService.Notify().SendNotify("casaos:file:recover", notify)
 	} else if t == "Dropbox" {
 		add := dropbox.Addition{}
 		add.Code = c.Query("code")
@@ -83,7 +83,7 @@ func GetRecoverStorage(c *gin.Context) {
 			c.String(200, `<p>Code cannot be empty</p><script>window.close()</script>`)
 			notify["status"] = "fail"
 			notify["message"] = "Code cannot be empty"
-			service.MyService.Notify().SendNotify("recover_status", notify)
+			service.MyService.Notify().SendNotify("casaos:file:recover", notify)
 			return
 		}
 		add.RootFolderID = ""
@@ -96,7 +96,7 @@ func GetRecoverStorage(c *gin.Context) {
 			c.String(200, `<p>Initialization failure:`+err.Error()+`</p><script>window.close()</script>`)
 			notify["status"] = "fail"
 			notify["message"] = "Initialization failure"
-			service.MyService.Notify().SendNotify("recover_status", notify)
+			service.MyService.Notify().SendNotify("casaos:file:recover", notify)
 			return
 		}
 		username, err := dropbox.GetUserInfo(c)
@@ -104,7 +104,7 @@ func GetRecoverStorage(c *gin.Context) {
 			c.String(200, `<p>Failed to get user information:`+err.Error()+`</p><script>window.close()</script>`)
 			notify["status"] = "fail"
 			notify["message"] = "Failed to get user information"
-			service.MyService.Notify().SendNotify("recover_status", notify)
+			service.MyService.Notify().SendNotify("casaos:file:recover", notify)
 			return
 		}
 		if len(username) > 0 {
@@ -118,7 +118,7 @@ func GetRecoverStorage(c *gin.Context) {
 			service.MyService.Storage().CheckAndMountByName(username)
 			notify["status"] = "warn"
 			notify["message"] = "The same configuration has been added"
-			service.MyService.Notify().SendNotify("recover_status", notify)
+			service.MyService.Notify().SendNotify("casaos:file:recover", notify)
 			return
 		}
 		dmap := make(map[string]string)
@@ -143,7 +143,7 @@ func GetRecoverStorage(c *gin.Context) {
 
 		notify["status"] = "success"
 		notify["message"] = "Success"
-		service.MyService.Notify().SendNotify("recover_status", notify)
+		service.MyService.Notify().SendNotify("casaos:file:recover", notify)
 	}
 
 	c.String(200, `<p>Just close the page</p><script>window.close()</script>`)

route/v1/system.go

@@ -14,6 +14,7 @@ import (
 
 	http2 "github.com/IceWhaleTech/CasaOS-Common/utils/http"
 	"github.com/IceWhaleTech/CasaOS-Common/utils/port"
+	"github.com/IceWhaleTech/CasaOS/common"
 	"github.com/IceWhaleTech/CasaOS/model"
 	"github.com/IceWhaleTech/CasaOS/pkg/config"
 	"github.com/IceWhaleTech/CasaOS/pkg/utils/common_err"
@@ -46,7 +47,7 @@ func GetSystemCheckVersion(c *gin.Context) {
 	data := make(map[string]interface{}, 3)
 	data["need_update"] = need
 	data["version"] = version
-	data["current_version"] = types.CURRENTVERSION
+	data["current_version"] = common.VERSION
 	c.JSON(common_err.SUCCESS, model.Result{Success: common_err.SUCCESS, Message: common_err.GetMsg(common_err.SUCCESS), Data: data})
 }
 
@@ -92,7 +93,7 @@ func GetSystemConfigDebug(c *gin.Context) {
 	 - Remote Version: %s
 	 - Browser: $Browser$ 
 	 - Version: $Version$
-`, sys.OS, types.CURRENTVERSION, disk.Total>>20, disk.Used>>20, array, version.Version)
+`, sys.OS, common.VERSION, disk.Total>>20, disk.Used>>20, array, version.Version)
 
 	//	array = append(array, fmt.Sprintf("disk,total:%v,used:%v,UsedPercent:%v", disk.Total>>20, disk.Used>>20, disk.UsedPercent))
 

service/notify.go

@@ -1,15 +1,20 @@
 package service
 
 import (
+	"context"
+	"encoding/json"
 	json2 "encoding/json"
 	"fmt"
+	"net/http"
 	"time"
 
-	notifyCommon "github.com/IceWhaleTech/CasaOS-Common/model/notify"
+	"github.com/IceWhaleTech/CasaOS-Common/utils/logger"
+	"github.com/IceWhaleTech/CasaOS/common"
 	model2 "github.com/IceWhaleTech/CasaOS/model"
 	"github.com/IceWhaleTech/CasaOS/model/notify"
 	"github.com/IceWhaleTech/CasaOS/service/model"
 	"github.com/IceWhaleTech/CasaOS/types"
+	"go.uber.org/zap"
 
 	socketio "github.com/googollee/go-socket.io"
 	"github.com/gorilla/websocket"
@@ -30,12 +35,11 @@ type NotifyServer interface {
 	GetList(c int) (list []model.AppNotify)
 	MarkRead(id string, state int)
 	//	SendText(m model.AppNotify)
-	SendUninstallAppBySocket(app notifyCommon.Application)
+	//	SendUninstallAppBySocket(app notifyCommon.Application)
 
 	SendFileOperateNotify(nowSend bool)
-	SendInstallAppBySocket(app notifyCommon.Application)
+	//SendInstallAppBySocket(app notifyCommon.Application)
-	SendStorageBySocket(message notify.StorageMessage)
+	SendNotify(name string, message map[string]interface{})
-	SendNotify(path string, message map[string]interface{})
 	SettingSystemTempData(message map[string]interface{})
 	GetSystemTempMap() map[string]interface{}
 }
@@ -51,18 +55,25 @@ func (i *notifyServer) SettingSystemTempData(message map[string]interface{}) {
 	}
 }
 
-func (i *notifyServer) SendNotify(path string, message map[string]interface{}) {
+func (i *notifyServer) SendNotify(name string, message map[string]interface{}) {
-	SocketServer.BroadcastToRoom("/", "public", path, message)
+	msg := make(map[string]string)
-}
+	for k, v := range message {
-
+		bt, _ := json.Marshal(v)
-func (i *notifyServer) SendStorageBySocket(message notify.StorageMessage) {
+		msg[k] = string(bt)
-	SocketServer.BroadcastToRoom("/", "public", "storage_status", message)
+	}
+	response, err := MyService.MessageBus().PublishEventWithResponse(context.Background(), common.SERVICENAME, name, msg)
+	if err != nil {
+		logger.Error("failed to publish event to message bus", zap.Error(err), zap.Any("event", msg))
+	}
+	if response.StatusCode() != http.StatusOK {
+		logger.Error("failed to publish event to message bus", zap.String("status", response.Status()), zap.Any("response", response))
+	}
+	// SocketServer.BroadcastToRoom("/", "public", path, message)
 }
 
 // Send periodic broadcast messages
 func (i *notifyServer) SendFileOperateNotify(nowSend bool) {
 	if nowSend {
-
 		len := 0
 		FileQueue.Range(func(k, v interface{}) bool {
 			len++
@@ -74,7 +85,18 @@ func (i *notifyServer) SendFileOperateNotify(nowSend bool) {
 		if len == 0 {
 			model.Data = []string{}
 			listMsg["file_operate"] = model
-			SocketServer.BroadcastToRoom("/", "public", "file_operate", listMsg)
+			msg := make(map[string]string)
+			for k, v := range listMsg {
+				bt, _ := json.Marshal(v)
+				msg[k] = string(bt)
+			}
+			response, err := MyService.MessageBus().PublishEventWithResponse(context.Background(), common.SERVICENAME, "casaos:file:operate", msg)
+			if err != nil {
+				logger.Error("failed to publish event to message bus", zap.Error(err), zap.Any("event", msg))
+			}
+			if response.StatusCode() != http.StatusOK {
+				logger.Error("failed to publish event to message bus", zap.String("status", response.Status()), zap.Any("response", response))
+			}
 			return
 		}
 
@@ -122,7 +144,19 @@ func (i *notifyServer) SendFileOperateNotify(nowSend bool) {
 		model.Data = list
 
 		listMsg["file_operate"] = model
-		SocketServer.BroadcastToRoom("/", "public", "file_operate", listMsg)
+		msg := make(map[string]string)
+		for k, v := range listMsg {
+			bt, _ := json.Marshal(v)
+			msg[k] = string(bt)
+		}
+		response, err := MyService.MessageBus().PublishEventWithResponse(context.Background(), common.SERVICENAME, "casaos:file:operate", msg)
+		if err != nil {
+			logger.Error("failed to publish event to message bus", zap.Error(err), zap.Any("event", msg))
+		}
+		if response.StatusCode() != http.StatusOK {
+			logger.Error("failed to publish event to message bus", zap.String("status", response.Status()), zap.Any("response", response))
+		}
+
 	} else {
 		for {
 
@@ -179,20 +213,30 @@ func (i *notifyServer) SendFileOperateNotify(nowSend bool) {
 			model.Data = list
 
 			listMsg["file_operate"] = model
-			SocketServer.BroadcastToRoom("/", "public", "file_operate", listMsg)
+			msg := make(map[string]string)
+			for k, v := range listMsg {
+				bt, _ := json.Marshal(v)
+				msg[k] = string(bt)
+			}
+			response, err := MyService.MessageBus().PublishEventWithResponse(context.Background(), common.SERVICENAME, "casaos:file:operate", msg)
+			if err != nil {
+				logger.Error("failed to publish event to message bus", zap.Error(err), zap.Any("event", msg))
+			}
+			if response.StatusCode() != http.StatusOK {
+				logger.Error("failed to publish event to message bus", zap.String("status", response.Status()), zap.Any("response", response))
+			}
 			time.Sleep(time.Second * 3)
 		}
 	}
 }
 
-func (i *notifyServer) SendInstallAppBySocket(app notifyCommon.Application) {
+// func (i *notifyServer) SendInstallAppBySocket(app notifyCommon.Application) {
-	SocketServer.BroadcastToRoom("/", "public", "app_install", app)
+// 	SocketServer.BroadcastToRoom("/", "public", "app_install", app)
+// }
 
-}
+// func (i *notifyServer) SendUninstallAppBySocket(app notifyCommon.Application) {
-
+// 	SocketServer.BroadcastToRoom("/", "public", "app_uninstall", app)
-func (i *notifyServer) SendUninstallAppBySocket(app notifyCommon.Application) {
+// }
-	SocketServer.BroadcastToRoom("/", "public", "app_uninstall", app)
-}
 
 func (i *notifyServer) SSR() {
 	server := socketio.NewServer(nil)

service/service.go

@@ -12,19 +12,17 @@ package service
 
 import (
 	"github.com/IceWhaleTech/CasaOS-Common/external"
-	"github.com/IceWhaleTech/CasaOS-Common/utils/logger"
+	"github.com/IceWhaleTech/CasaOS/codegen/message_bus"
-	socketio "github.com/googollee/go-socket.io"
+	"github.com/IceWhaleTech/CasaOS/pkg/config"
 	"github.com/gorilla/websocket"
 	"github.com/patrickmn/go-cache"
-	"go.uber.org/zap"
 	"gorm.io/gorm"
 )
 
 var Cache *cache.Cache
 
 var (
-	MyService    Repository
+	MyService Repository
-	SocketServer *socketio.Server
 )
 
 var (
@@ -47,13 +45,10 @@ type Repository interface {
 	FsListService() FsListService
 	FsLinkService() FsLinkService
 	FsService() FsService
+	MessageBus() *message_bus.ClientWithResponses
 }
 
-func NewService(db *gorm.DB, RuntimePath string, socket *socketio.Server) Repository {
+func NewService(db *gorm.DB, RuntimePath string) Repository {
-	if socket == nil {
-		logger.Error("socket is nil", zap.Any("error", "socket is nil"))
-	}
-	SocketServer = socket
 	gatewayManagement, err := external.NewManagementService(RuntimePath)
 	if err != nil && len(RuntimePath) > 0 {
 		panic(err)
@@ -149,3 +144,23 @@ func (c *store) Casa() CasaService {
 func (c *store) Health() HealthService {
 	return c.health
 }
+
+func (c *store) MessageBus() *message_bus.ClientWithResponses {
+	client, _ := message_bus.NewClientWithResponses("", func(c *message_bus.Client) error {
+		// error will never be returned, as we always want to return a client, even with wrong address,
+		// in order to avoid panic.
+		//
+		// If we don't avoid panic, message bus becomes a hard dependency, which is not what we want.
+
+		messageBusAddress, err := external.GetMessageBusAddress(config.CommonInfo.RuntimePath)
+		if err != nil {
+			c.Server = "message bus address not found"
+			return nil
+		}
+
+		c.Server = messageBusAddress
+		return nil
+	})
+
+	return client
+}

types/system.go

@@ -1,24 +0,0 @@
-/*
- * @Author: LinkLeong link@icewhale.org
- * @Date: 2022-11-15 15:51:44
- * @LastEditors: LinkLeong
- * @LastEditTime: 2022-11-15 15:56:03
- * @FilePath: /CasaOS/types/system.go
- * @Description:
- * @Website: https://www.casaos.io
- * Copyright (c) 2022 by icewhale, All Rights Reserved.
- */
-/*@Author: LinkLeong link@icewhale.com
- *@Date: 2022-02-17 18:53:22
- *@LastEditors: LinkLeong
- *@LastEditTime: 2022-09-06 14:27:42
- *@FilePath: /CasaOS/types/system.go
- *@Description:
- *@Website: https://www.casaos.io
- *Copyright (c) 2022 by icewhale, All Rights Reserved.
- */
-package types
-
-const CURRENTVERSION = "0.4.2"
-
-const BODY = " "