Upstream/waveterm
1
0
Fork 0
mirror of https://github.com/wavetermdev/waveterm.git synced 2024-12-22 16:48:23 +01:00
Code Issues Packages Projects Releases Wiki Activity

checkpoint for migrating to remoteinstance

Browse Source
This commit is contained in:
sawka 2022-07-05 16:54:49 -07:00
parent 123fdfe3bb
commit 2755be315d
6 changed files with 121 additions and 65 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
cmd/main-server.go
View File

@ -377,6 +377,7 @@ func WriteJsonSuccess(w http.ResponseWriter, data interface{}) {
type runCommandResponse struct {
Line *sstore.LineType `json:"line"`
Cmd *sstore.CmdType `json:"cmd"`
}
func HandleRunCommand(w http.ResponseWriter, r *http.Request) {
@ -400,16 +401,16 @@ func HandleRunCommand(w http.ResponseWriter, r *http.Request) {
WriteJsonError(w, fmt.Errorf("invalid sessionid '%s': %w", commandPk.SessionId, err))
return
}
line, err := ProcessFeCommandPacket(r.Context(), &commandPk)
resp, err := ProcessFeCommandPacket(r.Context(), &commandPk)
if err != nil {
WriteJsonError(w, err)
return
}
WriteJsonSuccess(w, &runCommandResponse{Line: line})
WriteJsonSuccess(w, resp)
return
}
func ProcessFeCommandPacket(ctx context.Context, pk *scpacket.FeCommandPacketType) (*sstore.LineType, error) {
func ProcessFeCommandPacket(ctx context.Context, pk *scpacket.FeCommandPacketType) (*runCommandResponse, error) {
commandStr := strings.TrimSpace(pk.CmdStr)
if commandStr == "" {
return nil, fmt.Errorf("invalid emtpty command")
@ -420,14 +421,14 @@ func ProcessFeCommandPacket(ctx context.Context, pk *scpacket.FeCommandPacketTyp
if err != nil {
return nil, err
}
return rtnLine, nil
return &runCommandResponse{Line: rtnLine}, nil
}
if strings.HasPrefix(commandStr, "cd ") {
newDir := strings.TrimSpace(commandStr[3:])
cdPacket := packet.MakeCdPacket()
cdPacket.PacketId = uuid.New().String()
cdPacket.ReqId = uuid.New().String()
cdPacket.Dir = newDir
localRemote := remote.GetRemote("local")
localRemote := remote.GetRemoteById(pk.RemoteState.RemoteId)
if localRemote != nil {
localRemote.Input.SendPacket(cdPacket)
}
@ -437,19 +438,8 @@ func ProcessFeCommandPacket(ctx context.Context, pk *scpacket.FeCommandPacketTyp
if err != nil {
return nil, err
}
runPacket := packet.MakeRunPacket()
runPacket.CK = base.MakeCommandKey(pk.SessionId, rtnLine.CmdId)
runPacket.Cwd = pk.RemoteState.Cwd
runPacket.Env = nil
runPacket.Command = commandStr
fmt.Printf("run-packet %v\n", runPacket)
go func() {
localRemote := remote.GetRemote("local")
if localRemote != nil {
localRemote.Input.SendPacket(runPacket)
}
}()
return rtnLine, nil
err = remote.RunCommand(pk, rtnLine.CmdId)
return &runCommandResponse{Line: rtnLine}, nil
}
// /api/start-session

2
db/migrations/000001_init.down.sql
View File

@ -1,6 +1,6 @@
DROP TABLE session;
DROP TABLE window;
DROP TABLE session_remote;
DROP TABLE remote_instance;
DROP TABLE line;
DROP TABLE remote;
DROP TABLE session_cmd;

11
db/migrations/000001_init.up.sql
View File

@ -14,13 +14,14 @@ CREATE TABLE window (
);
CREATE UNIQUE INDEX window_name_unique ON window(sessionid, name);
CREATE TABLE session_remote (
CREATE TABLE remote_instance (
riid varchar(36) PRIMARY KEY,
name varchar(50) NOT NULL,
sessionid varchar(36) NOT NULL,
windowid varchar(36) NOT NULL,
remotename varchar(50) NOT NULL,
remoteid varchar(36) NOT NULL,
cwd varchar(300) NOT NULL,
PRIMARY KEY (sessionid, windowid, remotename)
sessionscope boolean NOT NULL,
state json NOT NULL
);
CREATE TABLE line (
@ -55,7 +56,9 @@ CREATE TABLE remote (
CREATE TABLE session_cmd (
sessionid varchar(36) NOT NULL,
cmdid varchar(36) NOT NULL,
rsid varchar(36) NOT NULL,
remoteid varchar(36) NOT NULL,
remotestate json NOT NULL,
status varchar(10) NOT NULL,
startts bigint NOT NULL,
pid int NOT NULL,

70
pkg/remote/remote.go
View File

@ -6,12 +6,14 @@ import (
"io"
"os"
"os/exec"
"strings"
"sync"
"time"
"github.com/scripthaus-dev/mshell/pkg/base"
"github.com/scripthaus-dev/mshell/pkg/packet"
"github.com/scripthaus-dev/mshell/pkg/shexec"
"github.com/scripthaus-dev/sh2-server/pkg/scpacket"
"github.com/scripthaus-dev/sh2-server/pkg/sstore"
)
@ -29,15 +31,15 @@ var GlobalStore *Store
type Store struct {
Lock *sync.Mutex
Map map[string]*MShellProc
Map map[string]*MShellProc // key=remoteid
}
type RemoteState struct {
RemoteType string `json:"remotetype"`
RemoteId string `json:"remoteid"`
RemoteName string `json:"remotename"`
Status string `json:"status"`
Cwd string `json:"cwd"`
RemoteType string `json:"remotetype"`
RemoteId string `json:"remoteid"`
RemoteName string `json:"remotename"`
Status string `json:"status"`
DefaultState *sstore.RemoteState `json:"defaultstate"`
}
type MShellProc struct {
@ -57,8 +59,8 @@ type MShellProc struct {
}
type RpcEntry struct {
PacketId string
RespCh chan packet.RpcPacketType
ReqId string
RespCh chan packet.RpcResponsePacketType
}
func LoadRemotes(ctx context.Context) error {
@ -72,7 +74,7 @@ func LoadRemotes(ctx context.Context) error {
}
for _, remote := range allRemotes {
msh := MakeMShell(remote)
GlobalStore.Map[remote.RemoteName] = msh
GlobalStore.Map[remote.RemoteId] = msh
if remote.AutoConnect {
go msh.Launch()
}
@ -80,10 +82,21 @@ func LoadRemotes(ctx context.Context) error {
return nil
}
func GetRemote(name string) *MShellProc {
func GetRemoteByName(name string) *MShellProc {
GlobalStore.Lock.Lock()
defer GlobalStore.Lock.Unlock()
return GlobalStore.Map[name]
for _, msh := range GlobalStore.Map {
if msh.Remote.RemoteName == name {
return msh
}
}
return nil
}
func GetRemoteById(remoteId string) *MShellProc {
GlobalStore.Lock.Lock()
defer GlobalStore.Lock.Unlock()
return GlobalStore.Map[remoteId]
}
func GetAllRemoteState() []RemoteState {
@ -99,7 +112,7 @@ func GetAllRemoteState() []RemoteState {
Status: proc.Status,
}
if proc.InitPk != nil {
state.Cwd = proc.InitPk.HomeDir
state.DefaultState = &sstore.RemoteState{Cwd: proc.InitPk.HomeDir}
}
rtn = append(rtn, state)
}
@ -177,17 +190,37 @@ func (msh *MShellProc) IsConnected() bool {
return msh.Status == StatusConnected
}
func (runner *MShellProc) PacketRpc(pk packet.RpcPacketType, timeout time.Duration) (packet.RpcPacketType, error) {
func RunCommand(pk *scpacket.FeCommandPacketType, cmdId string) error {
msh := GetRemoteById(pk.RemoteState.RemoteId)
if msh == nil {
return fmt.Errorf("no remote id=%s found", pk.RemoteState.RemoteId)
}
if !msh.IsConnected() {
return fmt.Errorf("remote '%s' is not connected", msh.Remote.RemoteName)
}
runPacket := packet.MakeRunPacket()
runPacket.CK = base.MakeCommandKey(pk.SessionId, cmdId)
runPacket.Cwd = pk.RemoteState.Cwd
runPacket.Env = nil
runPacket.Command = strings.TrimSpace(pk.CmdStr)
fmt.Printf("run-packet %v\n", runPacket)
go func() {
msh.Input.SendPacket(runPacket)
}()
return nil
}
func (runner *MShellProc) PacketRpc(pk packet.RpcPacketType, timeout time.Duration) (packet.RpcResponsePacketType, error) {
if !runner.IsConnected() {
return nil, fmt.Errorf("runner is not connected")
}
if pk == nil {
return nil, fmt.Errorf("PacketRpc passed nil packet")
}
id := pk.GetPacketId()
respCh := make(chan packet.RpcPacketType)
id := pk.GetReqId()
respCh := make(chan packet.RpcResponsePacketType)
runner.WithLock(func() {
runner.RpcMap[id] = &RpcEntry{PacketId: id, RespCh: respCh}
runner.RpcMap[id] = &RpcEntry{ReqId: id, RespCh: respCh}
})
defer runner.WithLock(func() {
delete(runner.RpcMap, id)
@ -217,8 +250,9 @@ func (runner *MShellProc) ProcessPackets() {
}
})
for pk := range runner.Output.MainCh {
if rpcPk, ok := pk.(packet.RpcPacketType); ok {
rpcId := rpcPk.GetPacketId()
fmt.Printf("MSH> %s\n", packet.AsString(pk))
if rpcPk, ok := pk.(packet.RpcResponsePacketType); ok {
rpcId := rpcPk.GetResponseId()
runner.WithLock(func() {
entry := runner.RpcMap[rpcId]
if entry == nil {

4
pkg/sstore/dbops.go
View File

@ -108,7 +108,7 @@ func GetSessionById(ctx context.Context, id string) (*SessionType, error) {
rtnSession = &session
query = `SELECT sessionid, windowid, name, curremote, version FROM window WHERE sessionid = ?`
tx.SelectWrap(&session.Windows, query, session.SessionId)
query = `SELECT * FROM session_remote WHERE sessionid = ?`
query = `SELECT * FROM remote_instance WHERE sessionid = ?`
tx.SelectWrap(&session.Remotes, query, session.SessionId)
return nil
})
@ -149,7 +149,7 @@ func GetWindowLines(ctx context.Context, sessionId string, windowId string) ([]*
return lines, nil
}
// also creates window, and sessionremote
// also creates window
func InsertSessionWithName(ctx context.Context, sessionName string) error {
if sessionName == "" {
return fmt.Errorf("invalid session name '%s'", sessionName)

71
pkg/sstore/sstore.go
View File

@ -2,6 +2,9 @@ package sstore
import (
"context"
"database/sql/driver"
"encoding/json"
"fmt"
"log"
"path"
"sync"
@ -48,11 +51,11 @@ func GetDB() (*sqlx.DB, error) {
}
type SessionType struct {
SessionId string `json:"sessionid"`
Name string `json:"name"`
Windows []*WindowType `json:"windows"`
Cmds []*CmdType `json:"cmds"`
Remotes []*SessionRemote `json:"remotes"`
SessionId string `json:"sessionid"`
Name string `json:"name"`
Windows []*WindowType `json:"windows"`
Cmds []*CmdType `json:"cmds"`
Remotes []*RemoteInstance `json:"remotes"`
}
type WindowType struct {
@ -64,12 +67,36 @@ type WindowType struct {
Version int `json:"version"`
}
type SessionRemote struct {
SessionId string `json:"sessionid"`
WindowId string `json:"windowid"`
RemoteId string `json"remoteid"`
RemoteName string `json:"name"`
Cwd string `json:"cwd"`
type RemoteState struct {
Cwd string `json:"cwd"`
}
func (s *RemoteState) Scan(val interface{}) error {
if strVal, ok := val.(string); ok {
if strVal == "" {
return nil
}
err := json.Unmarshal([]byte(strVal), s)
if err != nil {
return err
}
return nil
}
return fmt.Errorf("cannot scan '%T' into RemoteState", val)
}
func (s *RemoteState) Value() (driver.Value, error) {
return json.Marshal(s)
}
type RemoteInstance struct {
RIId string `json:"riid"`
Name string `json:"name"`
SessionId string `json:"sessionid"`
WindowId string `json:"windowid"`
RemoteId string `json"remoteid"`
SessionScope bool `json:"sessionscope"`
State RemoteState `json:"state"`
}
type LineType struct {
@ -101,16 +128,18 @@ type RemoteType struct {
}
type CmdType struct {
RowId int64 `json:"rowid"`
SessionId string `json:"sessionid"`
CmdId string `json:"cmdid"`
RemoteId string `json:"remoteid"`
Status string `json:"status"`
StartTs int64 `json:"startts"`
DoneTs int64 `json:"donets"`
Pid int `json:"pid"`
RunnerPid int `json:"runnerpid"`
ExitCode int `json:"exitcode"`
RowId int64 `json:"rowid"`
SessionId string `json:"sessionid"`
CmdId string `json:"cmdid"`
RSId string `json:"rsid"`
RemoteId string `json:"remoteid"`
RemoteState string `json:"remotestate"`
Status string `json:"status"`
StartTs int64 `json:"startts"`
DoneTs int64 `json:"donets"`
Pid int `json:"pid"`
RunnerPid int `json:"runnerpid"`
ExitCode int `json:"exitcode"`
RunOut packet.PacketType `json:"runout"`
}