fix: add or replace MediaMTX paths idempotently (#54)
This commit is contained in:
@@ -20,6 +20,7 @@ type Source struct {
|
||||
}
|
||||
type PathStatus struct {
|
||||
Name string `json:"name"`
|
||||
Exists bool `json:"exists"`
|
||||
Ready bool `json:"ready"`
|
||||
Readers int `json:"readers"`
|
||||
BytesReceived int64 `json:"bytes_received"`
|
||||
@@ -46,7 +47,15 @@ func (c *HTTPController) Apply(ctx context.Context, source Source) error {
|
||||
}
|
||||
payload := map[string]any{"source": parsed.String(), "sourceOnDemand": true, "rtspTransport": "tcp"}
|
||||
data, _ := json.Marshal(payload)
|
||||
endpoint := c.base + "/v3/config/paths/replace/" + url.PathEscape(source.Path)
|
||||
configured, err := c.configured(ctx, source.Path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
action := "add"
|
||||
if configured {
|
||||
action = "replace"
|
||||
}
|
||||
endpoint := c.base + "/v3/config/paths/" + action + "/" + url.PathEscape(source.Path)
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(data))
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -63,6 +72,39 @@ func (c *HTTPController) Apply(ctx context.Context, source Source) error {
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *HTTPController) configured(ctx context.Context, path string) (bool, error) {
|
||||
endpoint := c.base + "/v3/config/paths/get/" + url.PathEscape(path)
|
||||
deadline := time.Now().Add(5 * time.Second)
|
||||
var res *http.Response
|
||||
for {
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
res, err = c.client.Do(req)
|
||||
if err == nil {
|
||||
break
|
||||
}
|
||||
if !time.Now().Before(deadline) {
|
||||
return false, fmt.Errorf("mediamtx control unavailable: %w", err)
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return false, ctx.Err()
|
||||
case <-time.After(100 * time.Millisecond):
|
||||
}
|
||||
}
|
||||
defer res.Body.Close()
|
||||
io.Copy(io.Discard, io.LimitReader(res.Body, 1<<20))
|
||||
if res.StatusCode == http.StatusNotFound {
|
||||
return false, nil
|
||||
}
|
||||
if res.StatusCode < 200 || res.StatusCode >= 300 {
|
||||
return false, fmt.Errorf("mediamtx config lookup returned %d", res.StatusCode)
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
func (c *HTTPController) Status(ctx context.Context, path string) (PathStatus, error) {
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.base+"/v3/paths/get/"+url.PathEscape(path), nil)
|
||||
if err != nil {
|
||||
@@ -74,7 +116,7 @@ func (c *HTTPController) Status(ctx context.Context, path string) (PathStatus, e
|
||||
}
|
||||
defer res.Body.Close()
|
||||
if res.StatusCode == http.StatusNotFound {
|
||||
return PathStatus{Name: path}, nil
|
||||
return PathStatus{Name: path, Exists: false}, nil
|
||||
}
|
||||
if res.StatusCode < 200 || res.StatusCode >= 300 {
|
||||
return PathStatus{}, fmt.Errorf("mediamtx status returned %d", res.StatusCode)
|
||||
@@ -88,5 +130,6 @@ func (c *HTTPController) Status(ctx context.Context, path string) (PathStatus, e
|
||||
if err := json.NewDecoder(io.LimitReader(res.Body, 1<<20)).Decode(&raw); err != nil {
|
||||
return PathStatus{}, err
|
||||
}
|
||||
return PathStatus{Name: raw.Name, Ready: raw.Ready, Readers: len(raw.Readers), BytesReceived: raw.BytesReceived}, nil
|
||||
return PathStatus{Name: raw.Name, Exists: true, Ready: raw.Ready, Readers: len(raw.Readers), BytesReceived: raw.BytesReceived}, nil
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user