filters don't have their own goroutine anymore

app/daemon.go

@@ -116,50 +116,31 @@ func (f *Filter) cleanOldMatches(match string) {
 	f.matches[match] = newMatches
 }
 
-func (f *Filter) handle() chan *string {
+func (f *Filter) handle(line *string) {
-	lines := make(chan *string)
+	if match := f.match(line); match != "" {
 
-	go func() {
+		entry := LogEntry{time.Now(), match, f.stream.name, f.name, false}
-		for line := range lines {
-			if match := f.match(line); match != "" {
 
-				entry := LogEntry{time.Now(), match, f.stream.name, f.name, false}
+		if f.Retry > 1 {
+			f.cleanOldMatches(match)
 
-				if f.Retry > 1 {
+			f.matches[match] = append(f.matches[match], time.Now())
-					f.cleanOldMatches(match)
-
-					f.matches[match] = append(f.matches[match], time.Now())
-				}
-
-				if f.Retry <= 1 || len(f.matches[match]) >= f.Retry {
-					entry.Exec = true
-					delete(f.matches, match)
-					f.execActions(match, time.Duration(0))
-				}
-
-				logs <- entry
-			}
 		}
-	}()
 
-	return lines
+		if f.Retry <= 1 || len(f.matches[match]) >= f.Retry {
+			entry.Exec = true
+			delete(f.matches, match)
+			f.execActions(match, time.Duration(0))
+		}
+
+		logs <- entry
+	}
 }
 
 func (s *Stream) handle(endedSignal chan *Stream) {
 	log.Printf("INFO  %s: start %s\n", s.name, s.Cmd)
 
 	lines := cmdStdout(s.Cmd)
-
-	var filterInputs = make([]chan *string, 0, len(s.Filters))
-	for _, filter := range s.Filters {
-		filterInputs = append(filterInputs, filter.handle())
-	}
-	defer func() {
-		for _, filterInput := range filterInputs {
-			close(filterInput)
-		}
-	}()
-
 	for {
 		select {
 		case line, ok := <-lines:
@@ -167,8 +148,8 @@ func (s *Stream) handle(endedSignal chan *Stream) {
 				endedSignal <- s
 				return
 			}
-			for _, filterInput := range filterInputs {
+			for _, filter := range s.Filters {
-				filterInput <- line
+				filter.handle(line)
 			}
 		case _, ok := <-stopStreams:
 			if !ok {