config: add Save/Load options

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>

.github/dependabot.yml

@@ -0,0 +1,19 @@
+# To get started with Dependabot version updates, you'll need to specify which
+# package ecosystems to update and where the package manifests are located.
+# Please see the documentation for all configuration options:
+# https://help.github.com/github/administering-a-repository/configuration-options-for-dependency-updates
+
+version: 2
+updates:
+
+  # Maintain dependencies for GitHub Actions
+  - package-ecosystem: "github-actions"
+    directory: "/"
+    schedule:
+      interval: "daily"
+
+  # Maintain dependencies for Golang
+  - package-ecosystem: "gomod"
+    directory: "/"
+    schedule:
+      interval: "daily"

.github/renovate.json

@@ -1,20 +0,0 @@
-{
-  "extends": [
-    "config:base"
-  ],
-  "postUpdateOptions": ["gomodTidy"],
-  "packageRules": [
-    {
-      "matchUpdateTypes": ["minor", "patch", "pin", "digest"],
-      "automerge": true
-    },
-    {
-      "groupName": "all deps",
-      "separateMajorMinor": true,
-      "groupSlug": "all",
-      "packagePatterns": [
-        "*"
-      ]
-    }
-  ]
-}

.github/stale.sh

@@ -1,13 +0,0 @@
-#!/bin/bash -ex
-
-export PATH=$PATH:$(pwd)/bin
-export GO111MODULE=on
-export GOBIN=$(pwd)/bin
-
-#go get github.com/rvflash/goup@v0.4.1
-
-#goup -v ./...
-#go get github.com/psampaz/go-mod-outdated@v0.6.0
-go list -u -m -mod=mod -json all | go-mod-outdated -update -direct -ci || true
-
-#go list -u -m -json all | go-mod-outdated -update

.github/workflows/build.yml

@@ -50,6 +50,7 @@ jobs:
         uses: actions/checkout@v2
       - name: lint
         uses: golangci/golangci-lint-action@v2
+        continue-on-error: true
         with:
           # Required: the version of golangci-lint is required and must be specified without patch version: we always use the latest patch version.
           version: v1.39

.github/workflows/pr.yml

@@ -50,6 +50,7 @@ jobs:
         uses: actions/checkout@v2
       - name: lint
         uses: golangci/golangci-lint-action@v2
+        continue-on-error: true
         with:
           # Required: the version of golangci-lint is required and must be specified without patch version: we always use the latest patch version.
           version: v1.39

client/noop.go

@@ -221,5 +221,8 @@ func (n *noopClient) Publish(ctx context.Context, p Message, opts ...PublishOpti
 	return n.opts.Broker.Publish(ctx, topic, &broker.Message{
 		Header: md,
 		Body:   body,
-	}, broker.PublishContext(options.Context))
+	},
+		broker.PublishContext(options.Context),
+		broker.PublishBodyOnly(options.BodyOnly),
+	)
 }

client/options.go

@@ -118,6 +118,8 @@ func NewPublishOptions(opts ...PublishOption) PublishOptions {
 
 // PublishOptions holds publish options
 type PublishOptions struct {
+	// BodyOnly will publish only message body
+	BodyOnly bool
 	// Context used for external options
 	Context context.Context
 	// Exchange topic exchange name
@@ -377,6 +379,13 @@ func WithExchange(e string) PublishOption {
 	}
 }
 
+// WithBodyOnly publish only message body
+func WithBodyOnly(b bool) PublishOption {
+	return func(o *PublishOptions) {
+		o.BodyOnly = b
+	}
+}
+
 // PublishContext sets the context in publish options
 func PublishContext(ctx context.Context) PublishOption {
 	return func(o *PublishOptions) {

codec/codec.go

@@ -28,6 +28,8 @@ var (
 	DefaultMaxMsgSize int = 1024 * 1024 * 4 // 4Mb
 	// DefaultCodec is the global default codec
 	DefaultCodec Codec = NewCodec()
+	// DefaultTagName specifies struct tag name to control codec Marshal/Unmarshal
+	DefaultTagName = "codec"
 )
 
 // MessageType specifies message type for codec

config/config.go

@@ -20,25 +20,27 @@ var (
 
 // Config is an interface abstraction for dynamic configuration
 type Config interface {
+	// Name returns name of config
 	Name() string
 	// Init the config
 	Init(opts ...Option) error
 	// Options in the config
 	Options() Options
 	// Load config from sources
-	Load(context.Context) error
+	Load(context.Context, ...LoadOption) error
 	// Save config to sources
-	Save(context.Context) error
+	Save(context.Context, ...SaveOption) error
 	// Watch a value for changes
-	//	Watch(interface{}) (Watcher, error)
+	//Watch(context.Context) (Watcher, error)
+	// String returns config type name
 	String() string
 }
 
 // Watcher is the config watcher
-// type Watcher interface {
-//	Next() (, error)
-//	Stop() error
-// }
+type Watcher interface {
+	// Next() (, error)
+	Stop() error
+}
 
 // Load loads config from config sources
 func Load(ctx context.Context, cs ...Config) error {

config/default.go

@@ -25,18 +25,27 @@ func (c *defaultConfig) Init(opts ...Option) error {
 	return nil
 }
 
-func (c *defaultConfig) Load(ctx context.Context) error {
+func (c *defaultConfig) Load(ctx context.Context, opts ...LoadOption) error {
 	for _, fn := range c.opts.BeforeLoad {
 		if err := fn(ctx, c); err != nil && !c.opts.AllowFail {
 			return err
 		}
 	}
 
+	options := NewLoadOptions(opts...)
+	mopts := []func(*mergo.Config){mergo.WithTypeCheck}
+	if options.Override {
+		mopts = append(mopts, mergo.WithOverride)
+	}
+	if options.Append {
+		mopts = append(mopts, mergo.WithAppendSlice)
+	}
+
 	src, err := rutil.Zero(c.opts.Struct)
 	if err == nil {
 		valueOf := reflect.ValueOf(src)
 		if err = c.fillValues(valueOf); err == nil {
-			err = mergo.Merge(c.opts.Struct, src, mergo.WithOverride, mergo.WithTypeCheck, mergo.WithAppendSlice)
+			err = mergo.Merge(c.opts.Struct, src, mopts...)
 		}
 	}
 
@@ -232,7 +241,7 @@ func (c *defaultConfig) fillValues(valueOf reflect.Value) error {
 	return nil
 }
 
-func (c *defaultConfig) Save(ctx context.Context) error {
+func (c *defaultConfig) Save(ctx context.Context, opts ...SaveOption) error {
 	for _, fn := range c.opts.BeforeSave {
 		if err := fn(ctx, c); err != nil && !c.opts.AllowFail {
 			return err

config/options.go

@@ -57,6 +57,52 @@ func NewOptions(opts ...Option) Options {
 	return options
 }
 
+// LoadOption function signature
+type LoadOption func(o *LoadOptions)
+
+// LoadOptions struct
+type LoadOptions struct {
+	Override bool
+	Append   bool
+}
+
+func NewLoadOptions(opts ...LoadOption) LoadOptions {
+	options := LoadOptions{}
+	for _, o := range opts {
+		o(&options)
+	}
+	return options
+}
+
+// LoadOverride override values when load
+func LoadOverride(b bool) LoadOption {
+	return func(o *LoadOptions) {
+		o.Override = b
+	}
+}
+
+// LoadAppend override values when load
+func LoadAppend(b bool) LoadOption {
+	return func(o *LoadOptions) {
+		o.Append = b
+	}
+}
+
+// SaveOption function signature
+type SaveOption func(o *SaveOptions)
+
+// SaveOptions struct
+type SaveOptions struct {
+}
+
+func NewSaveOptions(opts ...SaveOption) SaveOptions {
+	options := SaveOptions{}
+	for _, o := range opts {
+		o(&options)
+	}
+	return options
+}
+
 // AllowFail allows config source to fail
 func AllowFail(b bool) Option {
 	return func(o *Options) {

go.mod

@@ -9,5 +9,5 @@ require (
 	github.com/imdario/mergo v0.3.12
 	github.com/patrickmn/go-cache v2.1.0+incompatible
 	github.com/silas/dag v0.0.0-20210121180416-41cf55125c34
-	golang.org/x/net v0.0.0-20210423184538-5f58ad60dda6
+	golang.org/x/net v0.0.0-20210510120150-4163338589ed
 )

go.sum

@@ -10,8 +10,8 @@ github.com/patrickmn/go-cache v2.1.0+incompatible h1:HRMgzkcYKYpi3C8ajMPV8OFXaaR
 github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ=
 github.com/silas/dag v0.0.0-20210121180416-41cf55125c34 h1:vBfVmA5mZhsQa2jr1FOL9nfA37N/jnbBmi5XUfviVTI=
 github.com/silas/dag v0.0.0-20210121180416-41cf55125c34/go.mod h1:7RTUFBdIRC9nZ7/3RyRNH1bdqIShrDejd1YbLwgPS+I=
-golang.org/x/net v0.0.0-20210423184538-5f58ad60dda6 h1:0PC75Fz/kyMGhL0e1QnypqK2kQMqKt9csD1GnMJR+Zk=
-golang.org/x/net v0.0.0-20210423184538-5f58ad60dda6/go.mod h1:OJAsFXCWl8Ukc7SiCT/9KSuxbyM7479/AVlXFRxuMCk=
+golang.org/x/net v0.0.0-20210510120150-4163338589ed h1:p9UgmWI9wKpfYmgaV/IZKGdXc5qEK45tDwwwDyjS26I=
+golang.org/x/net v0.0.0-20210510120150-4163338589ed/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=

meter/wrapper/wrapper.go

@@ -11,18 +11,25 @@ import (
 )
 
 var (
-	ClientRequestDurationSeconds        = "client_request_duration_seconds"
-	ClientRequestLatencyMicroseconds    = "client_request_latency_microseconds"
-	ClientRequestTotal                  = "client_request_total"
-	ServerRequestDurationSeconds        = "server_request_duration_seconds"
-	ServerRequestLatencyMicroseconds    = "server_request_latency_microseconds"
-	ServerRequestTotal                  = "server_request_total"
-	PublishMessageDurationSeconds       = "publish_message_duration_seconds"
-	PublishMessageLatencyMicroseconds   = "publish_message_latency_microseconds"
-	PublishMessageTotal                 = "publish_message_total"
+	ClientRequestDurationSeconds     = "client_request_duration_seconds"
+	ClientRequestLatencyMicroseconds = "client_request_latency_microseconds"
+	ClientRequestTotal               = "client_request_total"
+	ClientRequestInflight            = "client_request_inflight"
+
+	ServerRequestDurationSeconds     = "server_request_duration_seconds"
+	ServerRequestLatencyMicroseconds = "server_request_latency_microseconds"
+	ServerRequestTotal               = "server_request_total"
+	ServerRequestInflight            = "server_request_inflight"
+
+	PublishMessageDurationSeconds     = "publish_message_duration_seconds"
+	PublishMessageLatencyMicroseconds = "publish_message_latency_microseconds"
+	PublishMessageTotal               = "publish_message_total"
+	PublishMessageInflight            = "publish_message_inflight"
+
 	SubscribeMessageDurationSeconds     = "subscribe_message_duration_seconds"
 	SubscribeMessageLatencyMicroseconds = "subscribe_message_latency_microseconds"
 	SubscribeMessageTotal               = "subscribe_message_total"
+	SubscribeMessageInflight            = "subscribe_message_inflight"
 
 	labelSuccess  = "success"
 	labelFailure  = "failure"
@@ -116,13 +123,16 @@ func (w *wrapper) CallFunc(ctx context.Context, addr string, req client.Request,
 			return w.callFunc(ctx, addr, req, rsp, opts)
 		}
 	}
-	ts := time.Now()
-	err := w.callFunc(ctx, addr, req, rsp, opts)
-	te := time.Since(ts)
 
 	labels := make([]string, 0, 4)
 	labels = append(labels, labelEndpoint, endpoint)
 
+	w.opts.Meter.Counter(ClientRequestInflight, labels...).Inc()
+	ts := time.Now()
+	err := w.callFunc(ctx, addr, req, rsp, opts)
+	te := time.Since(ts)
+	w.opts.Meter.Counter(ClientRequestInflight, labels...).Dec()
+
 	w.opts.Meter.Summary(ClientRequestLatencyMicroseconds, labels...).Update(te.Seconds())
 	w.opts.Meter.Histogram(ClientRequestDurationSeconds, labels...).Update(te.Seconds())
 
@@ -144,12 +154,14 @@ func (w *wrapper) Call(ctx context.Context, req client.Request, rsp interface{},
 		}
 	}
 
+	labels := make([]string, 0, 4)
+	labels = append(labels, labelEndpoint, endpoint)
+
+	w.opts.Meter.Counter(ClientRequestInflight, labels...).Inc()
 	ts := time.Now()
 	err := w.Client.Call(ctx, req, rsp, opts...)
 	te := time.Since(ts)
-
-	labels := make([]string, 0, 4)
-	labels = append(labels, labelEndpoint, endpoint)
+	w.opts.Meter.Counter(ClientRequestInflight, labels...).Dec()
 
 	w.opts.Meter.Summary(ClientRequestLatencyMicroseconds, labels...).Update(te.Seconds())
 	w.opts.Meter.Histogram(ClientRequestDurationSeconds, labels...).Update(te.Seconds())
@@ -172,12 +184,14 @@ func (w *wrapper) Stream(ctx context.Context, req client.Request, opts ...client
 		}
 	}
 
+	labels := make([]string, 0, 4)
+	labels = append(labels, labelEndpoint, endpoint)
+
+	w.opts.Meter.Counter(ClientRequestInflight, labels...).Inc()
 	ts := time.Now()
 	stream, err := w.Client.Stream(ctx, req, opts...)
 	te := time.Since(ts)
-
-	labels := make([]string, 0, 4)
-	labels = append(labels, labelEndpoint, endpoint)
+	w.opts.Meter.Counter(ClientRequestInflight, labels...).Dec()
 
 	w.opts.Meter.Summary(ClientRequestLatencyMicroseconds, labels...).Update(te.Seconds())
 	w.opts.Meter.Histogram(ClientRequestDurationSeconds, labels...).Update(te.Seconds())
@@ -195,12 +209,14 @@ func (w *wrapper) Stream(ctx context.Context, req client.Request, opts ...client
 func (w *wrapper) Publish(ctx context.Context, p client.Message, opts ...client.PublishOption) error {
 	endpoint := p.Topic()
 
+	labels := make([]string, 0, 4)
+	labels = append(labels, labelEndpoint, endpoint)
+
+	w.opts.Meter.Counter(PublishMessageInflight, labels...).Inc()
 	ts := time.Now()
 	err := w.Client.Publish(ctx, p, opts...)
 	te := time.Since(ts)
-
-	labels := make([]string, 0, 4)
-	labels = append(labels, labelEndpoint, endpoint)
+	w.opts.Meter.Counter(PublishMessageInflight, labels...).Dec()
 
 	w.opts.Meter.Summary(PublishMessageLatencyMicroseconds, labels...).Update(te.Seconds())
 	w.opts.Meter.Histogram(PublishMessageDurationSeconds, labels...).Update(te.Seconds())
@@ -231,12 +247,14 @@ func (w *wrapper) HandlerFunc(fn server.HandlerFunc) server.HandlerFunc {
 			}
 		}
 
+		labels := make([]string, 0, 4)
+		labels = append(labels, labelEndpoint, endpoint)
+
+		w.opts.Meter.Counter(ServerRequestInflight, labels...).Inc()
 		ts := time.Now()
 		err := fn(ctx, req, rsp)
 		te := time.Since(ts)
-
-		labels := make([]string, 0, 4)
-		labels = append(labels, labelEndpoint, endpoint)
+		w.opts.Meter.Counter(ServerRequestInflight, labels...).Dec()
 
 		w.opts.Meter.Summary(ServerRequestLatencyMicroseconds, labels...).Update(te.Seconds())
 		w.opts.Meter.Histogram(ServerRequestDurationSeconds, labels...).Update(te.Seconds())
@@ -263,12 +281,14 @@ func (w *wrapper) SubscriberFunc(fn server.SubscriberFunc) server.SubscriberFunc
 	return func(ctx context.Context, msg server.Message) error {
 		endpoint := msg.Topic()
 
+		labels := make([]string, 0, 4)
+		labels = append(labels, labelEndpoint, endpoint)
+
+		w.opts.Meter.Counter(SubscribeMessageInflight, labels...).Inc()
 		ts := time.Now()
 		err := fn(ctx, msg)
 		te := time.Since(ts)
-
-		labels := make([]string, 0, 4)
-		labels = append(labels, labelEndpoint, endpoint)
+		w.opts.Meter.Counter(SubscribeMessageInflight, labels...).Dec()
 
 		w.opts.Meter.Summary(SubscribeMessageLatencyMicroseconds, labels...).Update(te.Seconds())
 		w.opts.Meter.Histogram(SubscribeMessageDurationSeconds, labels...).Update(te.Seconds())

util/reflect/reflect.go

@@ -12,6 +12,7 @@ import (
 var (
 	ErrInvalidStruct = errors.New("invalid struct specified")
 	ErrInvalidValue  = errors.New("invalid value specified")
+	ErrNotFound      = errors.New("struct field not found")
 )
 
 type Option func(*Options)

util/reflect/struct.go

@@ -7,6 +7,7 @@ import (
 	"reflect"
 	"regexp"
 	"strings"
+	"time"
 )
 
 // ErrInvalidParam specifies invalid url query params
@@ -14,9 +15,99 @@ var ErrInvalidParam = errors.New("invalid url query param provided")
 
 var bracketSplitter = regexp.MustCompile(`\[|\]`)
 
+var timeKind = reflect.TypeOf(time.Time{}).Kind()
+
+type StructField struct {
+	Field reflect.StructField
+	Value reflect.Value
+}
+
+func StructFieldByTag(src interface{}, tkey string, tval string) (interface{}, error) {
+	sv := reflect.ValueOf(src)
+	if sv.Kind() == reflect.Ptr {
+		sv = sv.Elem()
+	}
+	if sv.Kind() != reflect.Struct {
+		return nil, ErrInvalidStruct
+	}
+
+	typ := sv.Type()
+	for idx := 0; idx < typ.NumField(); idx++ {
+		fld := typ.Field(idx)
+		val := sv.Field(idx)
+		if !val.CanSet() || len(fld.PkgPath) != 0 {
+			continue
+		}
+
+		if ts, ok := fld.Tag.Lookup(tkey); ok {
+			for _, p := range strings.Split(ts, ",") {
+				if p == tval {
+					if val.Kind() != reflect.Ptr && val.CanAddr() {
+						val = val.Addr()
+					}
+					return val.Interface(), nil
+				}
+			}
+		}
+
+		switch val.Kind() {
+		case reflect.Ptr:
+			if val = val.Elem(); val.Kind() == reflect.Struct {
+				if iface, err := StructFieldByTag(val.Interface(), tkey, tval); err == nil {
+					return iface, nil
+				}
+			}
+		case reflect.Struct:
+			if iface, err := StructFieldByTag(val.Interface(), tkey, tval); err == nil {
+				return iface, nil
+			}
+		}
+	}
+	return nil, ErrNotFound
+}
+
+func StructFieldByName(src interface{}, tkey string) (interface{}, error) {
+	sv := reflect.ValueOf(src)
+	if sv.Kind() == reflect.Ptr {
+		sv = sv.Elem()
+	}
+	if sv.Kind() != reflect.Struct {
+		return nil, ErrInvalidStruct
+	}
+
+	typ := sv.Type()
+	for idx := 0; idx < typ.NumField(); idx++ {
+		fld := typ.Field(idx)
+		val := sv.Field(idx)
+		if !val.CanSet() || len(fld.PkgPath) != 0 {
+			continue
+		}
+		if fld.Name == tkey {
+			if val.Kind() != reflect.Ptr && val.CanAddr() {
+				val = val.Addr()
+			}
+			return val.Interface(), nil
+		}
+
+		switch val.Kind() {
+		case reflect.Ptr:
+			if val = val.Elem(); val.Kind() == reflect.Struct {
+				if iface, err := StructFieldByName(val.Interface(), tkey); err == nil {
+					return iface, nil
+				}
+			}
+		case reflect.Struct:
+			if iface, err := StructFieldByName(val.Interface(), tkey); err == nil {
+				return iface, nil
+			}
+		}
+	}
+	return nil, ErrNotFound
+}
+
 // StructFields returns slice of struct fields
-func StructFields(src interface{}) ([]reflect.StructField, error) {
-	var fields []reflect.StructField
+func StructFields(src interface{}) ([]StructField, error) {
+	var fields []StructField
 
 	sv := reflect.ValueOf(src)
 	if sv.Kind() == reflect.Ptr {
@@ -33,14 +124,17 @@ func StructFields(src interface{}) ([]reflect.StructField, error) {
 		if !val.CanSet() || len(fld.PkgPath) != 0 {
 			continue
 		}
-		if val.Kind() == reflect.Struct {
+		switch val.Kind() {
+		case timeKind:
+			fields = append(fields, StructField{Field: fld, Value: val})
+		case reflect.Struct:
 			infields, err := StructFields(val.Interface())
 			if err != nil {
 				return nil, err
 			}
 			fields = append(fields, infields...)
-		} else {
-			fields = append(fields, fld)
+		default:
+			fields = append(fields, StructField{Field: fld, Value: val})
 		}
 	}
 

util/reflect/struct_test.go

@@ -5,6 +5,44 @@ import (
 	"testing"
 )
 
+func TestStructByTag(t *testing.T) {
+	type Str struct {
+		Name []string `json:"name" codec:"flatten"`
+	}
+
+	val := &Str{Name: []string{"first", "second"}}
+
+	iface, err := StructFieldByTag(val, "codec", "flatten")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if v, ok := iface.(*[]string); !ok {
+		t.Fatalf("not *[]string %v", iface)
+	} else if len(*v) != 2 {
+		t.Fatalf("invalid number %v", iface)
+	}
+}
+
+func TestStructByName(t *testing.T) {
+	type Str struct {
+		Name []string `json:"name" codec:"flatten"`
+	}
+
+	val := &Str{Name: []string{"first", "second"}}
+
+	iface, err := StructFieldByName(val, "Name")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if v, ok := iface.(*[]string); !ok {
+		t.Fatalf("not *[]string %v", iface)
+	} else if len(*v) != 2 {
+		t.Fatalf("invalid number %v", iface)
+	}
+}
+
 func TestStructURLValues(t *testing.T) {
 	type Str struct {
 		Str  *Str   `json:"str"`