Merge pull request #105 from unistack-org/improve

small fixes

register/memory.go

@@ -2,7 +2,6 @@ package register
 
 import (
 	"context"
-	"errors"
 	"sync"
 	"time"
 
@@ -438,7 +437,7 @@ func (m *watcher) Next() (*Result, error) {
 				return r, nil
 			}
 		case <-m.exit:
-			return nil, errors.New("watcher stopped")
+			return nil, ErrWatcherStopped
 		}
 	}
 }

register/memory_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"sync"
 	"testing"
 	"time"
 )
@@ -284,29 +285,39 @@ func TestMemoryWildcard(t *testing.T) {
 }
 
 func TestWatcher(t *testing.T) {
-	w := &watcher{
+	testSrv := &Service{Name: "foo", Version: "1.0.0"}
-		id:   "test",
-		res:  make(chan *Result),
-		exit: make(chan bool),
-		wo: WatchOptions{
-			Domain: WildcardDomain,
-		},
-	}
 
+	ctx := context.TODO()
+	m := NewRegister()
+	m.Init()
+	m.Connect(ctx)
+	wc, err := m.Watch(ctx)
+	if err != nil {
+		t.Fatalf("cant watch: %v", err)
+	}
+	defer wc.Stop()
+
+	var wg sync.WaitGroup
+	wg.Add(1)
 	go func() {
-		w.res <- &Result{
+		for {
-			Service: &Service{Name: "foo"},
+			ch, err := wc.Next()
+			if err != nil {
+				t.Fatal("unexpected err", err)
+			}
+			t.Logf("changes %#+v", ch.Service)
+			wc.Stop()
+			wg.Done()
+			return
 		}
 	}()
 
-	_, err := w.Next()
+	if err := m.Register(ctx, testSrv); err != nil {
-	if err != nil {
+		t.Fatalf("Register err: %v", err)
-		t.Fatal("unexpected err", err)
 	}
 
-	w.Stop()
+	wg.Wait()
-
+	if _, err := wc.Next(); err == nil {
-	if _, err := w.Next(); err == nil {
 		t.Fatal("expected error on Next()")
 	}
 }

service.go

@@ -54,8 +54,10 @@ type Service interface {
 	// Runtime(string) (runtime.Runtime, bool)
 	// Profile
 	// Profile(string) (profile.Profile, bool)
-	// Run the service
+	// Run the service and wait
 	Run() error
+	// Start the service
+	Start() error
 	// The service implementation
 	String() string
 }
@@ -257,7 +259,7 @@ func (s *service) Start() error {
 	s.RUnlock()
 
 	if config.Loggers[0].V(logger.InfoLevel) {
-		config.Loggers[0].Infof(s.opts.Context, "starting [service] %s", s.Name())
+		config.Loggers[0].Infof(s.opts.Context, "starting [service] %s-%s", s.Options().Name, s.Options().Version)
 	}
 
 	for _, fn := range s.opts.BeforeStart {

util/reflect/struct.go

@@ -245,8 +245,13 @@ func StructFields(src interface{}) ([]StructField, error) {
 
 		switch val.Kind() {
 		case reflect.Ptr:
-			//	if !val.IsValid()
+			if val.CanSet() && fld.Type.Elem().Kind() == reflect.Struct {
-			if reflect.Indirect(val).Kind() == reflect.Struct {
+				if val.IsNil() {
+					val.Set(reflect.New(fld.Type.Elem()))
+				}
+			}
+			switch reflect.Indirect(val).Kind() {
+			case reflect.Struct:
 				infields, err := StructFields(val.Interface())
 				if err != nil {
 					return nil, err
@@ -255,7 +260,7 @@ func StructFields(src interface{}) ([]StructField, error) {
 					infield.Path = fmt.Sprintf("%s.%s", fld.Name, infield.Path)
 					fields = append(fields, infield)
 				}
-			} else {
+			default:
 				fields = append(fields, StructField{Field: fld, Value: val, Path: fld.Name})
 			}
 		case reflect.Struct:
@@ -268,6 +273,7 @@ func StructFields(src interface{}) ([]StructField, error) {
 				fields = append(fields, infield)
 			}
 		default:
+
 			fields = append(fields, StructField{Field: fld, Value: val, Path: fld.Name})
 		}
 	}

util/reflect/struct_test.go

@@ -10,22 +10,28 @@ import (
 )
 
 func TestStructfields(t *testing.T) {
+	type NestedConfig struct {
+		Value string
+	}
 	type Config struct {
 		Time     time.Time
-		Nested   *Config
+		Nested   *NestedConfig
 		Metadata map[string]int
 		Broker   string
 		Addr     []string
 		Wait     time.Duration
 		Verbose  bool
 	}
-	cfg := &Config{Nested: &Config{}}
+	cfg := &Config{Nested: &NestedConfig{}}
 	fields, err := rutil.StructFields(cfg)
 	if err != nil {
 		t.Fatal(err)
 	}
-	if len(fields) != 13 {
+	if len(fields) != 7 {
-		t.Fatalf("invalid fields number: %v", fields)
+		for _, field := range fields {
+			t.Logf("field %#+v\n", field)
+		}
+		t.Fatalf("invalid fields number: %d != %d", 7, len(fields))
 	}
 }