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

send cmd updates for donepk

Browse Source
This commit is contained in:
sawka
2022-08-19 17:14:53 -07:00
parent 03cfabd9b6
commit b2a2b6252d
2 changed files with 31 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/remote/remote.go
+15 -9
View File
@@ -48,9 +48,8 @@ const (
var GlobalStore *Store var GlobalStore *Store
type Store struct { type Store struct {
Lock *sync.Mutex Lock *sync.Mutex
Map map[string]*MShellProc // key=remoteid Map map[string]*MShellProc // key=remoteid
CmdStatusCallback func(ck base.CommandKey, status string)
} }
type RemoteState struct { type RemoteState struct {
@@ -440,13 +439,17 @@ func makeDataAckPacket(ck base.CommandKey, fdNum int, ackLen int, err error) *pa
} }
func (msh *MShellProc) handleCmdDonePacket(donePk *packet.CmdDonePacketType) { func (msh *MShellProc) handleCmdDonePacket(donePk *packet.CmdDonePacketType) {
err := sstore.UpdateCmdDonePk(context.Background(), donePk) update, err := sstore.UpdateCmdDonePk(context.Background(), donePk)
if err != nil { if err != nil {
fmt.Printf("[error] updating cmddone: %v\n", err) fmt.Printf("[error] updating cmddone: %v\n", err)
return return
} }
if GlobalStore.CmdStatusCallback != nil { if update != nil {
GlobalStore.CmdStatusCallback(donePk.CK, sstore.CmdStatusDone) // TODO fix timing issue (this update gets to the FE before run-command returns for short lived commands)
go func() {
time.Sleep(10 * time.Millisecond)
sstore.MainBus.SendUpdate(donePk.CK.GetSessionId(), update)
}()
} }
return return
} }
@@ -461,10 +464,13 @@ func (msh *MShellProc) handleCmdErrorPacket(errPk *packet.CmdErrorPacketType) {
} }
func (msh *MShellProc) notifyHangups_nolock() { func (msh *MShellProc) notifyHangups_nolock() {
if GlobalStore.CmdStatusCallback != nil { for _, ck := range msh.RunningCmds {
for _, ck := range msh.RunningCmds { cmd, err := sstore.GetCmdById(context.Background(), ck.GetSessionId(), ck.GetCmdId())
GlobalStore.CmdStatusCallback(ck, sstore.CmdStatusHangup) if err != nil {
continue
} }
update := sstore.LineUpdate{Cmd: cmd}
sstore.MainBus.SendUpdate(ck.GetSessionId(), update)
} }
msh.RunningCmds = nil msh.RunningCmds = nil
} }
pkg/sstore/dbops.go
+16 -3
View File
@@ -427,15 +427,28 @@ func GetCmdById(ctx context.Context, sessionId string, cmdId string) (*CmdType,
return cmd, nil return cmd, nil
} }
func UpdateCmdDonePk(ctx context.Context, donePk *packet.CmdDonePacketType) error { func UpdateCmdDonePk(ctx context.Context, donePk *packet.CmdDonePacketType) (UpdatePacket, error) {
if donePk == nil || donePk.CK.IsEmpty() { if donePk == nil || donePk.CK.IsEmpty() {
return fmt.Errorf("invalid cmddone packet (no ck)") return nil, fmt.Errorf("invalid cmddone packet (no ck)")
} }
return WithTx(ctx, func(tx *TxWrap) error { var rtnCmd *CmdType
txErr := WithTx(ctx, func(tx *TxWrap) error {
query := `UPDATE cmd SET status = ?, donepk = ? WHERE sessionid = ? AND cmdid = ?` query := `UPDATE cmd SET status = ?, donepk = ? WHERE sessionid = ? AND cmdid = ?`
tx.ExecWrap(query, CmdStatusDone, quickJson(donePk), donePk.CK.GetSessionId(), donePk.CK.GetCmdId()) tx.ExecWrap(query, CmdStatusDone, quickJson(donePk), donePk.CK.GetSessionId(), donePk.CK.GetCmdId())
var err error
rtnCmd, err = GetCmdById(tx.Context(), donePk.CK.GetSessionId(), donePk.CK.GetCmdId())
if err != nil {
return err
}
return nil return nil
}) })
if txErr != nil {
return nil, txErr
}
if rtnCmd == nil {
return nil, fmt.Errorf("cmd data not found for ck[%s]", donePk.CK)
}
return LineUpdate{Cmd: rtnCmd}, nil
} }
func AppendCmdErrorPk(ctx context.Context, errPk *packet.CmdErrorPacketType) error { func AppendCmdErrorPk(ctx context.Context, errPk *packet.CmdErrorPacketType) error {