package main import ( "encoding/json" "fmt" "net/http" "strconv" "powerpod/gotool/pb" ) type accelStreamAPIRequest struct { Write bool `json:"write"` Enable bool `json:"enable"` ClientID uint32 `json:"client_id"` AllClients bool `json:"all_clients"` } type accelStreamAPIResponse struct { Enabled bool `json:"enabled"` ClientID uint32 `json:"client_id"` Success bool `json:"success"` SlavesUpdated uint32 `json:"slaves_updated"` Error string `json:"error,omitempty"` } type clientAccelStreamBody struct { Enable bool `json:"enable"` } func mountAccelStreamAPI(mux *http.ServeMux, link *managedSerial, hub *wsHub, ctl *accelStreamCtl) { mux.HandleFunc("GET /api/clients/{clientID}/accel-stream", func(w http.ResponseWriter, r *http.Request) { clientID, err := parsePathClientID(r) if err != nil { writeJSON(w, http.StatusBadRequest, accelStreamAPIResponse{Error: err.Error()}) return } serveAccelStreamGet(w, clientID, link, hub, ctl) }) mux.HandleFunc("PUT /api/clients/{clientID}/accel-stream", func(w http.ResponseWriter, r *http.Request) { clientID, err := parsePathClientID(r) if err != nil { writeJSON(w, http.StatusBadRequest, accelStreamAPIResponse{Error: err.Error()}) return } serveClientAccelStreamPut(w, r, clientID, link, hub, ctl) }) mux.HandleFunc("/api/accel-stream", func(w http.ResponseWriter, r *http.Request) { switch r.Method { case http.MethodGet: serveAccelStreamGetQuery(w, r, link, hub, ctl) case http.MethodPost: serveAccelStreamPost(w, r, link, hub, ctl) default: http.Error(w, "method not allowed", http.StatusMethodNotAllowed) } }) } func parsePathClientID(r *http.Request) (uint32, error) { s := r.PathValue("clientID") if s == "" { return 0, fmt.Errorf("client_id required") } v, err := strconv.ParseUint(s, 10, 32) if err != nil || v == 0 { return 0, fmt.Errorf("invalid client_id") } return uint32(v), nil } func applyAccelStreamClient(link *managedSerial, hub *wsHub, ctl *accelStreamCtl, clientID uint32, enable bool) accelStreamAPIResponse { resp, err := link.AccelStream(&pb.AccelStreamRequest{ Write: true, Enable: enable, ClientId: clientID, }) if err != nil { return accelStreamAPIResponse{ ClientID: clientID, Error: err.Error(), } } out := accelStreamAPIResponse{ Enabled: enable, ClientID: resp.GetClientId(), Success: resp.GetSuccess(), SlavesUpdated: resp.GetSlavesUpdated(), } if resp.GetSuccess() { if ctl != nil { ctl.Set(clientID, enable) } if hub != nil { hub.patchClientAccelStream(clientID, enable) } } else { out.Enabled = resp.GetEnabled() } return out } func serveAccelStreamGet(w http.ResponseWriter, clientID uint32, link *managedSerial, hub *wsHub, ctl *accelStreamCtl) { resp, err := link.AccelStreamPoll(&pb.AccelStreamRequest{ Write: false, ClientId: clientID, }) if err != nil { writeJSON(w, http.StatusServiceUnavailable, accelStreamAPIResponse{ ClientID: clientID, Error: err.Error(), }) return } if ctl != nil { ctl.Set(clientID, resp.GetEnabled()) } writeJSON(w, http.StatusOK, accelStreamAPIResponse{ Enabled: resp.GetEnabled(), ClientID: resp.GetClientId(), Success: resp.GetSuccess(), }) } func serveAccelStreamGetQuery(w http.ResponseWriter, r *http.Request, link *managedSerial, hub *wsHub, ctl *accelStreamCtl) { clientID, err := parseUintQuery(r, "client_id", 0) if err != nil || clientID == 0 { writeJSON(w, http.StatusBadRequest, accelStreamAPIResponse{Error: "client_id required"}) return } serveAccelStreamGet(w, clientID, link, hub, ctl) } func serveClientAccelStreamPut(w http.ResponseWriter, r *http.Request, clientID uint32, link *managedSerial, hub *wsHub, ctl *accelStreamCtl) { var body clientAccelStreamBody if err := json.NewDecoder(r.Body).Decode(&body); err != nil { writeJSON(w, http.StatusBadRequest, accelStreamAPIResponse{Error: "invalid JSON"}) return } out := applyAccelStreamClient(link, hub, ctl, clientID, body.Enable) status := http.StatusOK if out.Error != "" { status = http.StatusServiceUnavailable } else if !out.Success { status = http.StatusServiceUnavailable } writeJSON(w, status, out) } func serveAccelStreamPost(w http.ResponseWriter, r *http.Request, link *managedSerial, hub *wsHub, ctl *accelStreamCtl) { var body accelStreamAPIRequest if err := json.NewDecoder(r.Body).Decode(&body); err != nil { writeJSON(w, http.StatusBadRequest, accelStreamAPIResponse{Error: "invalid JSON"}) return } if body.AllClients { updated, err := applyAccelStreamAll(link, body.Enable) if err != nil { writeJSON(w, http.StatusServiceUnavailable, accelStreamAPIResponse{Error: err.Error()}) return } clients, _ := link.listClientsPoll() for _, c := range clients { if ctl != nil { ctl.Set(c.GetId(), body.Enable) } if hub != nil { hub.patchClientAccelStream(c.GetId(), body.Enable) } } writeJSON(w, http.StatusOK, accelStreamAPIResponse{ Enabled: body.Enable, Success: updated > 0, SlavesUpdated: updated, }) return } if body.ClientID == 0 { writeJSON(w, http.StatusBadRequest, accelStreamAPIResponse{Error: "client_id required"}) return } out := applyAccelStreamClient(link, hub, ctl, body.ClientID, body.Enable) status := http.StatusOK if out.Error != "" || !out.Success { status = http.StatusServiceUnavailable } writeJSON(w, status, out) } func applyAccelStreamAll(link *managedSerial, enable bool) (uint32, error) { clients, err := link.listClients() if err != nil { return 0, err } var updated uint32 for _, c := range clients { resp, err := link.AccelStream(&pb.AccelStreamRequest{ Write: true, Enable: enable, ClientId: c.GetId(), }) if err != nil { continue } if resp.GetSuccess() { updated++ } } if len(clients) == 0 { return 0, fmt.Errorf("no slaves registered") } if updated == 0 { return 0, fmt.Errorf("accel stream not applied to any slave") } return updated, nil }