API remove stream consumers (#12)

HYBIOT · Aug 30, 2024 · 7445123 · 7445123
1 parent ec1d82e
commit 7445123
Show file tree

Hide file tree

Showing 2 changed files with 30 additions and 0 deletions.
diff --git a/internal/streams/custom.go b/internal/streams/custom.go
@@ -56,3 +56,32 @@ func apiStreamsSpeed(w http.ResponseWriter, r *http.Request) {
 
 	http.Error(w, "", http.StatusNotFound)
 }
+
+func apiStreamsRemoveConsumers(w http.ResponseWriter, r *http.Request) {
+	if r.Method != "POST" {
+		http.Error(w, "", http.StatusNotFound)
+		return
+	}
+
+	query := r.URL.Query()
+
+	for _, streamName := range query["name"] {
+		if streamName == "" {
+			continue
+		}
+
+		streamsMu.RLock()
+		stream := Get(streamName)
+		streamsMu.RUnlock()
+
+		if stream == nil {
+			continue
+		}
+
+		for _, consumer := range stream.consumers {
+			stream.RemoveConsumer(consumer)
+		}
+	}
+
+	http.Error(w, "", http.StatusOK)
+}
diff --git a/internal/streams/streams.go b/internal/streams/streams.go
@@ -31,6 +31,7 @@ func Init() {
 
 	// custom
 	api.HandleFunc("api/custom/streams/speed", apiStreamsSpeed)
+	api.HandleFunc("api/custom/streams.removeConsumers", apiStreamsRemoveConsumers)
 
 	if cfg.Publish == nil {
 		return