powerpods/goTool/api_accel_stream.go

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
}