Merge pull request #686 from milosgajdos83/config-watcher

Introduce ErrStoppedWatcher for source.Source Watchers and fixed test

config/default_test.go

@@ -145,16 +145,26 @@ func TestFileChange(t *testing.T) {
 	if err != nil {
 		t.Error(err)
 	}
+
 	changeTimes := 0
+	done := make(chan bool)
+
 	go func() {
+		defer func() {
+			close(done)
+		}()
+
 		for {
 			v, err := watcher.Next()
 			if err != nil {
-				t.Error(err)
+				if err.Error() != "watcher stopped" {
+					t.Error(err)
+					return
+				}
 				return
 			}
 			changeTimes++
-			t.Logf("file change，%s", string(v.Bytes()))
+			t.Logf("file change，%s: %d", string(v.Bytes()), changeTimes)
 		}
 	}()
 
@@ -169,10 +179,17 @@ func TestFileChange(t *testing.T) {
 			t.Error(err)
 		}
 
-		time.Sleep(time.Second)
+		time.Sleep(500 * time.Millisecond)
 	}
 
-	if changeTimes != 4 {
+	if err := watcher.Stop(); err != nil {
-		t.Error(fmt.Errorf("watcher error: change times %d is not enough", changeTimes))
+		t.Fatalf("failed to stop watcher: %s", err)
+	}
+
+	// wait for the watcher to finish
+	<-done
+
+	if changeTimes != 5 {
+		t.Errorf("watcher error: change times %d is not enough", changeTimes)
 	}
 }

config/source/consul/watcher.go

@@ -1,7 +1,6 @@
 package consul
 
 import (
-	"errors"
 	"time"
 
 	"github.com/hashicorp/consul/api"
@@ -80,7 +79,7 @@ func (w *watcher) Next() (*source.ChangeSet, error) {
 	case cs := <-w.ch:
 		return cs, nil
 	case <-w.exit:
-		return nil, errors.New("watcher stopped")
+		return nil, source.ErrWatcherStopped
 	}
 }
 

config/source/env/env_test.go

@@ -86,8 +86,8 @@ func TestEnvvar_Prefixes(t *testing.T) {
 }
 
 func TestEnvvar_WatchNextNoOpsUntilStop(t *testing.T) {
-	source := NewSource(WithStrippedPrefix("GOMICRO_"))
+	src := NewSource(WithStrippedPrefix("GOMICRO_"))
-	w, err := source.Watch()
+	w, err := src.Watch()
 	if err != nil {
 		t.Error(err)
 	}
@@ -97,7 +97,7 @@ func TestEnvvar_WatchNextNoOpsUntilStop(t *testing.T) {
 		w.Stop()
 	}()
 
-	if _, err := w.Next(); err.Error() != "watcher stopped" {
+	if _, err := w.Next(); err != source.ErrWatcherStopped {
 		t.Errorf("expected watcher stopped error, got %v", err)
 	}
 }

config/source/env/watcher.go

@@ -1,8 +1,6 @@
 package env
 
 import (
-	"errors"
-
 	"github.com/micro/go-micro/config/source"
 )
 
@@ -13,7 +11,7 @@ type watcher struct {
 func (w *watcher) Next() (*source.ChangeSet, error) {
 	<-w.exit
 
-	return nil, errors.New("watcher stopped")
+	return nil, source.ErrWatcherStopped
 }
 
 func (w *watcher) Stop() error {

config/source/file/watcher.go

@@ -3,7 +3,6 @@
 package file
 
 import (
-	"errors"
 	"os"
 
 	"github.com/fsnotify/fsnotify"
@@ -36,7 +35,7 @@ func (w *watcher) Next() (*source.ChangeSet, error) {
 	// is it closed?
 	select {
 	case <-w.exit:
-		return nil, errors.New("watcher stopped")
+		return nil, source.ErrWatcherStopped
 	default:
 	}
 
@@ -59,7 +58,7 @@ func (w *watcher) Next() (*source.ChangeSet, error) {
 	case err := <-w.fw.Errors:
 		return nil, err
 	case <-w.exit:
-		return nil, errors.New("watcher stopped")
+		return nil, source.ErrWatcherStopped
 	}
 }
 

config/source/file/watcher_linux.go

@@ -3,7 +3,6 @@
 package file
 
 import (
-	"errors"
 	"os"
 
 	"github.com/fsnotify/fsnotify"
@@ -36,7 +35,7 @@ func (w *watcher) Next() (*source.ChangeSet, error) {
 	// is it closed?
 	select {
 	case <-w.exit:
-		return nil, errors.New("watcher stopped")
+		return nil, source.ErrWatcherStopped
 	default:
 	}
 
@@ -63,7 +62,7 @@ func (w *watcher) Next() (*source.ChangeSet, error) {
 	case err := <-w.fw.Errors:
 		return nil, err
 	case <-w.exit:
-		return nil, errors.New("watcher stopped")
+		return nil, source.ErrWatcherStopped
 	}
 }
 

config/source/source.go

@@ -2,9 +2,15 @@
 package source
 
 import (
+	"errors"
 	"time"
 )
 
+var (
+	// ErrWatcherStopped is returned when source watcher has been stopped
+	ErrWatcherStopped = errors.New("watcher stopped")
+)
+
 // Source is the source from which config is loaded
 type Source interface {
 	Read() (*ChangeSet, error)