Merge pull request 'add options in broker' (#333) from devstigneev/micro:master into master

Reviewed-on: #333

.gitignore

@@ -1,6 +1,8 @@
 # Develop tools
 /.vscode/
 /.idea/
+.idea
+.vscode
 
 # Binaries for programs and plugins
 *.exe
@@ -13,6 +15,7 @@
 _obj
 _test
 _build
+.DS_Store
 
 # Architecture specific extensions/prefixes
 *.[568vq]

broker/broker.go

@@ -4,6 +4,7 @@ package broker // import "go.unistack.org/micro/v4/broker"
 import (
 	"context"
 	"errors"
+	"time"
 
 	"go.unistack.org/micro/v4/metadata"
 	"go.unistack.org/micro/v4/options"
@@ -19,6 +20,8 @@ var (
 	ErrDisconnected = errors.New("broker disconnected")
 	// ErrInvalidMessage returns when message has nvalid format
 	ErrInvalidMessage = errors.New("broker message has invalid format")
+	// DefaultGracefulTimeout
+	DefaultGracefulTimeout = 5 * time.Second
 )
 
 // Broker is an interface used for asynchronous messaging.

broker/options.go

@@ -11,6 +11,7 @@ import (
 	"go.unistack.org/micro/v4/meter"
 	"go.unistack.org/micro/v4/options"
 	"go.unistack.org/micro/v4/register"
+	"go.unistack.org/micro/v4/sync"
 	"go.unistack.org/micro/v4/tracer"
 )
 
@@ -36,22 +37,27 @@ type Options struct {
 	Name string
 	// Address holds the broker address
 	Address []string
+
+	Wait *sync.WaitGroup
+
+	GracefulTimeout time.Duration
 }
 
 // NewOptions create new Options
 func NewOptions(opts ...options.Option) Options {
-	options := Options{
-		Register: register.DefaultRegister,
-		Logger:   logger.DefaultLogger,
-		Context:  context.Background(),
-		Meter:    meter.DefaultMeter,
-		Codecs:   make(map[string]codec.Codec),
-		Tracer:   tracer.DefaultTracer,
+	newOpts := Options{
+		Register:        register.DefaultRegister,
+		Logger:          logger.DefaultLogger,
+		Context:         context.Background(),
+		Meter:           meter.DefaultMeter,
+		Codecs:          make(map[string]codec.Codec),
+		Tracer:          tracer.DefaultTracer,
+		GracefulTimeout: DefaultGracefulTimeout,
 	}
 	for _, o := range opts {
-		o(&options)
+		o(&newOpts)
 	}
-	return options
+	return newOpts
 }
 
 // PublishOptions struct

logger/logger.go

@@ -64,8 +64,8 @@ func Error(ctx context.Context, msg string, attrs ...interface{}) {
 	DefaultLogger.Error(ctx, msg, attrs...)
 }
 
-// Debugf writes formatted msg to default logger on debug level
-func Debugf(ctx context.Context, msg string, attrs ...interface{}) {
+// Debug writes formatted msg to default logger on debug level
+func Debug(ctx context.Context, msg string, attrs ...interface{}) {
 	DefaultLogger.Debug(ctx, msg, attrs...)
 }
 

tracer/tracer.go

@@ -1,10 +1,8 @@
 // Package tracer provides an interface for distributed tracing
-package tracer // import "go.unistack.org/micro/v4/tracer"
+package tracer
 
 import (
 	"context"
-	"fmt"
-	"sort"
 
 	"go.unistack.org/micro/v4/logger"
 	"go.unistack.org/micro/v4/options"
@@ -71,37 +69,3 @@ type Span interface {
 	// SpanID returns span id
 	SpanID() string
 }
-
-// sort labels alphabeticaly by label name
-type byKey []interface{}
-
-func (k byKey) Len() int           { return len(k) / 2 }
-func (k byKey) Less(i, j int) bool { return fmt.Sprintf("%s", k[i*2]) < fmt.Sprintf("%s", k[j*2]) }
-func (k byKey) Swap(i, j int) {
-	k[i*2], k[j*2] = k[j*2], k[i*2]
-	k[i*2+1], k[j*2+1] = k[j*2+1], k[i*2+1]
-}
-
-func UniqLabels(labels []interface{}) []interface{} {
-	if len(labels)%2 == 1 {
-		labels = labels[:len(labels)-1]
-	}
-
-	if len(labels) > 2 {
-		sort.Sort(byKey(labels))
-
-		idx := 0
-		for {
-			if labels[idx] == labels[idx+2] {
-				copy(labels[idx:], labels[idx+2:])
-				labels = labels[:len(labels)-2]
-			} else {
-				idx += 2
-			}
-			if idx+2 >= len(labels) {
-				break
-			}
-		}
-	}
-	return labels
-}

util/reflect/reflect.go

@@ -1,6 +1,7 @@
-package reflect // import "go.unistack.org/micro/v4/util/reflect"
+package reflect
 
 import (
+	"encoding/json"
 	"errors"
 	"fmt"
 	"reflect"
@@ -45,15 +46,23 @@ func SliceAppend(b bool) Option {
 
 // Merge merges map[string]interface{} to destination struct
 func Merge(dst interface{}, mp map[string]interface{}, opts ...Option) error {
-	var err error
-	var sval reflect.Value
-	var fname string
-
 	options := Options{}
 	for _, o := range opts {
 		o(&options)
 	}
 
+	if unmarshaler, ok := dst.(json.Unmarshaler); ok {
+		buf, err := json.Marshal(mp)
+		if err == nil {
+			err = unmarshaler.UnmarshalJSON(buf)
+		}
+		return err
+	}
+
+	var err error
+	var sval reflect.Value
+	var fname string
+
 	dviface := reflect.ValueOf(dst)
 	if dviface.Kind() == reflect.Ptr {
 		dviface = dviface.Elem()

util/sort/sort.go

@@ -0,0 +1,40 @@
+package sort
+
+import (
+	"fmt"
+	"sort"
+)
+
+// sort labels alphabeticaly by label name
+type byKey []interface{}
+
+func (k byKey) Len() int           { return len(k) / 2 }
+func (k byKey) Less(i, j int) bool { return fmt.Sprintf("%s", k[i*2]) < fmt.Sprintf("%s", k[j*2]) }
+func (k byKey) Swap(i, j int) {
+	k[i*2], k[j*2] = k[j*2], k[i*2]
+	k[i*2+1], k[j*2+1] = k[j*2+1], k[i*2+1]
+}
+
+func Uniq(labels []interface{}) []interface{} {
+	if len(labels)%2 == 1 {
+		labels = labels[:len(labels)-1]
+	}
+
+	if len(labels) > 2 {
+		sort.Sort(byKey(labels))
+
+		idx := 0
+		for {
+			if labels[idx] == labels[idx+2] {
+				copy(labels[idx:], labels[idx+2:])
+				labels = labels[:len(labels)-2]
+			} else {
+				idx += 2
+			}
+			if idx+2 >= len(labels) {
+				break
+			}
+		}
+	}
+	return labels
+}