diff --git a/cmd/server/main-server.go b/cmd/server/main-server.go index c4646135..233596a7 100644 --- a/cmd/server/main-server.go +++ b/cmd/server/main-server.go @@ -25,6 +25,7 @@ import ( "github.com/wavetermdev/thenextwave/pkg/wcloud" "github.com/wavetermdev/thenextwave/pkg/wconfig" "github.com/wavetermdev/thenextwave/pkg/web" + "github.com/wavetermdev/thenextwave/pkg/wps" "github.com/wavetermdev/thenextwave/pkg/wshrpc/wshserver" "github.com/wavetermdev/thenextwave/pkg/wshutil" "github.com/wavetermdev/thenextwave/pkg/wstore" @@ -151,6 +152,7 @@ func createMainWshClient() { rpc := wshserver.GetMainRpcClient() wshutil.DefaultRouter.RegisterRoute("wavesrv", rpc) wshutil.DefaultRouter.SetDefaultRoute("wavesrv") + wps.Broker.SetClient(wshutil.DefaultRouter) } func main() { diff --git a/pkg/wshrpc/wshrpctypes.go b/pkg/wshrpc/wshrpctypes.go index 4720f3f5..03e8e480 100644 --- a/pkg/wshrpc/wshrpctypes.go +++ b/pkg/wshrpc/wshrpctypes.go @@ -49,6 +49,7 @@ const ( Command_Test = "test" Command_RemoteStreamFile = "remotestreamfile" Command_RemoteFileInfo = "remotefileinfo" + Command_Event = "event" ) type RespOrErrorUnion[T any] struct { diff --git a/pkg/wshutil/wshrouter.go b/pkg/wshutil/wshrouter.go index d04ff5bd..657eaabc 100644 --- a/pkg/wshutil/wshrouter.go +++ b/pkg/wshutil/wshrouter.go @@ -47,6 +47,25 @@ func noRouteErr(routeId string) error { return fmt.Errorf("no route for %q", routeId) } +func (router *WshRouter) SendEvent(fullRoute string, event wshrpc.WaveEvent) { + nextRoute, routeId := popRoute(fullRoute) + rpc := router.GetRpc(routeId) + if rpc == nil { + return + } + msg := RpcMessage{ + Command: wshrpc.Command_Event, + Route: nextRoute, + Data: event, + } + msgBytes, err := json.Marshal(msg) + if err != nil { + // nothing to do + return + } + rpc.SendRpcMessage(msgBytes) +} + func (router *WshRouter) handleNoRoute(msg RpcMessage) { nrErr := noRouteErr(msg.Route) if msg.ReqId == "" {