Create watcher

cmd/watch.go

@@ -23,8 +23,6 @@ package cmd
 
 import (
 	"log"
-	"net/url"
-	"strings"
 	"sync"
 
 	"github.com/Kashkovsky/hostmonitor/core"
@@ -42,50 +40,20 @@ var watchCmd = &cobra.Command{
 
 func runWatch(cmd *cobra.Command, args []string) {
 	log.Default().Println("Testing URLs from config ", watchConfig.ConfigUrl)
+	resMap := sync.Map{}
 	printer := core.NewPrinter()
-	res := sync.Map{}
-
-	c, _, err := doWatch()
-	if err != nil {
-		log.Fatalf("Fatal: %v", err)
-		return
-	}
-
-	for {
-		rec := <-c
-		res.Store(rec.Id, rec)
-		printer.ToTable(&res)
-	}
-}
-
-func doWatch() (chan core.TestResult, int, error) {
-	config, err := watchConfig.Update()
-	outC := make(chan core.TestResult, 50)
-	if err != nil {
-		log.Fatalf("Could not obtain a config: %v", err.Error())
-
-		return outC, 0, err
-	}
-
-	tester := core.NewTester(watchConfig, outC)
-
-	records := strings.Split(config, "\n")
-	for _, addr := range records {
-		u, err := url.Parse(addr)
-		if err != nil {
-			log.Default().Printf("Invalid url: %v, skipping...", addr)
-			continue
-		}
-
-		go tester.Test(u)
-	}
-
-	return outC, len(records), nil
+	watcher := core.NewWatcher(&watchConfig)
+
+	watcher.Watch(func(res core.TestResult) {
+		resMap.Store(res.Id, res)
+		printer.ToTable(&resMap)
+	})
 }
 
 func init() {
 	rootCmd.AddCommand(watchCmd)
 	watchCmd.Flags().StringVarP(&watchConfig.ConfigUrl, "configUrl", "c", core.ITArmyConfigURL, "Url of config containing url list")
 	watchCmd.Flags().IntVarP(&watchConfig.TestInterval, "testInterval", "i", 10, "Interval in seconds between test updates")
 	watchCmd.Flags().IntVarP(&watchConfig.RequestTimeout, "requestTimeout", "t", 5, "Request timeout")
+	watchCmd.Flags().IntVarP(&watchConfig.UpdateInterval, "updateInterval", "u", 60, "Config update interval in seconds")
 }

core/config.go

@@ -4,6 +4,7 @@ type WatchConfig struct {
 	ConfigUrl      string
 	TestInterval   int
 	RequestTimeout int
+	UpdateInterval int
 }
 
 func (c *WatchConfig) Update() (string, error) {

core/tester.go

@@ -16,34 +16,39 @@ type TestResult struct {
 
 type Tester struct {
 	requestTimeout time.Duration
-	testInterval time.Duration
-	out chan TestResult
+	testInterval   time.Duration
+	out            chan TestResult
+	quit           chan bool
 }
 
-func NewTester(config WatchConfig, out chan TestResult) Tester {
-	return Tester { 
-		requestTimeout: time.Duration(config.RequestTimeout) * time.Second, 
-		testInterval: time.Duration(config.TestInterval) * time.Second,
-		out: out,
+func NewTester(config *WatchConfig, out chan TestResult, quit chan bool) Tester {
+	return Tester{
+		requestTimeout: time.Duration(config.RequestTimeout) * time.Second,
+		testInterval:   time.Duration(config.TestInterval) * time.Second,
+		out:            out,
+		quit:           quit,
 	}
 }
 
-func (t *Tester)Test(url *url.URL) {
+func (t *Tester) Test(url *url.URL) {
 	tp := NewTransport(t.requestTimeout)
 
 	for {
-		t.out <- TestResult{Id: url.Host, url: *url, status: "Test"}
+		select {
+		case <-t.quit:
+			return
+		default:
+			_, err := tp.Dial(url.Scheme, url.Host)
 
-		_, err := tp.Dial(url.Scheme, url.Host)
+			if err != nil {
+				t.out <- TestResult{Id: url.Host, url: *url, status: formatError(err, url), duration: tp.ConnDuration()}
+				return
+			}
 
-		if err != nil {
-			t.out <- TestResult{Id: url.Host, url: *url, status: formatError(err, url), duration: tp.ConnDuration()}
-			return
+			t.out <- TestResult{Id: url.Host, url: *url, status: "OK", duration: tp.Duration()}
+			time.Sleep(t.testInterval)
 		}
 
-		t.out <- TestResult{Id: url.Host, url: *url, status: "OK", duration: tp.Duration()}
-		time.Sleep(t.testInterval)
-
 	}
 }
 

core/watcher.go

@@ -0,0 +1,75 @@
+package core
+
+import (
+	"log"
+	"net/url"
+	"strings"
+	"time"
+)
+
+type Watcher struct {
+	config *WatchConfig
+	tester Tester
+	quit   chan bool
+	out    chan TestResult
+}
+
+func NewWatcher(config *WatchConfig) Watcher {
+	outC := make(chan TestResult, 50)
+	quit := make(chan bool)
+	tester := NewTester(config, outC, quit)
+	return Watcher{
+		config: config,
+		tester: tester,
+		quit:   quit,
+		out:    outC,
+	}
+}
+
+func (w *Watcher) Watch(f func(TestResult)) {
+	for {
+		err := w.doWatch()
+
+		if err != nil {
+			log.Fatalf("Fatal: %v", err)
+			return
+		}
+
+		go func() {
+			for {
+				select {
+				case <-w.quit:
+					return
+				case rec := <-w.out:
+					f(rec)
+				}
+			}
+		}()
+
+		time.Sleep(time.Duration(w.config.UpdateInterval) * time.Second)
+		w.quit <- true
+	}
+}
+
+func (w *Watcher) doWatch() error {
+	log.Default().Println("Fetching new config...")
+	config, err := w.config.Update()
+
+	if err != nil {
+		log.Fatalf("Could not obtain a config: %v", err.Error())
+		return err
+	}
+
+	records := strings.Split(config, "\n")
+	for _, addr := range records {
+		u, err := url.Parse(addr)
+		if err != nil {
+			log.Default().Printf("Invalid url: %v, skipping...", addr)
+			continue
+		}
+
+		go w.tester.Test(u)
+	}
+
+	return nil
+}