627 lines
19 KiB
Go
627 lines
19 KiB
Go
package app
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"net"
|
||
"net/http"
|
||
"net/url"
|
||
"os"
|
||
"path/filepath"
|
||
"sort"
|
||
"strings"
|
||
"time"
|
||
)
|
||
|
||
type NodeView struct {
|
||
ID string `json:"id"`
|
||
Name string `json:"name"`
|
||
SubscriptionID string `json:"subscriptionID"`
|
||
SubscriptionName string `json:"subscriptionName"`
|
||
Type string `json:"type"`
|
||
UDP bool `json:"udp"`
|
||
Delay *int `json:"delay"`
|
||
Alive *bool `json:"alive"`
|
||
Selected bool `json:"selected"`
|
||
}
|
||
|
||
func (a *App) api() http.Handler {
|
||
mux := http.NewServeMux()
|
||
mux.HandleFunc("GET /api/v1/status", a.status)
|
||
mux.HandleFunc("GET /api/v1/subscriptions", a.subscriptions)
|
||
mux.HandleFunc("POST /api/v1/subscriptions", a.saveSubscription)
|
||
mux.HandleFunc("PUT /api/v1/subscriptions/{id}", a.saveSubscription)
|
||
mux.HandleFunc("DELETE /api/v1/subscriptions/{id}", a.deleteSubscription)
|
||
mux.HandleFunc("POST /api/v1/subscriptions/{id}/refresh", a.refreshSubscription)
|
||
mux.HandleFunc("GET /api/v1/nodes", a.nodes)
|
||
mux.HandleFunc("POST /api/v1/nodes/select", a.selectNode)
|
||
mux.HandleFunc("POST /api/v1/nodes/test", a.testNodes)
|
||
mux.HandleFunc("GET /api/v1/rules", a.rules)
|
||
mux.HandleFunc("POST /api/v1/rules", a.saveRule)
|
||
mux.HandleFunc("PUT /api/v1/rules/{id}", a.saveRule)
|
||
mux.HandleFunc("DELETE /api/v1/rules/{id}", a.deleteRule)
|
||
mux.HandleFunc("GET /api/v1/settings", func(w http.ResponseWriter, r *http.Request) { jsonResponse(w, 200, a.Snapshot().Settings) })
|
||
mux.HandleFunc("PUT /api/v1/settings", a.settings)
|
||
mux.HandleFunc("POST /api/v1/system/service", a.service)
|
||
mux.HandleFunc("GET /api/v1/system/network", a.network)
|
||
mux.HandleFunc("GET /api/v1/system/updates", a.coreUpdates)
|
||
mux.HandleFunc("GET /api/v1/system/connectivity", a.connectivitySites)
|
||
mux.HandleFunc("POST /api/v1/system/connectivity/test", a.testConnectivity)
|
||
mux.HandleFunc("GET /api/v1/system/backup", a.backup)
|
||
mux.HandleFunc("POST /api/v1/system/restore", a.restore)
|
||
mux.HandleFunc("GET /api/v1/connections", a.connections)
|
||
mux.HandleFunc("DELETE /api/v1/connections/{id}", a.closeConnection)
|
||
mux.HandleFunc("DELETE /api/v1/connections", a.closeConnection)
|
||
mux.HandleFunc("GET /api/v1/events", a.events)
|
||
mux.HandleFunc("GET /api/v1/logs", a.logs)
|
||
return mux
|
||
}
|
||
|
||
func (a *App) status(w http.ResponseWriter, r *http.Request) {
|
||
jsonResponse(w, 200, a.getStatus(r.Context()))
|
||
}
|
||
func (a *App) getStatus(ctx context.Context) map[string]any {
|
||
state := a.Snapshot()
|
||
ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
|
||
defer cancel()
|
||
var version map[string]any
|
||
err := a.core.Request(ctx, "GET", "/version", nil, &version)
|
||
count := 0
|
||
for _, sub := range state.Subscriptions {
|
||
if sub.Enabled {
|
||
count += len(sub.Nodes)
|
||
}
|
||
}
|
||
a.mu.RLock()
|
||
configError := a.lastError
|
||
a.mu.RUnlock()
|
||
result := map[string]any{"coreOnline": err == nil, "coreInstalled": a.core.Binary != "", "version": "", "uptimeSeconds": int(time.Since(a.started).Seconds()), "nodeCount": count, "subscriptionCount": len(state.Subscriptions), "settings": state.Settings, "configError": configError, "actualNode": "", "selectedAvailable": false}
|
||
_, tunErr := net.InterfaceByName("cw-tun")
|
||
result["tunActive"] = err == nil && state.Settings.TunEnabled && tunErr == nil
|
||
if err == nil {
|
||
result["version"] = version["version"]
|
||
var group map[string]any
|
||
if a.core.Request(ctx, "GET", "/proxies/"+ProxyGroup, nil, &group) == nil {
|
||
actual, _ := group["now"].(string)
|
||
if actual == AutoGroup {
|
||
var auto map[string]any
|
||
if a.core.Request(ctx, "GET", "/proxies/"+AutoGroup, nil, &auto) == nil {
|
||
actual, _ = auto["now"].(string)
|
||
}
|
||
}
|
||
result["actualNode"] = actual
|
||
result["selectedAvailable"] = actual != "" && actual != "REJECT"
|
||
}
|
||
}
|
||
if err != nil {
|
||
result["coreError"] = err.Error()
|
||
}
|
||
return result
|
||
}
|
||
|
||
func (a *App) subscriptions(w http.ResponseWriter, r *http.Request) {
|
||
items := []map[string]any{}
|
||
for _, s := range a.Snapshot().Subscriptions {
|
||
a.refreshMu.Lock()
|
||
refreshing := a.refreshing[s.ID]
|
||
a.refreshMu.Unlock()
|
||
items = append(items, map[string]any{"id": s.ID, "name": s.Name, "url": s.URL, "enabled": s.Enabled, "intervalHours": s.IntervalHours, "updatedAt": s.UpdatedAt, "attemptedAt": s.AttemptedAt, "lastError": s.LastError, "usage": s.Usage, "nodeCount": len(s.Nodes), "refreshing": refreshing})
|
||
}
|
||
jsonResponse(w, 200, items)
|
||
}
|
||
|
||
func (a *App) saveSubscription(w http.ResponseWriter, r *http.Request) {
|
||
var input struct {
|
||
Name string `json:"name"`
|
||
URL string `json:"url"`
|
||
Enabled bool `json:"enabled"`
|
||
IntervalHours int `json:"intervalHours"`
|
||
}
|
||
if err := decode(w, r, &input); err != nil {
|
||
writeError(w, 400, err.Error())
|
||
return
|
||
}
|
||
id := r.PathValue("id")
|
||
newSub := id == ""
|
||
if newSub {
|
||
id = randomID()
|
||
}
|
||
sub := Subscription{ID: id, Name: strings.TrimSpace(input.Name), URL: strings.TrimSpace(input.URL), Enabled: input.Enabled, IntervalHours: input.IntervalHours, Nodes: []map[string]any{}}
|
||
if err := validateSubscription(sub); err != nil {
|
||
writeError(w, 400, err.Error())
|
||
return
|
||
}
|
||
if newSub {
|
||
// Empty metadata can be saved before installing the core; importing nodes needs validation.
|
||
a.mu.Lock()
|
||
defer a.mu.Unlock()
|
||
candidate := cloneState(a.state)
|
||
if len(candidate.Subscriptions) >= 50 {
|
||
writeError(w, 400, "最多允许 50 个订阅")
|
||
return
|
||
}
|
||
candidate.Subscriptions = append(candidate.Subscriptions, sub)
|
||
if err := a.store.Save(candidate); err != nil {
|
||
writeError(w, 500, "保存失败")
|
||
return
|
||
}
|
||
a.state = candidate
|
||
} else {
|
||
err := a.change(r.Context(), func(s *State) error {
|
||
for i, old := range s.Subscriptions {
|
||
if old.ID == id {
|
||
sub.Nodes = old.Nodes
|
||
sub.UpdatedAt = old.UpdatedAt
|
||
sub.AttemptedAt = old.AttemptedAt
|
||
sub.LastError = old.LastError
|
||
sub.Usage = old.Usage
|
||
s.Subscriptions[i] = sub
|
||
return nil
|
||
}
|
||
}
|
||
return errors.New("订阅不存在")
|
||
})
|
||
if err != nil {
|
||
writeError(w, 400, err.Error())
|
||
return
|
||
}
|
||
}
|
||
jsonResponse(w, 200, map[string]any{"id": id, "ok": true})
|
||
}
|
||
|
||
func (a *App) deleteSubscription(w http.ResponseWriter, r *http.Request) {
|
||
id := r.PathValue("id")
|
||
err := a.change(r.Context(), func(state *State) error {
|
||
for i, sub := range state.Subscriptions {
|
||
if sub.ID == id {
|
||
state.Subscriptions = append(state.Subscriptions[:i], state.Subscriptions[i+1:]...)
|
||
return nil
|
||
}
|
||
}
|
||
return errors.New("订阅不存在")
|
||
})
|
||
if err != nil {
|
||
writeError(w, 400, err.Error())
|
||
return
|
||
}
|
||
jsonResponse(w, 200, map[string]bool{"ok": true})
|
||
}
|
||
func (a *App) refreshSubscription(w http.ResponseWriter, r *http.Request) {
|
||
if err := a.Refresh(r.Context(), r.PathValue("id")); err != nil {
|
||
writeError(w, 400, err.Error())
|
||
return
|
||
}
|
||
jsonResponse(w, 200, map[string]bool{"ok": true})
|
||
}
|
||
|
||
func nodeViews(state State, live map[string]map[string]any) []NodeView {
|
||
views := []NodeView{}
|
||
for _, sub := range state.Subscriptions {
|
||
if !sub.Enabled {
|
||
continue
|
||
}
|
||
for _, node := range sub.Nodes {
|
||
id, _ := node["name"].(string)
|
||
parts := strings.SplitN(id, "/", 3)
|
||
name := id
|
||
if len(parts) == 3 {
|
||
name = parts[2]
|
||
}
|
||
kind, _ := node["type"].(string)
|
||
udp, _ := node["udp"].(bool)
|
||
view := NodeView{ID: id, Name: name, SubscriptionID: sub.ID, SubscriptionName: sub.Name, Type: kind, UDP: udp, Selected: state.Settings.SelectedNode == id && state.Settings.SelectionMode == "manual"}
|
||
if info, ok := live[id]; ok {
|
||
if u, ok := info["udp"].(bool); ok {
|
||
view.UDP = u
|
||
}
|
||
if history, ok := info["history"].([]any); ok && len(history) > 0 {
|
||
if h, ok := history[len(history)-1].(map[string]any); ok {
|
||
if d, ok := h["delay"].(float64); ok {
|
||
delay := int(d)
|
||
view.Delay = &delay
|
||
alive := delay > 0
|
||
view.Alive = &alive
|
||
}
|
||
}
|
||
}
|
||
}
|
||
views = append(views, view)
|
||
}
|
||
}
|
||
return views
|
||
}
|
||
|
||
func (a *App) nodes(w http.ResponseWriter, r *http.Request) {
|
||
live := struct {
|
||
Proxies map[string]map[string]any `json:"proxies"`
|
||
}{Proxies: map[string]map[string]any{}}
|
||
_ = a.core.Request(r.Context(), "GET", "/proxies", nil, &live)
|
||
jsonResponse(w, 200, nodeViews(a.Snapshot(), live.Proxies))
|
||
}
|
||
func (a *App) selectNode(w http.ResponseWriter, r *http.Request) {
|
||
var input struct {
|
||
ID string `json:"id"`
|
||
Mode string `json:"mode"`
|
||
}
|
||
if err := decode(w, r, &input); err != nil {
|
||
writeError(w, 400, err.Error())
|
||
return
|
||
}
|
||
err := a.change(r.Context(), func(s *State) error {
|
||
if input.Mode == "auto" {
|
||
s.Settings.SelectionMode = "auto"
|
||
return nil
|
||
}
|
||
if input.Mode != "manual" {
|
||
return errors.New("无效的选择模式")
|
||
}
|
||
for _, node := range nodeViews(*s, nil) {
|
||
if node.ID == input.ID {
|
||
s.Settings.SelectionMode = "manual"
|
||
s.Settings.SelectedNode = input.ID
|
||
return nil
|
||
}
|
||
}
|
||
return errors.New("节点不存在或订阅未启用")
|
||
})
|
||
if err != nil {
|
||
writeError(w, 400, err.Error())
|
||
return
|
||
}
|
||
jsonResponse(w, 200, map[string]bool{"ok": true})
|
||
}
|
||
func (a *App) testNodes(w http.ResponseWriter, r *http.Request) {
|
||
var input struct {
|
||
IDs []string `json:"ids"`
|
||
}
|
||
if err := decode(w, r, &input); err != nil {
|
||
writeError(w, 400, err.Error())
|
||
return
|
||
}
|
||
state := a.Snapshot()
|
||
known := map[string]bool{}
|
||
for _, n := range nodeViews(state, nil) {
|
||
known[n.ID] = true
|
||
}
|
||
if len(input.IDs) == 0 || len(input.IDs) > 500 {
|
||
writeError(w, 400, "一次可测试 1–500 个节点")
|
||
return
|
||
}
|
||
for _, id := range input.IDs {
|
||
if !known[id] {
|
||
writeError(w, 400, "包含未知节点")
|
||
return
|
||
}
|
||
}
|
||
results := map[string]any{}
|
||
for start := 0; start < len(input.IDs); start += 8 {
|
||
end := start + 8
|
||
if end > len(input.IDs) {
|
||
end = len(input.IDs)
|
||
}
|
||
type outcome struct {
|
||
ID string
|
||
Delay int
|
||
Err string
|
||
}
|
||
ch := make(chan outcome, end-start)
|
||
for _, id := range input.IDs[start:end] {
|
||
go func(id string) {
|
||
ctx, cancel := context.WithTimeout(r.Context(), time.Duration(state.Settings.TestTimeoutMS+2000)*time.Millisecond)
|
||
defer cancel()
|
||
var response struct {
|
||
Delay int `json:"delay"`
|
||
}
|
||
path := "/proxies/" + url.PathEscape(id) + "/delay?url=" + url.QueryEscape(state.Settings.TestURL) + fmt.Sprintf("&timeout=%d", state.Settings.TestTimeoutMS)
|
||
err := a.core.Request(ctx, "GET", path, nil, &response)
|
||
message := ""
|
||
if err != nil {
|
||
message = err.Error()
|
||
}
|
||
ch <- outcome{id, response.Delay, message}
|
||
}(id)
|
||
}
|
||
for i := start; i < end; i++ {
|
||
v := <-ch
|
||
results[v.ID] = map[string]any{"delay": v.Delay, "error": v.Err}
|
||
}
|
||
if r.Context().Err() != nil {
|
||
return
|
||
}
|
||
}
|
||
jsonResponse(w, 200, results)
|
||
}
|
||
|
||
func (a *App) rules(w http.ResponseWriter, r *http.Request) {
|
||
jsonResponse(w, 200, map[string]any{"custom": a.Snapshot().Rules, "defaults": []map[string]string{{"match": "内网、回环、链路本地及组播地址", "action": "DIRECT"}, {"match": "自定义规则（按列表顺序）", "action": "CUSTOM"}, {"match": "大陆域名 / IP 规则库", "action": "DIRECT"}, {"match": "其余目标", "action": "PROXY"}}})
|
||
}
|
||
func (a *App) saveRule(w http.ResponseWriter, r *http.Request) {
|
||
var input struct {
|
||
Type string `json:"type"`
|
||
Value string `json:"value"`
|
||
Action string `json:"action"`
|
||
Enabled bool `json:"enabled"`
|
||
}
|
||
if err := decode(w, r, &input); err != nil {
|
||
writeError(w, 400, err.Error())
|
||
return
|
||
}
|
||
id := r.PathValue("id")
|
||
newRule := id == ""
|
||
if newRule {
|
||
id = randomID()
|
||
}
|
||
rule := Rule{ID: id, Type: input.Type, Value: strings.ToLower(strings.TrimSpace(input.Value)), Action: input.Action, Enabled: input.Enabled}
|
||
err := a.change(r.Context(), func(s *State) error {
|
||
if newRule {
|
||
s.Rules = append(s.Rules, rule)
|
||
return nil
|
||
}
|
||
for i, r := range s.Rules {
|
||
if r.ID == id {
|
||
s.Rules[i] = rule
|
||
return nil
|
||
}
|
||
}
|
||
return errors.New("规则不存在")
|
||
})
|
||
if err != nil {
|
||
writeError(w, 400, err.Error())
|
||
return
|
||
}
|
||
jsonResponse(w, 200, map[string]string{"id": id})
|
||
}
|
||
func (a *App) deleteRule(w http.ResponseWriter, r *http.Request) {
|
||
err := a.change(r.Context(), func(s *State) error {
|
||
for i, rule := range s.Rules {
|
||
if rule.ID == r.PathValue("id") {
|
||
s.Rules = append(s.Rules[:i], s.Rules[i+1:]...)
|
||
return nil
|
||
}
|
||
}
|
||
return errors.New("规则不存在")
|
||
})
|
||
if err != nil {
|
||
writeError(w, 400, err.Error())
|
||
return
|
||
}
|
||
jsonResponse(w, 200, map[string]bool{"ok": true})
|
||
}
|
||
|
||
func (a *App) settings(w http.ResponseWriter, r *http.Request) {
|
||
var settings Settings
|
||
if err := decode(w, r, &settings); err != nil {
|
||
writeError(w, 400, err.Error())
|
||
return
|
||
}
|
||
if settings.TunEnabled && !interfaceExists(settings.Interface, settings.LANIP) {
|
||
writeError(w, 400, "VM 地址与网卡不匹配；请在正式 VM 上先配置桥接网卡和固定 IP")
|
||
return
|
||
}
|
||
err := a.change(r.Context(), func(s *State) error { s.Settings = settings; return nil })
|
||
if err != nil {
|
||
writeError(w, 400, err.Error())
|
||
return
|
||
}
|
||
jsonResponse(w, 200, settings)
|
||
}
|
||
func (a *App) service(w http.ResponseWriter, r *http.Request) {
|
||
var input struct {
|
||
Action string `json:"action"`
|
||
}
|
||
if err := decode(w, r, &input); err != nil {
|
||
writeError(w, 400, err.Error())
|
||
return
|
||
}
|
||
if input.Action == "start" || input.Action == "restart" {
|
||
a.mu.Lock()
|
||
config, _ := compileConfig(a.state, a.core.Secret)
|
||
err := a.core.Apply(r.Context(), config)
|
||
a.mu.Unlock()
|
||
if err != nil {
|
||
writeError(w, 400, err.Error())
|
||
return
|
||
}
|
||
}
|
||
if err := a.core.Service(r.Context(), input.Action); err != nil {
|
||
writeError(w, 400, err.Error())
|
||
return
|
||
}
|
||
if input.Action != "stop" {
|
||
if err := a.core.Select(r.Context(), a.Snapshot()); err != nil {
|
||
writeError(w, 400, err.Error())
|
||
return
|
||
}
|
||
}
|
||
jsonResponse(w, 200, map[string]bool{"ok": true})
|
||
}
|
||
func (a *App) connections(w http.ResponseWriter, r *http.Request) {
|
||
var result map[string]any
|
||
if err := a.core.Request(r.Context(), "GET", "/connections", nil, &result); err != nil {
|
||
writeError(w, 503, err.Error())
|
||
return
|
||
}
|
||
jsonResponse(w, 200, result)
|
||
}
|
||
func (a *App) closeConnection(w http.ResponseWriter, r *http.Request) {
|
||
path := "/connections"
|
||
if id := r.PathValue("id"); id != "" {
|
||
path += "/" + url.PathEscape(id)
|
||
}
|
||
if err := a.core.Request(r.Context(), "DELETE", path, nil, nil); err != nil {
|
||
writeError(w, 503, err.Error())
|
||
return
|
||
}
|
||
jsonResponse(w, 200, map[string]bool{"ok": true})
|
||
}
|
||
func (a *App) backup(w http.ResponseWriter, r *http.Request) {
|
||
w.Header().Set("Content-Disposition", `attachment; filename="clashweb-backup.json"`)
|
||
jsonResponse(w, 200, Backup{1, now(), a.Snapshot()})
|
||
}
|
||
func (a *App) restore(w http.ResponseWriter, r *http.Request) {
|
||
var backup Backup
|
||
if err := decode(w, r, &backup); err != nil {
|
||
writeError(w, 400, err.Error())
|
||
return
|
||
}
|
||
if backup.Version != 1 {
|
||
writeError(w, 400, "不支持此备份版本")
|
||
return
|
||
}
|
||
if backup.State.Settings.TunEnabled && !interfaceExists(backup.State.Settings.Interface, backup.State.Settings.LANIP) {
|
||
writeError(w, 400, "备份中的网卡和 VM 地址与当前机器不匹配；请先修改备份中的网络设置或关闭 tunEnabled")
|
||
return
|
||
}
|
||
err := a.change(r.Context(), func(s *State) error { *s = backup.State; return nil })
|
||
if err != nil {
|
||
writeError(w, 400, err.Error())
|
||
return
|
||
}
|
||
jsonResponse(w, 200, map[string]bool{"ok": true})
|
||
}
|
||
func (a *App) network(w http.ResponseWriter, r *http.Request) {
|
||
interfaces := []map[string]any{}
|
||
list, _ := net.Interfaces()
|
||
for _, i := range list {
|
||
if i.Flags&net.FlagLoopback != 0 {
|
||
continue
|
||
}
|
||
ips := []string{}
|
||
addresses, _ := i.Addrs()
|
||
for _, addr := range addresses {
|
||
ip, _, err := net.ParseCIDR(addr.String())
|
||
if err == nil && ip.To4() != nil {
|
||
ips = append(ips, addr.String())
|
||
}
|
||
}
|
||
interfaces = append(interfaces, map[string]any{"name": i.Name, "addresses": ips, "up": i.Flags&net.FlagUp != 0})
|
||
}
|
||
sort.Slice(interfaces, func(i, j int) bool { return interfaces[i]["name"].(string) < interfaces[j]["name"].(string) })
|
||
forward, _ := os.ReadFile("/proc/sys/net/ipv4/ip_forward")
|
||
_, tunErr := os.Stat("/dev/net/tun")
|
||
_, domainErr := os.Stat(filepath.Join(a.core.Dir, "rules/cn-domain.mrs"))
|
||
_, ipErr := os.Stat(filepath.Join(a.core.Dir, "rules/cn-ip.mrs"))
|
||
jsonResponse(w, 200, map[string]any{"interfaces": interfaces, "ipForward": strings.TrimSpace(string(forward)) == "1", "tunDevice": tunErr == nil, "rulesInstalled": domainErr == nil && ipErr == nil, "serviceControl": a.core.Helper != "", "dhcpManaged": false, "guidance": "当前光猫 DHCP 无法关闭。请手动将测试设备的 IPv4 网关和 DNS 设为 PVE VM 的固定地址；不要在 VM 启动第二个 DHCP 服务。"})
|
||
}
|
||
|
||
func streamHeaders(w http.ResponseWriter) error {
|
||
if _, ok := w.(http.Flusher); !ok {
|
||
return errors.New("不支持流式响应")
|
||
}
|
||
w.Header().Set("Content-Type", "text/event-stream")
|
||
w.Header().Set("Cache-Control", "no-cache")
|
||
w.Header().Set("X-Accel-Buffering", "no")
|
||
return nil
|
||
}
|
||
func sendEvent(w http.ResponseWriter, event string, value any) error {
|
||
b, err := json.Marshal(value)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
_ = http.NewResponseController(w).SetWriteDeadline(time.Now().Add(10 * time.Second))
|
||
if _, err := fmt.Fprintf(w, "event: %s\ndata: %s\n\n", event, b); err != nil {
|
||
return err
|
||
}
|
||
w.(http.Flusher).Flush()
|
||
return nil
|
||
}
|
||
func (a *App) events(w http.ResponseWriter, r *http.Request) {
|
||
if err := streamHeaders(w); err != nil {
|
||
writeError(w, 500, err.Error())
|
||
return
|
||
}
|
||
ticker := time.NewTicker(2 * time.Second)
|
||
defer ticker.Stop()
|
||
var prevUp, prevDown float64
|
||
last := time.Now()
|
||
initialized := false
|
||
for {
|
||
if _, _, err := a.session(r); err != nil {
|
||
_ = sendEvent(w, "expired", nil)
|
||
return
|
||
}
|
||
var connections struct {
|
||
Upload float64 `json:"uploadTotal"`
|
||
Download float64 `json:"downloadTotal"`
|
||
Connections []json.RawMessage `json:"connections"`
|
||
}
|
||
ctx, cancel := context.WithTimeout(r.Context(), 3*time.Second)
|
||
err := a.core.Request(ctx, "GET", "/connections", nil, &connections)
|
||
cancel()
|
||
current := time.Now()
|
||
seconds := current.Sub(last).Seconds()
|
||
up, down := float64(0), float64(0)
|
||
if err == nil && initialized && seconds > 0 {
|
||
up = (connections.Upload - prevUp) / seconds
|
||
down = (connections.Download - prevDown) / seconds
|
||
if up < 0 {
|
||
up = 0
|
||
}
|
||
if down < 0 {
|
||
down = 0
|
||
}
|
||
}
|
||
if err == nil {
|
||
prevUp = connections.Upload
|
||
prevDown = connections.Download
|
||
}
|
||
initialized = err == nil
|
||
last = current
|
||
if sendEvent(w, "metrics", map[string]any{"up": up, "down": down, "upTotal": connections.Upload, "downTotal": connections.Download, "connections": len(connections.Connections), "coreOnline": err == nil, "time": now()}) != nil {
|
||
return
|
||
}
|
||
select {
|
||
case <-r.Context().Done():
|
||
return
|
||
case <-ticker.C:
|
||
}
|
||
}
|
||
}
|
||
func (a *App) logs(w http.ResponseWriter, r *http.Request) {
|
||
if err := streamHeaders(w); err != nil {
|
||
writeError(w, 500, err.Error())
|
||
return
|
||
}
|
||
ctx, cancel := context.WithCancel(r.Context())
|
||
defer cancel()
|
||
items := make(chan any, 32)
|
||
finished := make(chan error, 1)
|
||
go func() {
|
||
finished <- a.core.Logs(ctx, func(item any) error {
|
||
select {
|
||
case items <- item:
|
||
return nil
|
||
case <-ctx.Done():
|
||
return ctx.Err()
|
||
}
|
||
})
|
||
}()
|
||
ticker := time.NewTicker(15 * time.Second)
|
||
defer ticker.Stop()
|
||
for {
|
||
select {
|
||
case item := <-items:
|
||
if sendEvent(w, "log", item) != nil {
|
||
return
|
||
}
|
||
case err := <-finished:
|
||
if err != nil {
|
||
_ = sendEvent(w, "disconnected", map[string]string{"error": "内核日志连接已断开，正在等待重连"})
|
||
}
|
||
return
|
||
case <-ticker.C:
|
||
if _, _, err := a.session(r); err != nil {
|
||
_ = sendEvent(w, "expired", nil)
|
||
return
|
||
}
|
||
if sendEvent(w, "heartbeat", nil) != nil {
|
||
return
|
||
}
|
||
case <-ctx.Done():
|
||
return
|
||
}
|
||
}
|
||
}
|