128 lines
3.0 KiB
Go
128 lines
3.0 KiB
Go
package agent
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"io"
|
|
"net/http"
|
|
|
|
"github.com/x-grk/longpolling-remote-access/internal/proto"
|
|
)
|
|
|
|
// runTask fetches one task, executes it, and posts the result back.
|
|
func (a *Agent) runTask(ctx context.Context, taskID string) {
|
|
env, err := a.fetchTask(ctx, taskID)
|
|
if err != nil {
|
|
return
|
|
}
|
|
var res proto.TaskResult
|
|
switch env.Type {
|
|
case proto.TaskSysUpdate:
|
|
res = a.doUpdate()
|
|
case proto.TaskSysSpeed:
|
|
res = a.doSpeed(ctx, env)
|
|
case proto.TaskFSPut:
|
|
res = a.doFSPut(ctx, env)
|
|
case proto.TaskFSSend:
|
|
res = a.doFSSend(ctx, env)
|
|
case proto.TaskSSHFSSend:
|
|
res = a.doSSHFSSend(ctx, env)
|
|
case proto.TaskSSHFSPut:
|
|
res = a.doSSHFSPut(ctx, env)
|
|
case proto.TaskSSHFSOpen, proto.TaskSSHFSClose, proto.TaskSSHFSOp:
|
|
res = resultOf(a.sshfsTask(ctx, env))
|
|
case proto.TaskFSShipOut:
|
|
res = a.doShipOut(ctx, env)
|
|
case proto.TaskFSShipIn:
|
|
res = a.doShipIn(ctx, env)
|
|
default:
|
|
res = a.execTask(env)
|
|
}
|
|
a.postResult(ctx, taskID, res)
|
|
}
|
|
|
|
func (a *Agent) fetchTask(ctx context.Context, taskID string) (proto.TaskEnvelope, error) {
|
|
var env proto.TaskEnvelope
|
|
req, _ := http.NewRequestWithContext(ctx, http.MethodGet, a.host()+"/api/agent/task/"+taskID, nil)
|
|
a.authHeaders(req)
|
|
resp, err := a.http.Do(req)
|
|
if err != nil {
|
|
return env, err
|
|
}
|
|
defer resp.Body.Close()
|
|
if resp.StatusCode != http.StatusOK {
|
|
return env, errStatus(resp.StatusCode)
|
|
}
|
|
err = json.NewDecoder(resp.Body).Decode(&env)
|
|
return env, err
|
|
}
|
|
|
|
func (a *Agent) postResult(ctx context.Context, taskID string, res proto.TaskResult) {
|
|
body, _ := json.Marshal(res)
|
|
req, _ := http.NewRequestWithContext(ctx, http.MethodPost, a.host()+"/api/agent/task/"+taskID+"/result", bytes.NewReader(body))
|
|
a.authHeaders(req)
|
|
req.Header.Set("Content-Type", "application/json")
|
|
resp, err := a.http.Do(req)
|
|
if err != nil {
|
|
return
|
|
}
|
|
io.Copy(io.Discard, resp.Body)
|
|
resp.Body.Close()
|
|
}
|
|
|
|
// execTask runs a single file-manager or user-management operation natively
|
|
// and returns a result.
|
|
func (a *Agent) execTask(env proto.TaskEnvelope) proto.TaskResult {
|
|
data, err := dispatchTask(env)
|
|
if err != nil {
|
|
return proto.TaskResult{Error: err.Error()}
|
|
}
|
|
raw, err := json.Marshal(data)
|
|
if err != nil {
|
|
return proto.TaskResult{Error: err.Error()}
|
|
}
|
|
return proto.TaskResult{OK: true, Data: raw}
|
|
}
|
|
|
|
type statusErr int
|
|
|
|
func (e statusErr) Error() string { return "server status " + itoa(int(e)) }
|
|
func errStatus(code int) error { return statusErr(code) }
|
|
|
|
func itoa(n int) string { return itoa64(int64(n)) }
|
|
|
|
func itoa64(n int64) string {
|
|
if n == 0 {
|
|
return "0"
|
|
}
|
|
neg := n < 0
|
|
if neg {
|
|
n = -n
|
|
}
|
|
var b [24]byte
|
|
i := len(b)
|
|
for n > 0 {
|
|
i--
|
|
b[i] = byte('0' + n%10)
|
|
n /= 10
|
|
}
|
|
if neg {
|
|
i--
|
|
b[i] = '-'
|
|
}
|
|
return string(b[i:])
|
|
}
|
|
|
|
// resultOf makes a task's result of what it returned.
|
|
func resultOf(data any, err error) proto.TaskResult {
|
|
if err != nil {
|
|
return proto.TaskResult{Error: err.Error()}
|
|
}
|
|
raw, err := json.Marshal(data)
|
|
if err != nil {
|
|
return proto.TaskResult{Error: err.Error()}
|
|
}
|
|
return proto.TaskResult{OK: true, Data: raw}
|
|
}
|