remove RespIterator and update to use chan idiom

Signed-off-by: allenzhli <[email protected]>

Author: allenzhli

pkg/alertmanager/api.go

@@ -204,11 +204,11 @@ func (am *MultitenantAlertmanager) ListAllConfigs(w http.ResponseWriter, r *http
 	}
 
 	done := make(chan struct{})
-	iter := util.NewRespIter(make(chan []byte))
+	iter := make(chan []byte)
 
 	go func() {
 		util.StreamWriteResponse(w, iter, "text/yaml")
-		done <- struct{}{}
+		close(done)
 	}()
 
 	err = concurrency.ForEachUser(r.Context(), userIDs, fetchConcurrency, func(ctx context.Context, userID string) error {
@@ -229,13 +229,16 @@ func (am *MultitenantAlertmanager) ListAllConfigs(w http.ResponseWriter, r *http
 			return err
 		}
 
-		iter.Put(data)
+		select {
+		case iter <- data:
+		case <-done: // stop early, if sending response has already finished
+		}
 
 		return nil
 	})
 	if err != nil {
 		level.Error(logger).Log("msg", "failed to list all alertmanager configs", "err", err)
 	}
-	iter.Close()
+	close(iter)
 	<-done
 }

pkg/ruler/ruler.go

@@ -830,11 +830,11 @@ func (r *Ruler) ListAllRules(w http.ResponseWriter, req *http.Request) {
 	}
 
 	done := make(chan struct{})
-	iter := util.NewRespIter(make(chan []byte))
+	iter := make(chan []byte)
 
 	go func() {
 		util.StreamWriteResponse(w, iter, "text/yaml")
-		done <- struct{}{}
+		close(done)
 	}()
 
 	err = concurrency.ForEachUser(req.Context(), userIDs, fetchRulesConcurrency, func(ctx context.Context, userID string) error {
@@ -850,13 +850,16 @@ func (r *Ruler) ListAllRules(w http.ResponseWriter, req *http.Request) {
 			return err
 		}
 
-		iter.Put(data)
+		select {
+		case iter <- data:
+		case <-done: // stop early, if sending response has already finished
+		}
 
 		return nil
 	})
 	if err != nil {
 		level.Error(logger).Log("msg", "failed to list all alertmanager configs", "err", err)
 	}
-	iter.Close()
+	close(iter)
 	<-done
 }

pkg/util/http.go

@@ -98,36 +98,8 @@ func RenderHTTPResponse(w http.ResponseWriter, v interface{}, t *template.Templa
 	}
 }
 
-// RespIterator is a iterator for stream http response
-type RespIterator interface {
-	// Close must be called to release resources once the iterator is not
-	// used anymore.
-	Close()
-	// Next returns a channel that will be closed once the iterator is
-	// exhausted.
-	Next() <-chan []byte
-	// Put input a new item to iterator
-	Put(v []byte)
-}
-
-type respIter struct {
-	ch chan []byte
-}
-
-// NewRespIter returns a new streamResp
-func NewRespIter(ch chan []byte) RespIterator {
-	return &respIter{ch: ch}
-}
-func (it respIter) Next() <-chan []byte {
-	return it.ch
-}
-func (it respIter) Close() { close(it.ch) }
-func (it respIter) Put(v []byte) {
-	it.ch <- v
-}
-
 // StreamWriteResponse stream writes data as http response
-func StreamWriteResponse(w http.ResponseWriter, iter RespIterator, contentType string) {
+func StreamWriteResponse(w http.ResponseWriter, iter chan []byte, contentType string) {
 	flusher, ok := w.(http.Flusher)
 	if !ok {
 		http.Error(w, "expected http.ResponseWriter to be an http.Flusher", http.StatusInternalServerError)
@@ -140,7 +112,7 @@ func StreamWriteResponse(w http.ResponseWriter, iter RespIterator, contentType s
 	w.WriteHeader(http.StatusOK)
 	flusher.Flush()
 
-	for m := range iter.Next() {
+	for m := range iter {
 		_, _ = w.Write(m)
 		flusher.Flush()
 	}

pkg/util/http_test.go

@@ -118,17 +118,17 @@ func TestStreamWriteYAMLResponse(t *testing.T) {
 	w := httptest.NewRecorder()
 
 	done := make(chan struct{})
-	iter := util.NewRespIter(make(chan []byte))
+	iter := make(chan []byte)
 	go func() {
 		util.StreamWriteResponse(w, iter, "text/yaml")
-		done <- struct{}{}
+		close(done)
 	}()
 	for k, v := range tt.value {
 		data, err := yaml.Marshal(map[string]*testStruct{k: v})
 		assert.Nil(t, err)
-		iter.Put(data)
+		iter <- data
 	}
-	iter.Close()
+	close(iter)
 	<-done
 	assert.Equal(t, tt.expectedContentType, w.Header().Get("Content-Type"))
 	assert.Equal(t, 200, w.Code)