157 lines
3.2 KiB
Go
157 lines
3.2 KiB
Go
package app
|
|
|
|
import (
|
|
"bufio"
|
|
"flag"
|
|
// "fmt"
|
|
"log"
|
|
"os"
|
|
"os/exec"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
)
|
|
|
|
// Executes a command and channel-send its stdout
|
|
func cmdStdout(commandline []string) chan *string {
|
|
lines := make(chan *string)
|
|
|
|
go func() {
|
|
cmd := exec.Command(commandline[0], commandline[1:]...)
|
|
stdout, err := cmd.StdoutPipe()
|
|
if err != nil {
|
|
log.Fatal("couldn't open stdout on command:", err)
|
|
}
|
|
if err := cmd.Start(); err != nil {
|
|
log.Fatal("couldn't start command:", err)
|
|
}
|
|
defer stdout.Close()
|
|
|
|
scanner := bufio.NewScanner(stdout)
|
|
for scanner.Scan() {
|
|
line := scanner.Text()
|
|
lines <- &line
|
|
}
|
|
close(lines)
|
|
}()
|
|
|
|
return lines
|
|
}
|
|
|
|
// Whether one of the filter's regexes is matched on a line
|
|
func (f *Filter) match(line *string) string {
|
|
for _, regex := range f.compiledRegex {
|
|
|
|
if matches := regex.FindStringSubmatch(*line); matches != nil {
|
|
|
|
match := matches[regex.SubexpIndex(f.patternName)]
|
|
|
|
log.Printf("INFO %s.%s: match [%v]\n", f.stream.name, f.name, match)
|
|
return match
|
|
}
|
|
}
|
|
return ""
|
|
}
|
|
|
|
func (f *Filter) execActions(match string) {
|
|
for _, a := range f.Actions {
|
|
wgActions.Add(1)
|
|
go a.exec(match)
|
|
}
|
|
}
|
|
|
|
func (a *Action) exec(match string) {
|
|
defer wgActions.Done()
|
|
if a.afterDuration != 0 {
|
|
time.Sleep(a.afterDuration)
|
|
}
|
|
|
|
computedCommand := make([]string, 0, len(a.Cmd))
|
|
for _, item := range a.Cmd {
|
|
computedCommand = append(computedCommand, strings.ReplaceAll(item, a.filter.patternWithBraces, match))
|
|
}
|
|
|
|
log.Printf("INFO %s.%s.%s: run %s\n", a.filter.stream.name, a.filter.name, a.name, computedCommand)
|
|
|
|
cmd := exec.Command(computedCommand[0], computedCommand[1:]...)
|
|
|
|
if ret := cmd.Run(); ret != nil {
|
|
log.Printf("ERR %s.%s.%s: run %s, code %s\n", a.filter.stream.name, a.filter.name, a.name, computedCommand, ret)
|
|
}
|
|
}
|
|
|
|
func (f *Filter) handle() chan *string {
|
|
lines := make(chan *string)
|
|
|
|
go func() {
|
|
for line := range lines {
|
|
if match := f.match(line); match != "" {
|
|
f.execActions(match)
|
|
}
|
|
}
|
|
}()
|
|
|
|
return lines
|
|
}
|
|
|
|
func multiplex(input chan *string, outputs []chan *string) {
|
|
var wg sync.WaitGroup
|
|
for item := range input {
|
|
for _, output := range outputs {
|
|
wg.Add(1)
|
|
go func(s *string) {
|
|
output <- s
|
|
wg.Done()
|
|
}(item)
|
|
}
|
|
}
|
|
for _, output := range outputs {
|
|
wg.Wait()
|
|
close(output)
|
|
}
|
|
}
|
|
|
|
func (s *Stream) handle(signal 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())
|
|
}
|
|
|
|
multiplex(lines, filterInputs)
|
|
|
|
signal <- s
|
|
}
|
|
|
|
var wgActions sync.WaitGroup
|
|
|
|
func Main() {
|
|
confFilename := flag.String("c", "", "configuration file. see an example at https://framagit.org/ppom/reaction/-/blob/main/reaction.yml")
|
|
flag.Parse()
|
|
|
|
if *confFilename == "" {
|
|
flag.PrintDefaults()
|
|
os.Exit(2)
|
|
}
|
|
|
|
conf := parseConf(*confFilename)
|
|
|
|
endSignals := make(chan *Stream)
|
|
|
|
for _, stream := range conf.Streams {
|
|
go stream.handle(endSignals)
|
|
}
|
|
|
|
for i := 0; i < len(conf.Streams); i++ {
|
|
finishedStream := <-endSignals
|
|
log.Printf("ERR %s stream finished", finishedStream.name)
|
|
}
|
|
|
|
wgActions.Wait()
|
|
|
|
os.Exit(3)
|
|
}
|