waveterm/backup
0
0
Fork 0
mirror of https://github.com/wavetermdev/backup.git synced 2026-08-05 13:57:07 -07:00
Code Issues Packages Projects Releases Wiki Activity

wsh routing + proxy (#224)

Browse Source 
lots of changes, including:
* source/route to rpcmessage
* rpcproxy
* wshrouter
* bug fixing
* wps uses routeids not clients
This commit is contained in:
Mike Sawka
2024-08-13 16:52:35 -07:00
committed by GitHub
parent 01d788b780
commit 844451ea0d
25 changed files with 716 additions and 323 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/wshrpc/wshserver/wshserver.go
+34 -48
View File
@@ -30,26 +30,33 @@ import (
const SimpleId_This = "this"
type WshServer struct{}
func (*WshServer) WshServerImpl() {}
var WshServerImpl = WshServer{}
func (ws *WshServer) TestCommand(ctx context.Context, data string) error {
defer func() {
if r := recover(); r != nil {
log.Printf("panic in TestCommand: %v", r)
}
}()
rpc := wshutil.GetWshRpcFromContext(ctx)
if rpc == nil {
rpcSource := wshutil.GetRpcSourceFromContext(ctx)
log.Printf("TEST src:%s | %s\n", rpcSource, data)
if rpcSource == "" {
return nil
}
go func() {
wshclient.MessageCommand(rpc, wshrpc.CommandMessageData{Message: "test message"}, &wshrpc.WshRpcCommandOpts{NoResponse: true})
resp, err := wshclient.RemoteFileInfoCommand(rpc, "~/work/wails/thenextwave/README.md", nil)
mainClient := GetMainRpcClient()
wshclient.MessageCommand(mainClient, wshrpc.CommandMessageData{Message: "test message"}, &wshrpc.RpcOpts{NoResponse: true, Route: rpcSource})
resp, err := wshclient.RemoteFileInfoCommand(mainClient, "~/work/wails/thenextwave/README.md", &wshrpc.RpcOpts{Route: rpcSource})
if err != nil {
log.Printf("error getting remote file info: %v", err)
return
}
log.Printf("remote file info: %#v\n", resp)
rch := wshclient.RemoteStreamFileCommand(rpc, wshrpc.CommandRemoteStreamFileData{Path: "~/work/wails/thenextwave/README.md"}, nil)
rch := wshclient.RemoteStreamFileCommand(mainClient, wshrpc.CommandRemoteStreamFileData{Path: "~/work/wails/thenextwave/README.md"}, &wshrpc.RpcOpts{Route: rpcSource})
for msg := range rch {
if msg.Error != nil {
log.Printf("error in stream: %v", msg.Error)
@@ -66,22 +73,6 @@ func (ws *WshServer) TestCommand(ctx context.Context, data string) error {
return nil
}
func (ws *WshServer) AuthenticateCommand(ctx context.Context, data string) error {
w := wshutil.GetWshRpcFromContext(ctx)
if w == nil {
return fmt.Errorf("no wshrpc in context")
}
newCtx, err := wshutil.ValidateAndExtractRpcContextFromToken(data)
if err != nil {
return fmt.Errorf("error validating token: %w", err)
}
if newCtx == nil {
return fmt.Errorf("no context found in jwt token")
}
w.SetRpcContext(*newCtx)
return nil
}
// for testing
func (ws *WshServer) MessageCommand(ctx context.Context, data wshrpc.CommandMessageData) error {
log.Printf("MESSAGE: %s | %q\n", data.ORef, data.Message)
@@ -228,17 +219,12 @@ func sendWaveObjUpdate(oref waveobj.ORef) {
})
}
func resolveSimpleId(ctx context.Context, simpleId string) (*waveobj.ORef, error) {
func resolveSimpleId(ctx context.Context, data wshrpc.CommandResolveIdsData, simpleId string) (*waveobj.ORef, error) {
if simpleId == SimpleId_This {
wshRpc := wshutil.GetWshRpcFromContext(ctx)
if wshRpc == nil {
return nil, fmt.Errorf("no wshrpc in context")
if data.BlockId == "" {
return nil, fmt.Errorf("no blockid in request")
}
rpcCtx := wshRpc.GetRpcContext()
if rpcCtx.BlockId == "" {
return nil, fmt.Errorf("no blockid in rpc context")
}
return &waveobj.ORef{OType: wstore.OType_Block, OID: rpcCtx.BlockId}, nil
return &waveobj.ORef{OType: wstore.OType_Block, OID: data.BlockId}, nil
}
if strings.Contains(simpleId, ":") {
rtn, err := waveobj.ParseORef(simpleId)
@@ -254,7 +240,7 @@ func (ws *WshServer) ResolveIdsCommand(ctx context.Context, data wshrpc.CommandR
rtn := wshrpc.CommandResolveIdsRtnData{}
rtn.ResolvedIds = make(map[string]waveobj.ORef)
for _, simpleId := range data.Ids {
oref, err := resolveSimpleId(ctx, simpleId)
oref, err := resolveSimpleId(ctx, data, simpleId)
if err != nil || oref == nil {
continue
}
@@ -471,40 +457,40 @@ func (ws *WshServer) EventRecvCommand(ctx context.Context, data wshrpc.WaveEvent
}
func (ws *WshServer) EventPublishCommand(ctx context.Context, data wshrpc.WaveEvent) error {
wrpc := wshutil.GetWshRpcFromContext(ctx)
if wrpc == nil {
return fmt.Errorf("no wshrpc in context")
rpcSource := wshutil.GetRpcSourceFromContext(ctx)
if rpcSource == "" {
return fmt.Errorf("no rpc source set")
}
if data.Sender == "" {
data.Sender = wrpc.ClientId()
data.Sender = rpcSource
}
wps.Broker.Publish(data)
return nil
}
func (ws *WshServer) EventSubCommand(ctx context.Context, data wshrpc.SubscriptionRequest) error {
wrpc := wshutil.GetWshRpcFromContext(ctx)
if wrpc == nil {
return fmt.Errorf("no wshrpc in context")
rpcSource := wshutil.GetRpcSourceFromContext(ctx)
if rpcSource == "" {
return fmt.Errorf("no rpc source set")
}
wps.Broker.Subscribe(wrpc, data)
wps.Broker.Subscribe(rpcSource, data)
return nil
}
func (ws *WshServer) EventUnsubCommand(ctx context.Context, data wshrpc.SubscriptionRequest) error {
wrpc := wshutil.GetWshRpcFromContext(ctx)
if wrpc == nil {
return fmt.Errorf("no wshrpc in context")
rpcSource := wshutil.GetRpcSourceFromContext(ctx)
if rpcSource == "" {
return fmt.Errorf("no rpc source set")
}
wps.Broker.Unsubscribe(wrpc, data)
wps.Broker.Unsubscribe(rpcSource, data)
return nil
}
func (ws *WshServer) EventUnsubAllCommand(ctx context.Context) error {
wrpc := wshutil.GetWshRpcFromContext(ctx)
if wrpc == nil {
return fmt.Errorf("no wshrpc in context")
rpcSource := wshutil.GetRpcSourceFromContext(ctx)
if rpcSource == "" {
return fmt.Errorf("no rpc source set")
}
wps.Broker.UnsubscribeAll(wrpc)
wps.Broker.UnsubscribeAll(rpcSource)
return nil
}
pkg/wshrpc/wshserver/wshserverutil.go
+34 -73
View File
@@ -4,89 +4,42 @@
package wshserver
import (
"context"
"fmt"
"log"
"net"
"reflect"
"sync"
"github.com/wavetermdev/thenextwave/pkg/wshrpc"
"github.com/wavetermdev/thenextwave/pkg/wshutil"
)
// this file contains the generic types and functions that create and power the WSH server
const (
DefaultOutputChSize = 32
DefaultInputChSize = 32
)
type WshServer struct{}
func (*WshServer) WshServerImpl() {}
type WshServerMethodDecl struct {
Command string
CommandType string
MethodName string
Method reflect.Value
CommandDataType reflect.Type
DefaultResponseDataType reflect.Type
RequestDataTypes []reflect.Type // for streaming requests
ResponseDataTypes []reflect.Type // for streaming responses
}
var WshServerImpl = WshServer{}
var contextRType = reflect.TypeOf((*context.Context)(nil)).Elem()
var wshCommandDeclMap = wshrpc.GenerateWshCommandDeclMap()
func GetWshServerMethod(command string, commandType string, methodName string, methodFunc any) *WshServerMethodDecl {
methodVal := reflect.ValueOf(methodFunc)
methodType := methodVal.Type()
if methodType.Kind() != reflect.Func {
panic(fmt.Sprintf("methodVal must be a function got [%v]", methodType))
}
if methodType.In(0) != contextRType {
panic(fmt.Sprintf("methodVal must have a context as the first argument %v", methodType))
}
var defResponseType reflect.Type
if methodType.NumOut() > 1 {
defResponseType = methodType.Out(0)
}
var cdataType reflect.Type
if methodType.NumIn() > 1 {
cdataType = methodType.In(1)
}
rtn := &WshServerMethodDecl{
Command: command,
CommandType: commandType,
MethodName: methodName,
Method: methodVal,
CommandDataType: cdataType,
DefaultResponseDataType: defResponseType,
}
return rtn
}
func decodeRtnVals(rtnVals []reflect.Value) (any, error) {
switch len(rtnVals) {
case 0:
return nil, nil
case 1:
errIf := rtnVals[0].Interface()
if errIf == nil {
return nil, nil
func handleDomainSocketClient(conn net.Conn) {
proxy := wshutil.MakeRpcProxy()
go func() {
writeErr := wshutil.AdaptOutputChToStream(proxy.ToRemoteCh, conn)
if writeErr != nil {
log.Printf("error writing to domain socket: %v\n", writeErr)
}
return nil, errIf.(error)
case 2:
errIf := rtnVals[1].Interface()
if errIf == nil {
return rtnVals[0].Interface(), nil
}
return rtnVals[0].Interface(), errIf.(error)
default:
return nil, fmt.Errorf("too many return values: %d", len(rtnVals))
}()
go func() {
// when input is closed, close the connection
defer conn.Close()
wshutil.AdaptStreamToMsgCh(conn, proxy.FromRemoteCh)
}()
rpcCtx, err := proxy.HandleAuthentication()
if err != nil {
conn.Close()
log.Printf("error handling authentication: %v\n", err)
return
}
// now that we're authenticated, set the ctx and attach to the router
log.Printf("domain socket connection authenticated: %#v\n", rpcCtx)
proxy.SetRpcContext(rpcCtx)
wshutil.DefaultRouter.RegisterRoute("controller:"+rpcCtx.BlockId, proxy)
}
func RunWshRpcOverListener(listener net.Listener) {
@@ -98,11 +51,19 @@ func RunWshRpcOverListener(listener net.Listener) {
continue
}
log.Print("got domain socket connection\n")
// TODO deal with closing connection
go wshutil.SetupConnRpcClient(conn, &WshServerImpl)
go handleDomainSocketClient(conn)
}
}
func MakeWshServer(inputCh chan []byte, outputCh chan []byte, initialCtx wshrpc.RpcContext) {
wshutil.MakeWshRpc(inputCh, outputCh, initialCtx, &WshServerImpl)
var waveSrvClient_Singleton *wshutil.WshRpc
var waveSrvClient_Once = &sync.Once{}
// returns the wavesrv main rpc client singleton
func GetMainRpcClient() *wshutil.WshRpc {
waveSrvClient_Once.Do(func() {
inputCh := make(chan []byte, DefaultInputChSize)
outputCh := make(chan []byte, DefaultOutputChSize)
waveSrvClient_Singleton = wshutil.MakeWshRpc(inputCh, outputCh, wshrpc.RpcContext{}, &WshServerImpl)
})
return waveSrvClient_Singleton
}