move memory implementations to core micro repo

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

.github/renovate.json

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

.github/workflows/build.yml

@@ -9,7 +9,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
     - name: setup
-      uses: actions/setup-go@v1
+      uses: actions/setup-go@v2
       with:
         go-version: 1.15
     - name: cache
@@ -49,7 +49,7 @@ jobs:
       - name: checkout
         uses: actions/checkout@v2
       - name: lint
-        uses: golangci/golangci-lint-action@v1
+        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.

.github/workflows/pr.yml

@@ -9,7 +9,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
     - name: setup
-      uses: actions/setup-go@v1
+      uses: actions/setup-go@v2
       with:
         go-version: 1.15
     - name: cache
@@ -49,7 +49,7 @@ jobs:
       - name: checkout
         uses: actions/checkout@v2
       - name: lint
-        uses: golangci/golangci-lint-action@v1
+        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.

api/api.go

@@ -98,6 +98,7 @@ func Encode(e *Endpoint) map[string]string {
 	set("method", strings.Join(e.Method, ","))
 	set("path", strings.Join(e.Path, ","))
 	set("host", strings.Join(e.Host, ","))
+	set("body", e.Body)
 
 	return ep
 }
@@ -118,6 +119,7 @@ func Decode(e metadata.Metadata) *Endpoint {
 	ephost, _ := e.Get("host")
 	ep.Host = []string{ephost}
 	ep.Handler, _ = e.Get("handler")
+	ep.Body, _ = e.Get("body")
 
 	return ep
 }

broker/memory.go

@@ -0,0 +1,247 @@
+package broker
+
+import (
+	"context"
+	"errors"
+	"math/rand"
+	"sync"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/unistack-org/micro/v3/logger"
+	maddr "github.com/unistack-org/micro/v3/util/addr"
+	mnet "github.com/unistack-org/micro/v3/util/net"
+)
+
+type memoryBroker struct {
+	opts Options
+
+	addr string
+	sync.RWMutex
+	connected   bool
+	Subscribers map[string][]*memorySubscriber
+}
+
+type memoryEvent struct {
+	opts    Options
+	topic   string
+	err     error
+	message interface{}
+}
+
+type memorySubscriber struct {
+	id      string
+	topic   string
+	exit    chan bool
+	handler Handler
+	opts    SubscribeOptions
+	ctx     context.Context
+}
+
+func (m *memoryBroker) Options() Options {
+	return m.opts
+}
+
+func (m *memoryBroker) Address() string {
+	return m.addr
+}
+
+func (m *memoryBroker) Connect(ctx context.Context) error {
+	m.Lock()
+	defer m.Unlock()
+
+	if m.connected {
+		return nil
+	}
+
+	// use 127.0.0.1 to avoid scan of all network interfaces
+	addr, err := maddr.Extract("127.0.0.1")
+	if err != nil {
+		return err
+	}
+	i := rand.Intn(20000)
+	// set addr with port
+	addr = mnet.HostPort(addr, 10000+i)
+
+	m.addr = addr
+	m.connected = true
+
+	return nil
+}
+
+func (m *memoryBroker) Disconnect(ctx context.Context) error {
+	m.Lock()
+	defer m.Unlock()
+
+	if !m.connected {
+		return nil
+	}
+
+	m.connected = false
+
+	return nil
+}
+
+func (m *memoryBroker) Init(opts ...Option) error {
+	for _, o := range opts {
+		o(&m.opts)
+	}
+	return nil
+}
+
+func (m *memoryBroker) Publish(ctx context.Context, topic string, msg *Message, opts ...PublishOption) error {
+	m.RLock()
+	if !m.connected {
+		m.RUnlock()
+		return errors.New("not connected")
+	}
+
+	subs, ok := m.Subscribers[topic]
+	m.RUnlock()
+	if !ok {
+		return nil
+	}
+
+	var v interface{}
+	if m.opts.Codec != nil {
+		buf, err := m.opts.Codec.Marshal(msg)
+		if err != nil {
+			return err
+		}
+		v = buf
+	} else {
+		v = msg
+	}
+
+	p := &memoryEvent{
+		topic:   topic,
+		message: v,
+		opts:    m.opts,
+	}
+
+	eh := m.opts.ErrorHandler
+
+	for _, sub := range subs {
+		if err := sub.handler(p); err != nil {
+			p.err = err
+			if sub.opts.ErrorHandler != nil {
+				eh = sub.opts.ErrorHandler
+			}
+			if eh != nil {
+				eh(p)
+			} else {
+				if m.opts.Logger.V(logger.ErrorLevel) {
+					m.opts.Logger.Error(m.opts.Context, err.Error())
+				}
+			}
+			continue
+		}
+	}
+
+	return nil
+}
+
+func (m *memoryBroker) Subscribe(ctx context.Context, topic string, handler Handler, opts ...SubscribeOption) (Subscriber, error) {
+	m.RLock()
+	if !m.connected {
+		m.RUnlock()
+		return nil, errors.New("not connected")
+	}
+	m.RUnlock()
+
+	options := NewSubscribeOptions(opts...)
+
+	id, err := uuid.NewRandom()
+	if err != nil {
+		return nil, err
+	}
+
+	sub := &memorySubscriber{
+		exit:    make(chan bool, 1),
+		id:      id.String(),
+		topic:   topic,
+		handler: handler,
+		opts:    options,
+		ctx:     ctx,
+	}
+
+	m.Lock()
+	m.Subscribers[topic] = append(m.Subscribers[topic], sub)
+	m.Unlock()
+
+	go func() {
+		<-sub.exit
+		m.Lock()
+		var newSubscribers []*memorySubscriber
+		for _, sb := range m.Subscribers[topic] {
+			if sb.id == sub.id {
+				continue
+			}
+			newSubscribers = append(newSubscribers, sb)
+		}
+		m.Subscribers[topic] = newSubscribers
+		m.Unlock()
+	}()
+
+	return sub, nil
+}
+
+func (m *memoryBroker) String() string {
+	return "memory"
+}
+
+func (m *memoryBroker) Name() string {
+	return m.opts.Name
+}
+
+func (m *memoryEvent) Topic() string {
+	return m.topic
+}
+
+func (m *memoryEvent) Message() *Message {
+	switch v := m.message.(type) {
+	case *Message:
+		return v
+	case []byte:
+		msg := &Message{}
+		if err := m.opts.Codec.Unmarshal(v, msg); err != nil {
+			if m.opts.Logger.V(logger.ErrorLevel) {
+				m.opts.Logger.Error(m.opts.Context, "[memory]: failed to unmarshal: %v", err)
+			}
+			return nil
+		}
+		return msg
+	}
+
+	return nil
+}
+
+func (m *memoryEvent) Ack() error {
+	return nil
+}
+
+func (m *memoryEvent) Error() error {
+	return m.err
+}
+
+func (m *memorySubscriber) Options() SubscribeOptions {
+	return m.opts
+}
+
+func (m *memorySubscriber) Topic() string {
+	return m.topic
+}
+
+func (m *memorySubscriber) Unsubscribe(ctx context.Context) error {
+	m.exit <- true
+	return nil
+}
+
+func NewBroker(opts ...Option) Broker {
+	rand.Seed(time.Now().UnixNano())
+
+	return &memoryBroker{
+		opts:        NewOptions(opts...),
+		Subscribers: make(map[string][]*memorySubscriber),
+	}
+}

broker/memory_test.go

@@ -0,0 +1,50 @@
+package broker
+
+import (
+	"context"
+	"fmt"
+	"testing"
+)
+
+func TestMemoryBroker(t *testing.T) {
+	b := NewBroker()
+	ctx := context.Background()
+
+	if err := b.Connect(ctx); err != nil {
+		t.Fatalf("Unexpected connect error %v", err)
+	}
+
+	topic := "test"
+	count := 10
+
+	fn := func(p Event) error {
+		return nil
+	}
+
+	sub, err := b.Subscribe(ctx, topic, fn)
+	if err != nil {
+		t.Fatalf("Unexpected error subscribing %v", err)
+	}
+
+	for i := 0; i < count; i++ {
+		message := &Message{
+			Header: map[string]string{
+				"foo": "bar",
+				"id":  fmt.Sprintf("%d", i),
+			},
+			Body: []byte(`hello world`),
+		}
+
+		if err := b.Publish(ctx, topic, message); err != nil {
+			t.Fatalf("Unexpected error publishing %d", i)
+		}
+	}
+
+	if err := sub.Unsubscribe(ctx); err != nil {
+		t.Fatalf("Unexpected error unsubscribing from %s: %v", topic, err)
+	}
+
+	if err := b.Disconnect(ctx); err != nil {
+		t.Fatalf("Unexpected connect error %v", err)
+	}
+}

broker/noop.go

@@ -1,81 +0,0 @@
-package broker
-
-import "context"
-
-type noopBroker struct {
-	opts Options
-}
-
-type noopSubscriber struct {
-	topic string
-	opts  SubscribeOptions
-}
-
-// NewBroker returns new noop broker
-func NewBroker(opts ...Option) Broker {
-	return &noopBroker{opts: NewOptions(opts...)}
-}
-
-func (n *noopBroker) Name() string {
-	return n.opts.Name
-}
-
-// Init initialize broker
-func (n *noopBroker) Init(opts ...Option) error {
-	for _, o := range opts {
-		o(&n.opts)
-	}
-
-	return nil
-}
-
-// Options returns broker Options
-func (n *noopBroker) Options() Options {
-	return n.opts
-}
-
-// Address returns broker address
-func (n *noopBroker) Address() string {
-	return ""
-}
-
-// Connect connects to broker
-func (n *noopBroker) Connect(ctx context.Context) error {
-	return nil
-}
-
-// Disconnect disconnects from broker
-func (n *noopBroker) Disconnect(ctx context.Context) error {
-	return nil
-}
-
-// Publish publishes message to broker
-func (n *noopBroker) Publish(ctx context.Context, topic string, m *Message, opts ...PublishOption) error {
-	return nil
-}
-
-// Subscribe subscribes to broker topic
-func (n *noopBroker) Subscribe(ctx context.Context, topic string, h Handler, opts ...SubscribeOption) (Subscriber, error) {
-	options := NewSubscribeOptions(opts...)
-	return &noopSubscriber{topic: topic, opts: options}, nil
-}
-
-// String return broker string representation
-func (n *noopBroker) String() string {
-	return "noop"
-}
-
-// Options returns subscriber options
-func (n *noopSubscriber) Options() SubscribeOptions {
-	return n.opts
-}
-
-// TOpic returns subscriber topic
-func (n *noopSubscriber) Topic() string {
-	return n.topic
-}
-
-// Unsubscribe unsbscribes from broker topic
-func (n *noopSubscriber) Unsubscribe(ctx context.Context) error {
-	return nil
-}

client/noop.go

@@ -46,7 +46,16 @@ type noopRequest struct {
 
 // NewClient returns new noop client
 func NewClient(opts ...Option) Client {
-	return &noopClient{opts: NewOptions(opts...)}
+	nc := &noopClient{opts: NewOptions(opts...)}
+	// wrap in reverse
+
+	c := Client(nc)
+
+	for i := len(nc.opts.Wrappers); i > 0; i-- {
+		c = nc.opts.Wrappers[i-1](c)
+	}
+
+	return c
 }
 
 func (n *noopClient) Name() string {
@@ -187,7 +196,7 @@ func (n *noopClient) Publish(ctx context.Context, p Message, opts ...PublishOpti
 
 	options := NewPublishOptions(opts...)
 
-	md, ok := metadata.FromContext(ctx)
+	md, ok := metadata.FromOutgoingContext(ctx)
 	if !ok {
 		md = metadata.New(0)
 	}

codec/noop.go

@@ -1,6 +1,7 @@
 package codec
 
 import (
+	"encoding/json"
 	"io"
 	"io/ioutil"
 )
@@ -40,7 +41,7 @@ func (c *noopCodec) ReadBody(conn io.Reader, b interface{}) error {
 	case *Frame:
 		v.Data = buf
 	default:
-		return ErrInvalidMessage
+		return json.Unmarshal(buf, v)
 	}
 
 	return nil
@@ -64,7 +65,11 @@ func (c *noopCodec) Write(conn io.Writer, m *Message, b interface{}) error {
 	case []byte:
 		v = vb
 	default:
-		return ErrInvalidMessage
+		var err error
+		v, err = json.Marshal(vb)
+		if err != nil {
+			return err
+		}
 	}
 	_, err := conn.Write(v)
 	return err
@@ -98,30 +103,34 @@ func (c *noopCodec) Marshal(v interface{}) ([]byte, error) {
 	case *Message:
 		return ve.Body, nil
 	}
-	return nil, ErrInvalidMessage
+
+	return json.Marshal(v)
 }
 
 func (c *noopCodec) Unmarshal(d []byte, v interface{}) error {
-	var err error
 	if v == nil {
 		return nil
 	}
 	switch ve := v.(type) {
 	case string:
 		ve = string(d)
+		return nil
 	case *string:
 		*ve = string(d)
+		return nil
 	case []byte:
 		ve = d
+		return nil
 	case *[]byte:
 		*ve = d
+		return nil
 	case *Frame:
 		ve.Data = d
+		return nil
 	case *Message:
 		ve.Body = d
-	default:
-		err = ErrInvalidMessage
+		return nil
 	}
 
-	return err
+	return json.Unmarshal(d, v)
 }

config/config.go

@@ -22,6 +22,7 @@ var (
 
 // Config is an interface abstraction for dynamic configuration
 type Config interface {
+	Name() string
 	// Init the config
 	Init(opts ...Option) error
 	// Options in the config

config/default.go

@@ -251,6 +251,10 @@ func (c *defaultConfig) String() string {
 	return "default"
 }
 
+func (c *defaultConfig) Name() string {
+	return c.opts.Name
+}
+
 func NewConfig(opts ...Option) Config {
 	options := NewOptions(opts...)
 	if len(options.StructTag) == 0 {

events/events.go

@@ -1,49 +0,0 @@
-// Package events contains interfaces for managing events within distributed systems
-package events
-
-import (
-	"context"
-	"encoding/json"
-	"errors"
-	"time"
-
-	"github.com/unistack-org/micro/v3/metadata"
-)
-
-var (
-	// ErrMissingTopic is returned if a blank topic was provided to publish
-	ErrMissingTopic = errors.New("Missing topic")
-	// ErrEncodingMessage is returned from publish if there was an error encoding the message option
-	ErrEncodingMessage = errors.New("Error encoding message")
-)
-
-// Stream of events
-type Stream interface {
-	Publish(ctx context.Context, topic string, msg interface{}, opts ...PublishOption) error
-	Subscribe(ctx context.Context, topic string, opts ...SubscribeOption) (<-chan Event, error)
-}
-
-// Store of events
-type Store interface {
-	Read(ctx context.Context, opts ...ReadOption) ([]*Event, error)
-	Write(ctx context.Context, event *Event, opts ...WriteOption) error
-}
-
-// Event is the object returned by the broker when you subscribe to a topic
-type Event struct {
-	// ID to uniquely identify the event
-	ID string
-	// Topic of event, e.g. "register.service.created"
-	Topic string
-	// Timestamp of the event
-	Timestamp time.Time
-	// Metadata contains the encoded event was indexed by
-	Metadata metadata.Metadata
-	// Payload contains the encoded message
-	Payload []byte
-}
-
-// Unmarshal the events message into an object
-func (e *Event) Unmarshal(v interface{}) error {
-	return json.Unmarshal(e.Payload, v)
-}

events/options.go

@@ -1,124 +0,0 @@
-package events
-
-import (
-	"time"
-
-	"github.com/unistack-org/micro/v3/metadata"
-)
-
-// PublishOptions contains all the options which can be provided when publishing an event
-type PublishOptions struct {
-	// Metadata contains any keys which can be used to query the data, for example a customer id
-	Metadata metadata.Metadata
-	// Timestamp to set for the event, if the timestamp is a zero value, the current time will be used
-	Timestamp time.Time
-}
-
-// PublishOption sets attributes on PublishOptions
-type PublishOption func(o *PublishOptions)
-
-// WithMetadata sets the Metadata field on PublishOptions
-func WithMetadata(md metadata.Metadata) PublishOption {
-	return func(o *PublishOptions) {
-		o.Metadata = metadata.Copy(md)
-	}
-}
-
-// WithTimestamp sets the timestamp field on PublishOptions
-func WithTimestamp(t time.Time) PublishOption {
-	return func(o *PublishOptions) {
-		o.Timestamp = t
-	}
-}
-
-// SubscribeOptions contains all the options which can be provided when subscribing to a topic
-type SubscribeOptions struct {
-	// Queue is the name of the subscribers queue, if two subscribers have the same queue the message
-	// should only be published to one of them
-	Queue string
-	// StartAtTime is the time from which the messages should be consumed from. If not provided then
-	// the messages will be consumed starting from the moment the Subscription starts.
-	StartAtTime time.Time
-}
-
-// SubscribeOption sets attributes on SubscribeOptions
-type SubscribeOption func(o *SubscribeOptions)
-
-// WithQueue sets the Queue fielf on SubscribeOptions to the value provided
-func WithQueue(q string) SubscribeOption {
-	return func(o *SubscribeOptions) {
-		o.Queue = q
-	}
-}
-
-// WithStartAtTime sets the StartAtTime field on SubscribeOptions to the value provided
-func WithStartAtTime(t time.Time) SubscribeOption {
-	return func(o *SubscribeOptions) {
-		o.StartAtTime = t
-	}
-}
-
-// WriteOptions contains all the options which can be provided when writing an event to a store
-type WriteOptions struct {
-	// TTL is the duration the event should be recorded for, a zero value TTL indicates the event should
-	// be stored indefinitely
-	TTL time.Duration
-}
-
-// WriteOption sets attributes on WriteOptions
-type WriteOption func(o *WriteOptions)
-
-// WithTTL sets the TTL attribute on WriteOptions
-func WithTTL(d time.Duration) WriteOption {
-	return func(o *WriteOptions) {
-		o.TTL = d
-	}
-}
-
-// ReadOptions contains all the options which can be provided when reading events from a store
-type ReadOptions struct {
-	// Topic to read events from, if no topic is provided events from all topics will be returned
-	Topic string
-	// Query to filter the results using. The store will query the metadata provided when the event
-	// was written to the store
-	Query map[string]string
-	// Limit the number of results to return
-	Limit int
-	// Offset the results by this number, useful for paginated queries
-	Offset int
-}
-
-// ReadOption sets attributes on ReadOptions
-type ReadOption func(o *ReadOptions)
-
-// ReadTopic sets the topic attribute on ReadOptions
-func ReadTopic(t string) ReadOption {
-	return func(o *ReadOptions) {
-		o.Topic = t
-	}
-}
-
-// ReadFilter sets a key and value in the query
-func ReadFilter(key, value string) ReadOption {
-	return func(o *ReadOptions) {
-		if o.Query == nil {
-			o.Query = map[string]string{key: value}
-		} else {
-			o.Query[key] = value
-		}
-	}
-}
-
-// ReadLimit sets the limit attribute on ReadOptions
-func ReadLimit(l int) ReadOption {
-	return func(o *ReadOptions) {
-		o.Limit = 1
-	}
-}
-
-// ReadOffset sets the offset attribute on ReadOptions
-func ReadOffset(l int) ReadOption {
-	return func(o *ReadOptions) {
-		o.Offset = 1
-	}
-}

go.mod

@@ -7,15 +7,16 @@ require (
 	github.com/dgrijalva/jwt-go v3.2.0+incompatible
 	github.com/ef-ds/deque v1.0.4
 	github.com/golang/protobuf v1.4.3
-	github.com/google/uuid v1.1.5
+	github.com/google/uuid v1.2.0
 	github.com/imdario/mergo v0.3.11
 	github.com/kr/text v0.2.0 // indirect
-	github.com/miekg/dns v1.1.35
+	github.com/miekg/dns v1.1.38
 	github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e // indirect
 	github.com/oxtoacart/bpool v0.0.0-20190530202638-03653db5a59c
+	github.com/patrickmn/go-cache v2.1.0+incompatible
 	github.com/stretchr/testify v1.7.0
 	golang.org/x/crypto v0.0.0-20201221181555-eec23a3978ad // indirect
-	golang.org/x/net v0.0.0-20201224014010-6772e930b67b
+	golang.org/x/net v0.0.0-20210119194325-5f4716e94777
 	golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e // indirect
 	google.golang.org/protobuf v1.25.0
 	gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f // indirect

go.sum

@@ -31,20 +31,22 @@ github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMyw
 github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.0 h1:/QaMHBdZ26BB3SSst0Iwl10Epc+xhTquomWX0oZEB6w=
 github.com/google/go-cmp v0.5.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
-github.com/google/uuid v1.1.5 h1:kxhtnfFVi+rYdOALN0B3k9UT86zVJKfBimRaciULW4I=
-github.com/google/uuid v1.1.5/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/google/uuid v1.2.0 h1:qJYtXnJRWmpe7m/3XlyhrsLrEURqHRM2kxzoxXqyUDs=
+github.com/google/uuid v1.2.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/imdario/mergo v0.3.11 h1:3tnifQM4i+fbajXKBHXWEH+KvNHqojZ778UH75j3bGA=
 github.com/imdario/mergo v0.3.11/go.mod h1:jmQim1M+e3UYxmgPu/WyfjB3N3VflVyUjjjwH0dnCYA=
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
-github.com/miekg/dns v1.1.35 h1:oTfOaDH+mZkdcgdIjH6yBajRGtIwcwcaR+rt23ZSrJs=
-github.com/miekg/dns v1.1.35/go.mod h1:KNUDUusw/aVsxyTYZM1oqvCicbwhgbNgztCETuNZ7xM=
+github.com/miekg/dns v1.1.38 h1:MtIY+fmHUVVgv1AXzmKMWcwdCYxTRPG1EDjpqF4RCEw=
+github.com/miekg/dns v1.1.38/go.mod h1:KNUDUusw/aVsxyTYZM1oqvCicbwhgbNgztCETuNZ7xM=
 github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e h1:fD57ERR4JtEqsWbfPhv4DMiApHyliiK5xCTNVSPiaAs=
 github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno=
 github.com/oxtoacart/bpool v0.0.0-20190530202638-03653db5a59c h1:rp5dCmg/yLR3mgFuSOe4oEnDDmGLROTvMragMUXpTQw=
 github.com/oxtoacart/bpool v0.0.0-20190530202638-03653db5a59c/go.mod h1:X07ZCGwUbLaax7L0S3Tw4hpejzu63ZrrQiUe6W0hcy0=
+github.com/patrickmn/go-cache v2.1.0+incompatible h1:HRMgzkcYKYpi3C8ajMPV8OFXaaRUnok+kx1WdO15EQc=
+github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
@@ -52,7 +54,6 @@ github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
-github.com/unistack-org/micro v1.18.0 h1:EbFiII0bKV0Xcua7o6J30MFmm4/g0Hv3ECOKzsUBihU=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20201221181555-eec23a3978ad h1:DN0cp81fZ3njFcrLCytUHRSUkqBjfTo4Tx9RJTWs0EY=
@@ -69,8 +70,8 @@ golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20190923162816-aa69164e4478/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
-golang.org/x/net v0.0.0-20201224014010-6772e930b67b h1:iFwSg7t5GZmB/Q5TjiEAsdoLDrdJRC1RiF2WhuV29Qw=
-golang.org/x/net v0.0.0-20201224014010-6772e930b67b/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
+golang.org/x/net v0.0.0-20210119194325-5f4716e94777 h1:003p0dJM77cxMSyCPFphvZf/Y5/NXf5fzg6ufd1/Oew=
+golang.org/x/net v0.0.0-20210119194325-5f4716e94777/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=

metadata/context.go

@@ -5,35 +5,104 @@ import (
 	"context"
 )
 
+type mdIncomingKey struct{}
+type mdOutgoingKey struct{}
+type mdKey struct{}
+
+// FromIncomingContext returns metadata from incoming ctx
+// returned metadata shoud not be modified or race condition happens
+func FromIncomingContext(ctx context.Context) (Metadata, bool) {
+	if ctx == nil {
+		return nil, false
+	}
+	md, ok := ctx.Value(mdIncomingKey{}).(*rawMetadata)
+	if !ok {
+		return nil, false
+	}
+	return md.md, ok
+}
+
+// FromOutgoingContext returns metadata from outgoing ctx
+// returned metadata shoud not be modified or race condition happens
+func FromOutgoingContext(ctx context.Context) (Metadata, bool) {
+	if ctx == nil {
+		return nil, false
+	}
+	md, ok := ctx.Value(mdOutgoingKey{}).(*rawMetadata)
+	if !ok {
+		return nil, false
+	}
+	return md.md, ok
+}
+
 // FromContext returns metadata from the given context
+// returned metadata shoud not be modified or race condition happens
+//
+// Deprecated: use FromIncomingContext or FromOutgoingContext
 func FromContext(ctx context.Context) (Metadata, bool) {
 	if ctx == nil {
 		return nil, false
 	}
-	md, ok := ctx.Value(metadataKey{}).(Metadata)
+	md, ok := ctx.Value(mdKey{}).(*rawMetadata)
 	if !ok {
-		return nil, ok
+		return nil, false
 	}
-	nmd := Copy(md)
-	return nmd, ok
+	return md.md, ok
 }
 
 // NewContext creates a new context with the given metadata
+//
+// Deprecated: use NewIncomingContext or NewOutgoingContext
 func NewContext(ctx context.Context, md Metadata) context.Context {
 	if ctx == nil {
 		ctx = context.Background()
 	}
-	return context.WithValue(ctx, metadataKey{}, Copy(md))
+	ctx = context.WithValue(ctx, mdKey{}, &rawMetadata{md})
+	ctx = context.WithValue(ctx, mdIncomingKey{}, &rawMetadata{})
+	ctx = context.WithValue(ctx, mdOutgoingKey{}, &rawMetadata{})
+	return ctx
 }
 
-// MergeContext merges metadata to existing metadata, overwriting if specified
-func MergeContext(ctx context.Context, pmd Metadata, overwrite bool) context.Context {
+// SetOutgoingContext modify outgoing context with given metadata
+func SetOutgoingContext(ctx context.Context, md Metadata) bool {
+	if ctx == nil {
+		return false
+	}
+	if omd, ok := ctx.Value(mdOutgoingKey{}).(*rawMetadata); ok {
+		omd.md = md
+		return true
+	}
+	return false
+}
+
+// SetIncomingContext modify incoming context with given metadata
+func SetIncomingContext(ctx context.Context, md Metadata) bool {
+	if ctx == nil {
+		return false
+	}
+	if omd, ok := ctx.Value(mdIncomingKey{}).(*rawMetadata); ok {
+		omd.md = md
+		return true
+	}
+	return false
+}
+
+// NewIncomingContext creates a new context with incoming metadata attached
+func NewIncomingContext(ctx context.Context, md Metadata) context.Context {
 	if ctx == nil {
 		ctx = context.Background()
 	}
-	md, ok := FromContext(ctx)
-	if !ok {
-		return context.WithValue(ctx, metadataKey{}, Copy(pmd))
-	}
-	return context.WithValue(ctx, metadataKey{}, Merge(md, pmd, overwrite))
+	ctx = context.WithValue(ctx, mdIncomingKey{}, &rawMetadata{md})
+	ctx = context.WithValue(ctx, mdOutgoingKey{}, &rawMetadata{})
+	return ctx
+}
+
+// NewOutgoingContext creates a new context with outcoming metadata attached
+func NewOutgoingContext(ctx context.Context, md Metadata) context.Context {
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	ctx = context.WithValue(ctx, mdOutgoingKey{}, &rawMetadata{md})
+	ctx = context.WithValue(ctx, mdIncomingKey{}, &rawMetadata{})
+	return ctx
 }

metadata/metadata.go

@@ -2,7 +2,6 @@
 package metadata
 
 import (
-	"context"
 	"net/textproto"
 	"sort"
 )
@@ -12,16 +11,18 @@ var (
 	HeaderPrefix = "Micro-"
 )
 
-type metadataKey struct{}
-
 // Metadata is our way of representing request headers internally.
 // They're used at the RPC level and translate back and forth
 // from Transport headers.
 type Metadata map[string]string
 
+type rawMetadata struct {
+	md Metadata
+}
+
 var (
-	// DefaultMetadataSize used when need to init new Metadata
-	DefaultMetadataSize = 6
+	// defaultMetadataSize used when need to init new Metadata
+	defaultMetadataSize = 2
 )
 
 type Iterator struct {
@@ -72,12 +73,9 @@ func (md Metadata) Set(key, val string) {
 // Del is used to remove value from metadata
 func (md Metadata) Del(key string) {
 	// fast path
-	if _, ok := md[key]; ok {
-		delete(md, key)
-	} else {
-		// slow path
-		delete(md, textproto.CanonicalMIMEHeaderKey(key))
-	}
+	delete(md, key)
+	// slow path
+	delete(md, textproto.CanonicalMIMEHeaderKey(key))
 }
 
 // Copy makes a copy of the metadata
@@ -89,39 +87,10 @@ func Copy(md Metadata) Metadata {
 	return nmd
 }
 
-// Del deletes key from metadata
-func Del(ctx context.Context, key string) context.Context {
-	md, ok := FromContext(ctx)
-	if !ok {
-		md = New(0)
-	}
-	md.Del(key)
-	return context.WithValue(ctx, metadataKey{}, md)
-}
-
-// Set add key with val to metadata
-func Set(ctx context.Context, key, val string) context.Context {
-	md, ok := FromContext(ctx)
-	if !ok {
-		md = New(0)
-	}
-	md.Set(key, val)
-	return context.WithValue(ctx, metadataKey{}, md)
-}
-
-// Get returns a single value from metadata in the context
-func Get(ctx context.Context, key string) (string, bool) {
-	md, ok := FromContext(ctx)
-	if !ok {
-		return "", ok
-	}
-	return md.Get(key)
-}
-
 // New return new sized metadata
 func New(size int) Metadata {
 	if size == 0 {
-		size = DefaultMetadataSize
+		size = defaultMetadataSize
 	}
 	return make(Metadata, size)
 }

metadata/metadata_test.go

@@ -3,10 +3,31 @@ package metadata
 import (
 	"context"
 	"fmt"
-	"reflect"
 	"testing"
 )
 
+func testCtx(ctx context.Context) {
+	md := New(2)
+	md.Set("Key1", "Val1_new")
+	md.Set("Key3", "Val3")
+	SetOutgoingContext(ctx, md)
+}
+
+func TestPassing(t *testing.T) {
+	ctx := context.TODO()
+	md1 := New(2)
+	md1.Set("Key1", "Val1")
+	md1.Set("Key2", "Val2")
+
+	ctx = NewIncomingContext(ctx, md1)
+	testCtx(ctx)
+	md, ok := FromOutgoingContext(ctx)
+	if !ok {
+		t.Fatalf("missing metadata from outgoing context")
+	}
+	fmt.Printf("%#+v\n", md)
+}
+
 func TestMerge(t *testing.T) {
 	omd := Metadata{
 		"key1": "val1",
@@ -32,26 +53,27 @@ func TestIterator(t *testing.T) {
 	var k, v string
 
 	for iter.Next(&k, &v) {
-		fmt.Printf("k: %s, v: %s\n", k, v)
+		//fmt.Printf("k: %s, v: %s\n", k, v)
 	}
 }
 
 func TestMedataCanonicalKey(t *testing.T) {
-	ctx := Set(context.TODO(), "x-request-id", "12345")
-	v, ok := Get(ctx, "x-request-id")
+	md := New(1)
+	md.Set("x-request-id", "12345")
+	v, ok := md.Get("x-request-id")
 	if !ok {
 		t.Fatalf("failed to get x-request-id")
 	} else if v != "12345" {
 		t.Fatalf("invalid metadata value: %s != %s", "12345", v)
 	}
 
-	v, ok = Get(ctx, "X-Request-Id")
+	v, ok = md.Get("X-Request-Id")
 	if !ok {
 		t.Fatalf("failed to get x-request-id")
 	} else if v != "12345" {
 		t.Fatalf("invalid metadata value: %s != %s", "12345", v)
 	}
-	v, ok = Get(ctx, "X-Request-ID")
+	v, ok = md.Get("X-Request-ID")
 	if !ok {
 		t.Fatalf("failed to get x-request-id")
 	} else if v != "12345" {
@@ -61,9 +83,11 @@ func TestMedataCanonicalKey(t *testing.T) {
 }
 
 func TestMetadataSet(t *testing.T) {
-	ctx := Set(context.TODO(), "Key", "val")
+	md := New(1)
 
-	val, ok := Get(ctx, "Key")
+	md.Set("Key", "val")
+
+	val, ok := md.Get("Key")
 	if !ok {
 		t.Fatal("key Key not found")
 	}
@@ -78,15 +102,8 @@ func TestMetadataDelete(t *testing.T) {
 		"Baz": "empty",
 	}
 
-	ctx := NewContext(context.TODO(), md)
-	ctx = Del(ctx, "Baz")
-
-	emd, ok := FromContext(ctx)
-	if !ok {
-		t.Fatal("key Key not found")
-	}
-
-	_, ok = emd["Baz"]
+	md.Del("Baz")
+	_, ok := md.Get("Baz")
 	if ok {
 		t.Fatal("key Baz not deleted")
 	}
@@ -137,42 +154,3 @@ func TestMetadataContext(t *testing.T) {
 		t.Errorf("Expected metadata length 1 got %d", i)
 	}
 }
-
-func TestMergeContext(t *testing.T) {
-	type args struct {
-		existing  Metadata
-		append    Metadata
-		overwrite bool
-	}
-	tests := []struct {
-		name string
-		args args
-		want Metadata
-	}{
-		{
-			name: "matching key, overwrite false",
-			args: args{
-				existing:  Metadata{"Foo": "bar", "Sumo": "demo"},
-				append:    Metadata{"Sumo": "demo2"},
-				overwrite: false,
-			},
-			want: Metadata{"Foo": "bar", "Sumo": "demo"},
-		},
-		{
-			name: "matching key, overwrite true",
-			args: args{
-				existing:  Metadata{"Foo": "bar", "Sumo": "demo"},
-				append:    Metadata{"Sumo": "demo2"},
-				overwrite: true,
-			},
-			want: Metadata{"Foo": "bar", "Sumo": "demo2"},
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			if got, _ := FromContext(MergeContext(NewContext(context.TODO(), tt.args.existing), tt.args.append, tt.args.overwrite)); !reflect.DeepEqual(got, tt.want) {
-				t.Errorf("MergeContext() = %v, want %v", got, tt.want)
-			}
-		})
-	}
-}

network/transport/memory.go

@@ -0,0 +1,263 @@
+package transport
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"math/rand"
+	"net"
+	"sync"
+	"time"
+
+	maddr "github.com/unistack-org/micro/v3/util/addr"
+	mnet "github.com/unistack-org/micro/v3/util/net"
+)
+
+type memorySocket struct {
+	recv chan *Message
+	send chan *Message
+	// sock exit
+	exit chan bool
+	// listener exit
+	lexit chan bool
+
+	local  string
+	remote string
+
+	// for send/recv transport.Timeout
+	timeout time.Duration
+	ctx     context.Context
+	sync.RWMutex
+}
+
+type memoryClient struct {
+	*memorySocket
+	opts DialOptions
+}
+
+type memoryListener struct {
+	addr  string
+	exit  chan bool
+	conn  chan *memorySocket
+	lopts ListenOptions
+	topts Options
+	sync.RWMutex
+	ctx context.Context
+}
+
+type memoryTransport struct {
+	opts Options
+	sync.RWMutex
+	listeners map[string]*memoryListener
+}
+
+func (ms *memorySocket) Recv(m *Message) error {
+	ms.RLock()
+	defer ms.RUnlock()
+
+	ctx := ms.ctx
+	if ms.timeout > 0 {
+		var cancel context.CancelFunc
+		ctx, cancel = context.WithTimeout(ms.ctx, ms.timeout)
+		defer cancel()
+	}
+
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-ms.exit:
+		return errors.New("connection closed")
+	case <-ms.lexit:
+		return errors.New("server connection closed")
+	case cm := <-ms.recv:
+		*m = *cm
+	}
+	return nil
+}
+
+func (ms *memorySocket) Local() string {
+	return ms.local
+}
+
+func (ms *memorySocket) Remote() string {
+	return ms.remote
+}
+
+func (ms *memorySocket) Send(m *Message) error {
+	ms.RLock()
+	defer ms.RUnlock()
+
+	ctx := ms.ctx
+	if ms.timeout > 0 {
+		var cancel context.CancelFunc
+		ctx, cancel = context.WithTimeout(ms.ctx, ms.timeout)
+		defer cancel()
+	}
+
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-ms.exit:
+		return errors.New("connection closed")
+	case <-ms.lexit:
+		return errors.New("server connection closed")
+	case ms.send <- m:
+	}
+	return nil
+}
+
+func (ms *memorySocket) Close() error {
+	ms.Lock()
+	defer ms.Unlock()
+	select {
+	case <-ms.exit:
+		return nil
+	default:
+		close(ms.exit)
+	}
+	return nil
+}
+
+func (m *memoryListener) Addr() string {
+	return m.addr
+}
+
+func (m *memoryListener) Close() error {
+	m.Lock()
+	defer m.Unlock()
+	select {
+	case <-m.exit:
+		return nil
+	default:
+		close(m.exit)
+	}
+	return nil
+}
+
+func (m *memoryListener) Accept(fn func(Socket)) error {
+	for {
+		select {
+		case <-m.exit:
+			return nil
+		case c := <-m.conn:
+			go fn(&memorySocket{
+				lexit:   c.lexit,
+				exit:    c.exit,
+				send:    c.recv,
+				recv:    c.send,
+				local:   c.Remote(),
+				remote:  c.Local(),
+				timeout: m.topts.Timeout,
+				ctx:     m.topts.Context,
+			})
+		}
+	}
+}
+
+func (m *memoryTransport) Dial(ctx context.Context, addr string, opts ...DialOption) (Client, error) {
+	m.RLock()
+	defer m.RUnlock()
+
+	listener, ok := m.listeners[addr]
+	if !ok {
+		return nil, errors.New("could not dial " + addr)
+	}
+
+	options := NewDialOptions(opts...)
+
+	client := &memoryClient{
+		&memorySocket{
+			send:    make(chan *Message),
+			recv:    make(chan *Message),
+			exit:    make(chan bool),
+			lexit:   listener.exit,
+			local:   addr,
+			remote:  addr,
+			timeout: m.opts.Timeout,
+			ctx:     m.opts.Context,
+		},
+		options,
+	}
+
+	// pseudo connect
+	select {
+	case <-listener.exit:
+		return nil, errors.New("connection error")
+	case listener.conn <- client.memorySocket:
+	}
+
+	return client, nil
+}
+
+func (m *memoryTransport) Listen(ctx context.Context, addr string, opts ...ListenOption) (Listener, error) {
+	m.Lock()
+	defer m.Unlock()
+
+	options := NewListenOptions(opts...)
+
+	host, port, err := net.SplitHostPort(addr)
+	if err != nil {
+		return nil, err
+	}
+
+	addr, err = maddr.Extract(host)
+	if err != nil {
+		return nil, err
+	}
+
+	// if zero port then randomly assign one
+	if len(port) > 0 && port == "0" {
+		i := rand.Intn(20000)
+		port = fmt.Sprintf("%d", 10000+i)
+	}
+
+	// set addr with port
+	addr = mnet.HostPort(addr, port)
+
+	if _, ok := m.listeners[addr]; ok {
+		return nil, errors.New("already listening on " + addr)
+	}
+
+	listener := &memoryListener{
+		lopts: options,
+		topts: m.opts,
+		addr:  addr,
+		conn:  make(chan *memorySocket),
+		exit:  make(chan bool),
+		ctx:   m.opts.Context,
+	}
+
+	m.listeners[addr] = listener
+
+	return listener, nil
+}
+
+func (m *memoryTransport) Init(opts ...Option) error {
+	for _, o := range opts {
+		o(&m.opts)
+	}
+	return nil
+}
+
+func (m *memoryTransport) Options() Options {
+	return m.opts
+}
+
+func (m *memoryTransport) String() string {
+	return "memory"
+}
+
+func (m *memoryTransport) Name() string {
+	return m.opts.Name
+}
+
+func NewTransport(opts ...Option) Transport {
+	options := NewOptions(opts...)
+
+	rand.Seed(time.Now().UnixNano())
+
+	return &memoryTransport{
+		opts:      options,
+		listeners: make(map[string]*memoryListener),
+	}
+}

network/transport/memory_test.go

@@ -0,0 +1,93 @@
+package transport
+
+import (
+	"context"
+	"os"
+	"testing"
+)
+
+func TestMemoryTransport(t *testing.T) {
+	tr := NewTransport()
+	ctx := context.Background()
+	// bind / listen
+	l, err := tr.Listen(ctx, "127.0.0.1:8080")
+	if err != nil {
+		t.Fatalf("Unexpected error listening %v", err)
+	}
+	defer l.Close()
+
+	// accept
+	go func() {
+		if err := l.Accept(func(sock Socket) {
+			for {
+				var m Message
+				if err := sock.Recv(&m); err != nil {
+					return
+				}
+				if len(os.Getenv("INTEGRATION_TESTS")) == 0 {
+					t.Logf("Server Received %s", string(m.Body))
+				}
+				if err := sock.Send(&Message{
+					Body: []byte(`pong`),
+				}); err != nil {
+					return
+				}
+			}
+		}); err != nil {
+			t.Fatalf("Unexpected error accepting %v", err)
+		}
+	}()
+
+	// dial
+	c, err := tr.Dial(ctx, "127.0.0.1:8080")
+	if err != nil {
+		t.Fatalf("Unexpected error dialing %v", err)
+	}
+	defer c.Close()
+
+	// send <=> receive
+	for i := 0; i < 3; i++ {
+		if err := c.Send(&Message{
+			Body: []byte(`ping`),
+		}); err != nil {
+			return
+		}
+		var m Message
+		if err := c.Recv(&m); err != nil {
+			return
+		}
+		if len(os.Getenv("INTEGRATION_TESTS")) == 0 {
+			t.Logf("Client Received %s", string(m.Body))
+		}
+	}
+
+}
+
+func TestListener(t *testing.T) {
+	tr := NewTransport()
+	ctx := context.Background()
+	// bind / listen on random port
+	l, err := tr.Listen(ctx, ":0")
+	if err != nil {
+		t.Fatalf("Unexpected error listening %v", err)
+	}
+	defer l.Close()
+
+	// try again
+	l2, err := tr.Listen(ctx, ":0")
+	if err != nil {
+		t.Fatalf("Unexpected error listening %v", err)
+	}
+	defer l2.Close()
+
+	// now make sure it still fails
+	l3, err := tr.Listen(ctx, ":8080")
+	if err != nil {
+		t.Fatalf("Unexpected error listening %v", err)
+	}
+	defer l3.Close()
+
+	if _, err := tr.Listen(ctx, ":8080"); err == nil {
+		t.Fatal("Expected error binding to :8080 got nil")
+	}
+}

network/transport/noop.go

@@ -1,77 +0,0 @@
-package transport
-
-import "context"
-
-type noopTransport struct {
-	opts Options
-}
-
-// NewTransport creates new noop transport
-func NewTransport(opts ...Option) Transport {
-	return &noopTransport{opts: NewOptions(opts...)}
-}
-
-func (t *noopTransport) Init(opts ...Option) error {
-	for _, o := range opts {
-		o(&t.opts)
-	}
-	return nil
-}
-
-func (t *noopTransport) Options() Options {
-	return t.opts
-}
-
-func (t *noopTransport) Dial(ctx context.Context, addr string, opts ...DialOption) (Client, error) {
-	options := NewDialOptions(opts...)
-	return &noopClient{opts: options}, nil
-}
-
-func (t *noopTransport) Listen(ctx context.Context, addr string, opts ...ListenOption) (Listener, error) {
-	options := NewListenOptions(opts...)
-	return &noopListener{opts: options}, nil
-}
-
-func (t *noopTransport) String() string {
-	return "noop"
-}
-
-type noopClient struct {
-	opts DialOptions
-}
-
-func (c *noopClient) Close() error {
-	return nil
-}
-
-func (c *noopClient) Local() string {
-	return ""
-}
-
-func (c *noopClient) Remote() string {
-	return ""
-}
-
-func (c *noopClient) Recv(*Message) error {
-	return nil
-}
-
-func (c *noopClient) Send(*Message) error {
-	return nil
-}
-
-type noopListener struct {
-	opts ListenOptions
-}
-
-func (l *noopListener) Addr() string {
-	return ""
-}
-
-func (l *noopListener) Accept(fn func(Socket)) error {
-	return nil
-}
-
-func (l *noopListener) Close() error {
-	return nil
-}

options.go

@@ -134,6 +134,14 @@ func BrokerServer(n string) BrokerOption {
 	}
 }
 
+// Client to be used for service
+func Client(c ...client.Client) Option {
+	return func(o *Options) error {
+		o.Clients = c
+		return nil
+	}
+}
+
 // Clients to be used for service
 func Clients(c ...client.Client) Option {
 	return func(o *Options) error {
@@ -161,6 +169,14 @@ func Profile(p profile.Profile) Option {
 }
 */
 
+// Server to be used for service
+func Server(s ...server.Server) Option {
+	return func(o *Options) error {
+		o.Servers = s
+		return nil
+	}
+}
+
 // Servers to be used for service
 func Servers(s ...server.Server) Option {
 	return func(o *Options) error {
@@ -169,6 +185,14 @@ func Servers(s ...server.Server) Option {
 	}
 }
 
+// Store sets the store to use
+func Store(s ...store.Store) Option {
+	return func(o *Options) error {
+		o.Stores = s
+		return nil
+	}
+}
+
 // Stores sets the store to use
 func Stores(s ...store.Store) Option {
 	return func(o *Options) error {
@@ -275,6 +299,14 @@ func LoggerServer(n string) LoggerOption {
 }
 */
 
+// Meter set the meter to use
+func Meter(m ...meter.Meter) Option {
+	return func(o *Options) error {
+		o.Meters = m
+		return nil
+	}
+}
+
 // Meters set the meter to use
 func Meters(m ...meter.Meter) Option {
 	return func(o *Options) error {
@@ -450,6 +482,14 @@ func Auth(a auth.Auth) Option {
 }
 */
 
+// Config sets the config for the service
+func Config(c ...config.Config) Option {
+	return func(o *Options) error {
+		o.Configs = c
+		return nil
+	}
+}
+
 // Configs sets the configs for the service
 func Configs(c ...config.Config) Option {
 	return func(o *Options) error {

register/memory.go

@@ -0,0 +1,541 @@
+package register
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/unistack-org/micro/v3/logger"
+)
+
+var (
+	sendEventTime = 10 * time.Millisecond
+	ttlPruneTime  = time.Second
+)
+
+type node struct {
+	*Node
+	TTL      time.Duration
+	LastSeen time.Time
+}
+
+type record struct {
+	Name      string
+	Version   string
+	Metadata  map[string]string
+	Nodes     map[string]*node
+	Endpoints []*Endpoint
+}
+
+type memory struct {
+	opts Options
+	// records is a KV map with domain name as the key and a services map as the value
+	records  map[string]services
+	watchers map[string]*watcher
+	sync.RWMutex
+}
+
+// services is a KV map with service name as the key and a map of records as the value
+type services map[string]map[string]*record
+
+// NewRegister returns an initialized in-memory register
+func NewRegister(opts ...Option) Register {
+	r := &memory{
+		opts:     NewOptions(opts...),
+		records:  make(map[string]services),
+		watchers: make(map[string]*watcher),
+	}
+
+	go r.ttlPrune()
+
+	return r
+}
+
+func (m *memory) ttlPrune() {
+	prune := time.NewTicker(ttlPruneTime)
+	defer prune.Stop()
+
+	for {
+		select {
+		case <-prune.C:
+			m.Lock()
+			for domain, services := range m.records {
+				for service, versions := range services {
+					for version, record := range versions {
+						for id, n := range record.Nodes {
+							if n.TTL != 0 && time.Since(n.LastSeen) > n.TTL {
+								if m.opts.Logger.V(logger.DebugLevel) {
+									m.opts.Logger.Debugf(m.opts.Context, "Register TTL expired for node %s of service %s", n.Id, service)
+								}
+								delete(m.records[domain][service][version].Nodes, id)
+							}
+						}
+					}
+				}
+			}
+			m.Unlock()
+		}
+	}
+}
+
+func (m *memory) sendEvent(r *Result) {
+	m.RLock()
+	watchers := make([]*watcher, 0, len(m.watchers))
+	for _, w := range m.watchers {
+		watchers = append(watchers, w)
+	}
+	m.RUnlock()
+
+	for _, w := range watchers {
+		select {
+		case <-w.exit:
+			m.Lock()
+			delete(m.watchers, w.id)
+			m.Unlock()
+		default:
+			select {
+			case w.res <- r:
+			case <-time.After(sendEventTime):
+			}
+		}
+	}
+}
+
+func (m *memory) Connect(ctx context.Context) error {
+	return nil
+}
+
+func (m *memory) Disconnect(ctx context.Context) error {
+	return nil
+}
+
+func (m *memory) Init(opts ...Option) error {
+	for _, o := range opts {
+		o(&m.opts)
+	}
+
+	// add services
+	m.Lock()
+	defer m.Unlock()
+
+	return nil
+}
+
+func (m *memory) Options() Options {
+	return m.opts
+}
+
+func (m *memory) Register(ctx context.Context, s *Service, opts ...RegisterOption) error {
+	m.Lock()
+	defer m.Unlock()
+
+	options := NewRegisterOptions(opts...)
+
+	// get the services for this domain from the register
+	srvs, ok := m.records[options.Domain]
+	if !ok {
+		srvs = make(services)
+	}
+
+	// domain is set in metadata so it can be passed to watchers
+	if s.Metadata == nil {
+		s.Metadata = map[string]string{"domain": options.Domain}
+	} else {
+		s.Metadata["domain"] = options.Domain
+	}
+
+	// ensure the service name exists
+	r := serviceToRecord(s, options.TTL)
+	if _, ok := srvs[s.Name]; !ok {
+		srvs[s.Name] = make(map[string]*record)
+	}
+
+	if _, ok := srvs[s.Name][s.Version]; !ok {
+		srvs[s.Name][s.Version] = r
+		if m.opts.Logger.V(logger.DebugLevel) {
+			m.opts.Logger.Debugf(m.opts.Context, "Register added new service: %s, version: %s", s.Name, s.Version)
+		}
+		m.records[options.Domain] = srvs
+		go m.sendEvent(&Result{Action: "create", Service: s})
+	}
+
+	var addedNodes bool
+
+	for _, n := range s.Nodes {
+		// check if already exists
+		if _, ok := srvs[s.Name][s.Version].Nodes[n.Id]; ok {
+			continue
+		}
+
+		metadata := make(map[string]string)
+
+		// make copy of metadata
+		for k, v := range n.Metadata {
+			metadata[k] = v
+		}
+
+		// set the domain
+		metadata["domain"] = options.Domain
+
+		// add the node
+		srvs[s.Name][s.Version].Nodes[n.Id] = &node{
+			Node: &Node{
+				Id:       n.Id,
+				Address:  n.Address,
+				Metadata: metadata,
+			},
+			TTL:      options.TTL,
+			LastSeen: time.Now(),
+		}
+
+		addedNodes = true
+	}
+
+	if addedNodes {
+		if m.opts.Logger.V(logger.DebugLevel) {
+			m.opts.Logger.Debugf(m.opts.Context, "Register added new node to service: %s, version: %s", s.Name, s.Version)
+		}
+		go m.sendEvent(&Result{Action: "update", Service: s})
+	} else {
+		// refresh TTL and timestamp
+		for _, n := range s.Nodes {
+			if m.opts.Logger.V(logger.DebugLevel) {
+				m.opts.Logger.Debugf(m.opts.Context, "Updated registration for service: %s, version: %s", s.Name, s.Version)
+			}
+			srvs[s.Name][s.Version].Nodes[n.Id].TTL = options.TTL
+			srvs[s.Name][s.Version].Nodes[n.Id].LastSeen = time.Now()
+		}
+	}
+
+	m.records[options.Domain] = srvs
+	return nil
+}
+
+func (m *memory) Deregister(ctx context.Context, s *Service, opts ...DeregisterOption) error {
+	m.Lock()
+	defer m.Unlock()
+
+	options := NewDeregisterOptions(opts...)
+
+	// domain is set in metadata so it can be passed to watchers
+	if s.Metadata == nil {
+		s.Metadata = map[string]string{"domain": options.Domain}
+	} else {
+		s.Metadata["domain"] = options.Domain
+	}
+
+	// if the domain doesn't exist, there is nothing to deregister
+	services, ok := m.records[options.Domain]
+	if !ok {
+		return nil
+	}
+
+	// if no services with this name and version exist, there is nothing to deregister
+	versions, ok := services[s.Name]
+	if !ok {
+		return nil
+	}
+
+	version, ok := versions[s.Version]
+	if !ok {
+		return nil
+	}
+
+	// deregister all of the service nodes from this version
+	for _, n := range s.Nodes {
+		if _, ok := version.Nodes[n.Id]; ok {
+			if m.opts.Logger.V(logger.DebugLevel) {
+				m.opts.Logger.Debugf(m.opts.Context, "Register removed node from service: %s, version: %s", s.Name, s.Version)
+			}
+			delete(version.Nodes, n.Id)
+		}
+	}
+
+	// if the nodes not empty, we replace the version in the store and exist, the rest of the logic
+	// is cleanup
+	if len(version.Nodes) > 0 {
+		m.records[options.Domain][s.Name][s.Version] = version
+		go m.sendEvent(&Result{Action: "update", Service: s})
+		return nil
+	}
+
+	// if this version was the only version of the service, we can remove the whole service from the
+	// register and exit
+	if len(versions) == 1 {
+		delete(m.records[options.Domain], s.Name)
+		go m.sendEvent(&Result{Action: "delete", Service: s})
+
+		if m.opts.Logger.V(logger.DebugLevel) {
+			m.opts.Logger.Debugf(m.opts.Context, "Register removed service: %s", s.Name)
+		}
+		return nil
+	}
+
+	// there are other versions of the service running, so only remove this version of it
+	delete(m.records[options.Domain][s.Name], s.Version)
+	go m.sendEvent(&Result{Action: "delete", Service: s})
+	if m.opts.Logger.V(logger.DebugLevel) {
+		m.opts.Logger.Debugf(m.opts.Context, "Register removed service: %s, version: %s", s.Name, s.Version)
+	}
+
+	return nil
+}
+
+func (m *memory) LookupService(ctx context.Context, name string, opts ...LookupOption) ([]*Service, error) {
+	options := NewLookupOptions(opts...)
+
+	// if it's a wildcard domain, return from all domains
+	if options.Domain == WildcardDomain {
+		m.RLock()
+		recs := m.records
+		m.RUnlock()
+
+		var services []*Service
+
+		for domain := range recs {
+			srvs, err := m.LookupService(ctx, name, append(opts, LookupDomain(domain))...)
+			if err == ErrNotFound {
+				continue
+			} else if err != nil {
+				return nil, err
+			}
+			services = append(services, srvs...)
+		}
+
+		if len(services) == 0 {
+			return nil, ErrNotFound
+		}
+		return services, nil
+	}
+
+	m.RLock()
+	defer m.RUnlock()
+
+	// check the domain exists
+	services, ok := m.records[options.Domain]
+	if !ok {
+		return nil, ErrNotFound
+	}
+
+	// check the service exists
+	versions, ok := services[name]
+	if !ok || len(versions) == 0 {
+		return nil, ErrNotFound
+	}
+
+	// serialize the response
+	result := make([]*Service, len(versions))
+
+	var i int
+
+	for _, r := range versions {
+		result[i] = recordToService(r, options.Domain)
+		i++
+	}
+
+	return result, nil
+}
+
+func (m *memory) ListServices(ctx context.Context, opts ...ListOption) ([]*Service, error) {
+	options := NewListOptions(opts...)
+
+	// if it's a wildcard domain, list from all domains
+	if options.Domain == WildcardDomain {
+		m.RLock()
+		recs := m.records
+		m.RUnlock()
+
+		var services []*Service
+
+		for domain := range recs {
+			srvs, err := m.ListServices(ctx, append(opts, ListDomain(domain))...)
+			if err != nil {
+				return nil, err
+			}
+			services = append(services, srvs...)
+		}
+
+		return services, nil
+	}
+
+	m.RLock()
+	defer m.RUnlock()
+
+	// ensure the domain exists
+	services, ok := m.records[options.Domain]
+	if !ok {
+		return make([]*Service, 0), nil
+	}
+
+	// serialize the result, each version counts as an individual service
+	var result []*Service
+
+	for domain, service := range services {
+		for _, version := range service {
+			result = append(result, recordToService(version, domain))
+		}
+	}
+
+	return result, nil
+}
+
+func (m *memory) Watch(ctx context.Context, opts ...WatchOption) (Watcher, error) {
+	wo := NewWatchOptions(opts...)
+
+	// construct the watcher
+	w := &watcher{
+		exit: make(chan bool),
+		res:  make(chan *Result),
+		id:   uuid.New().String(),
+		wo:   wo,
+	}
+
+	m.Lock()
+	m.watchers[w.id] = w
+	m.Unlock()
+
+	return w, nil
+}
+
+func (m *memory) Name() string {
+	return m.opts.Name
+}
+
+func (m *memory) String() string {
+	return "memory"
+}
+
+type watcher struct {
+	id   string
+	wo   WatchOptions
+	res  chan *Result
+	exit chan bool
+}
+
+func (m *watcher) Next() (*Result, error) {
+	for {
+		select {
+		case r := <-m.res:
+			if r.Service == nil {
+				continue
+			}
+
+			if len(m.wo.Service) > 0 && m.wo.Service != r.Service.Name {
+				continue
+			}
+
+			// extract domain from service metadata
+			var domain string
+			if r.Service.Metadata != nil && len(r.Service.Metadata["domain"]) > 0 {
+				domain = r.Service.Metadata["domain"]
+			} else {
+				domain = DefaultDomain
+			}
+
+			// only send the event if watching the wildcard or this specific domain
+			if m.wo.Domain == WildcardDomain || m.wo.Domain == domain {
+				return r, nil
+			}
+		case <-m.exit:
+			return nil, errors.New("watcher stopped")
+		}
+	}
+}
+
+func (m *watcher) Stop() {
+	select {
+	case <-m.exit:
+		return
+	default:
+		close(m.exit)
+	}
+}
+
+func serviceToRecord(s *Service, ttl time.Duration) *record {
+	metadata := make(map[string]string, len(s.Metadata))
+	for k, v := range s.Metadata {
+		metadata[k] = v
+	}
+
+	nodes := make(map[string]*node, len(s.Nodes))
+	for _, n := range s.Nodes {
+		nodes[n.Id] = &node{
+			Node:     n,
+			TTL:      ttl,
+			LastSeen: time.Now(),
+		}
+	}
+
+	endpoints := make([]*Endpoint, len(s.Endpoints))
+	for i, e := range s.Endpoints {
+		endpoints[i] = e
+	}
+
+	return &record{
+		Name:      s.Name,
+		Version:   s.Version,
+		Metadata:  metadata,
+		Nodes:     nodes,
+		Endpoints: endpoints,
+	}
+}
+
+func recordToService(r *record, domain string) *Service {
+	metadata := make(map[string]string, len(r.Metadata))
+	for k, v := range r.Metadata {
+		metadata[k] = v
+	}
+
+	// set the domain in metadata so it can be determined when a wildcard query is performed
+	metadata["domain"] = domain
+
+	endpoints := make([]*Endpoint, len(r.Endpoints))
+	for i, e := range r.Endpoints {
+		request := new(Value)
+		if e.Request != nil {
+			*request = *e.Request
+		}
+		response := new(Value)
+		if e.Response != nil {
+			*response = *e.Response
+		}
+
+		metadata := make(map[string]string, len(e.Metadata))
+		for k, v := range e.Metadata {
+			metadata[k] = v
+		}
+
+		endpoints[i] = &Endpoint{
+			Name:     e.Name,
+			Request:  request,
+			Response: response,
+			Metadata: metadata,
+		}
+	}
+
+	nodes := make([]*Node, len(r.Nodes))
+	i := 0
+	for _, n := range r.Nodes {
+		metadata := make(map[string]string, len(n.Metadata))
+		for k, v := range n.Metadata {
+			metadata[k] = v
+		}
+
+		nodes[i] = &Node{
+			Id:       n.Id,
+			Address:  n.Address,
+			Metadata: metadata,
+		}
+		i++
+	}
+
+	return &Service{
+		Name:      r.Name,
+		Version:   r.Version,
+		Metadata:  metadata,
+		Endpoints: endpoints,
+		Nodes:     nodes,
+	}
+}

register/memory_test.go

@@ -0,0 +1,313 @@
+package register
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"testing"
+	"time"
+)
+
+var (
+	testData = map[string][]*Service{
+		"foo": {
+			{
+				Name:    "foo",
+				Version: "1.0.0",
+				Nodes: []*Node{
+					{
+						Id:      "foo-1.0.0-123",
+						Address: "localhost:9999",
+					},
+					{
+						Id:      "foo-1.0.0-321",
+						Address: "localhost:9999",
+					},
+				},
+			},
+			{
+				Name:    "foo",
+				Version: "1.0.1",
+				Nodes: []*Node{
+					{
+						Id:      "foo-1.0.1-321",
+						Address: "localhost:6666",
+					},
+				},
+			},
+			{
+				Name:    "foo",
+				Version: "1.0.3",
+				Nodes: []*Node{
+					{
+						Id:      "foo-1.0.3-345",
+						Address: "localhost:8888",
+					},
+				},
+			},
+		},
+		"bar": {
+			{
+				Name:    "bar",
+				Version: "default",
+				Nodes: []*Node{
+					{
+						Id:      "bar-1.0.0-123",
+						Address: "localhost:9999",
+					},
+					{
+						Id:      "bar-1.0.0-321",
+						Address: "localhost:9999",
+					},
+				},
+			},
+			{
+				Name:    "bar",
+				Version: "latest",
+				Nodes: []*Node{
+					{
+						Id:      "bar-1.0.1-321",
+						Address: "localhost:6666",
+					},
+				},
+			},
+		},
+	}
+)
+
+func TestMemoryRegistry(t *testing.T) {
+	ctx := context.TODO()
+	m := NewRegister()
+
+	fn := func(k string, v []*Service) {
+		services, err := m.LookupService(ctx, k)
+		if err != nil {
+			t.Errorf("Unexpected error getting service %s: %v", k, err)
+		}
+
+		if len(services) != len(v) {
+			t.Errorf("Expected %d services for %s, got %d", len(v), k, len(services))
+		}
+
+		for _, service := range v {
+			var seen bool
+			for _, s := range services {
+				if s.Version == service.Version {
+					seen = true
+					break
+				}
+			}
+			if !seen {
+				t.Errorf("expected to find version %s", service.Version)
+			}
+		}
+	}
+
+	// register data
+	for _, v := range testData {
+		serviceCount := 0
+		for _, service := range v {
+			if err := m.Register(ctx, service); err != nil {
+				t.Errorf("Unexpected register error: %v", err)
+			}
+			serviceCount++
+			// after the service has been registered we should be able to query it
+			services, err := m.LookupService(ctx, service.Name)
+			if err != nil {
+				t.Errorf("Unexpected error getting service %s: %v", service.Name, err)
+			}
+			if len(services) != serviceCount {
+				t.Errorf("Expected %d services for %s, got %d", serviceCount, service.Name, len(services))
+			}
+		}
+	}
+
+	// using test data
+	for k, v := range testData {
+		fn(k, v)
+	}
+
+	services, err := m.ListServices(ctx)
+	if err != nil {
+		t.Errorf("Unexpected error when listing services: %v", err)
+	}
+
+	totalServiceCount := 0
+	for _, testSvc := range testData {
+		for range testSvc {
+			totalServiceCount++
+		}
+	}
+
+	if len(services) != totalServiceCount {
+		t.Errorf("Expected total service count: %d, got: %d", totalServiceCount, len(services))
+	}
+
+	// deregister
+	for _, v := range testData {
+		for _, service := range v {
+			if err := m.Deregister(ctx, service); err != nil {
+				t.Errorf("Unexpected deregister error: %v", err)
+			}
+		}
+	}
+
+	// after all the service nodes have been deregistered we should not get any results
+	for _, v := range testData {
+		for _, service := range v {
+			services, err := m.LookupService(ctx, service.Name)
+			if err != ErrNotFound {
+				t.Errorf("Expected error: %v, got: %v", ErrNotFound, err)
+			}
+			if len(services) != 0 {
+				t.Errorf("Expected %d services for %s, got %d", 0, service.Name, len(services))
+			}
+		}
+	}
+}
+
+func TestMemoryRegistryTTL(t *testing.T) {
+	m := NewRegister()
+	ctx := context.TODO()
+
+	for _, v := range testData {
+		for _, service := range v {
+			if err := m.Register(ctx, service, RegisterTTL(time.Millisecond)); err != nil {
+				t.Fatal(err)
+			}
+		}
+	}
+
+	time.Sleep(ttlPruneTime * 2)
+
+	for name := range testData {
+		svcs, err := m.LookupService(ctx, name)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		for _, svc := range svcs {
+			if len(svc.Nodes) > 0 {
+				t.Fatalf("Service %q still has nodes registered", name)
+			}
+		}
+	}
+}
+
+func TestMemoryRegistryTTLConcurrent(t *testing.T) {
+	concurrency := 1000
+	waitTime := ttlPruneTime * 2
+	m := NewRegister()
+	ctx := context.TODO()
+	for _, v := range testData {
+		for _, service := range v {
+			if err := m.Register(ctx, service, RegisterTTL(waitTime/2)); err != nil {
+				t.Fatal(err)
+			}
+		}
+	}
+
+	if len(os.Getenv("IN_TRAVIS_CI")) == 0 {
+		t.Logf("test will wait %v, then check TTL timeouts", waitTime)
+	}
+
+	errChan := make(chan error, concurrency)
+	syncChan := make(chan struct{})
+
+	for i := 0; i < concurrency; i++ {
+		go func() {
+			<-syncChan
+			for name := range testData {
+				svcs, err := m.LookupService(ctx, name)
+				if err != nil {
+					errChan <- err
+					return
+				}
+
+				for _, svc := range svcs {
+					if len(svc.Nodes) > 0 {
+						errChan <- fmt.Errorf("Service %q still has nodes registered", name)
+						return
+					}
+				}
+			}
+
+			errChan <- nil
+		}()
+	}
+
+	time.Sleep(waitTime)
+	close(syncChan)
+
+	for i := 0; i < concurrency; i++ {
+		if err := <-errChan; err != nil {
+			t.Fatal(err)
+		}
+	}
+}
+
+func TestMemoryWildcard(t *testing.T) {
+	m := NewRegister()
+	ctx := context.TODO()
+
+	testSrv := &Service{Name: "foo", Version: "1.0.0"}
+
+	if err := m.Register(ctx, testSrv, RegisterDomain("one")); err != nil {
+		t.Fatalf("Register err: %v", err)
+	}
+	if err := m.Register(ctx, testSrv, RegisterDomain("two")); err != nil {
+		t.Fatalf("Register err: %v", err)
+	}
+
+	if recs, err := m.ListServices(ctx, ListDomain("one")); err != nil {
+		t.Errorf("List err: %v", err)
+	} else if len(recs) != 1 {
+		t.Errorf("Expected 1 record, got %v", len(recs))
+	}
+
+	if recs, err := m.ListServices(ctx, ListDomain("*")); err != nil {
+		t.Errorf("List err: %v", err)
+	} else if len(recs) != 2 {
+		t.Errorf("Expected 2 records, got %v", len(recs))
+	}
+
+	if recs, err := m.LookupService(ctx, testSrv.Name, LookupDomain("one")); err != nil {
+		t.Errorf("Lookup err: %v", err)
+	} else if len(recs) != 1 {
+		t.Errorf("Expected 1 record, got %v", len(recs))
+	}
+
+	if recs, err := m.LookupService(ctx, testSrv.Name, LookupDomain("*")); err != nil {
+		t.Errorf("Lookup err: %v", err)
+	} else if len(recs) != 2 {
+		t.Errorf("Expected 2 records, got %v", len(recs))
+	}
+}
+
+func TestWatcher(t *testing.T) {
+	w := &watcher{
+		id:   "test",
+		res:  make(chan *Result),
+		exit: make(chan bool),
+		wo: WatchOptions{
+			Domain: WildcardDomain,
+		},
+	}
+
+	go func() {
+		w.res <- &Result{
+			Service: &Service{Name: "foo"},
+		}
+	}()
+
+	_, err := w.Next()
+	if err != nil {
+		t.Fatal("unexpected err", err)
+	}
+
+	w.Stop()
+
+	if _, err := w.Next(); err == nil {
+		t.Fatal("expected error on Next()")
+	}
+}

register/noop.go

@@ -1,85 +0,0 @@
-package register
-
-import (
-	"context"
-)
-
-type noopRegister struct {
-	opts Options
-}
-
-func (n *noopRegister) Name() string {
-	return n.opts.Name
-}
-
-// Init initialize register
-func (n *noopRegister) Init(opts ...Option) error {
-	for _, o := range opts {
-		o(&n.opts)
-	}
-	return nil
-}
-
-// Options returns options struct
-func (n *noopRegister) Options() Options {
-	return n.opts
-}
-
-// Connect opens connection to register
-func (n *noopRegister) Connect(ctx context.Context) error {
-	return nil
-}
-
-// Disconnect close connection to register
-func (n *noopRegister) Disconnect(ctx context.Context) error {
-	return nil
-}
-
-// Register registers service
-func (n *noopRegister) Register(ctx context.Context, svc *Service, opts ...RegisterOption) error {
-	return nil
-}
-
-// Deregister deregisters service
-func (n *noopRegister) Deregister(ctx context.Context, svc *Service, opts ...DeregisterOption) error {
-	return nil
-}
-
-// LookupService returns servive info
-func (n *noopRegister) LookupService(ctx context.Context, name string, opts ...LookupOption) ([]*Service, error) {
-	return []*Service{}, nil
-}
-
-// ListServices listing services
-func (n *noopRegister) ListServices(ctx context.Context, opts ...ListOption) ([]*Service, error) {
-	return []*Service{}, nil
-}
-
-// Watch is used to watch for service changes
-func (n *noopRegister) Watch(ctx context.Context, opts ...WatchOption) (Watcher, error) {
-	return &noopWatcher{done: make(chan struct{}), opts: NewWatchOptions(opts...)}, nil
-}
-
-// String returns register string representation
-func (n *noopRegister) String() string {
-	return "noop"
-}
-
-type noopWatcher struct {
-	opts WatchOptions
-	done chan struct{}
-}
-
-func (n *noopWatcher) Next() (*Result, error) {
-	<-n.done
-	return nil, ErrWatcherStopped
-}
-
-func (n *noopWatcher) Stop() {
-	close(n.done)
-}
-
-// NewRegister returns a new noop register
-func NewRegister(opts ...Option) Register {
-	return &noopRegister{opts: NewOptions(opts...)}
-}

register/options.go

@@ -246,13 +246,13 @@ func DeregisterDomain(d string) DeregisterOption {
 	}
 }
 
-func GetContext(ctx context.Context) LookupOption {
+func LookupContext(ctx context.Context) LookupOption {
 	return func(o *LookupOptions) {
 		o.Context = ctx
 	}
 }
 
-func GetDomain(d string) LookupOption {
+func LookupDomain(d string) LookupOption {
 	return func(o *LookupOptions) {
 		o.Domain = d
 	}

server/noop.go

@@ -46,7 +46,17 @@ type noopServer struct {
 
 // NewServer returns new noop server
 func NewServer(opts ...Option) Server {
-	return &noopServer{opts: NewOptions(opts...)}
+	n := &noopServer{opts: NewOptions(opts...)}
+	if n.handlers == nil {
+		n.handlers = make(map[string]Handler)
+	}
+	if n.subscribers == nil {
+		n.subscribers = make(map[*subscriber][]broker.Subscriber)
+	}
+	if n.exit == nil {
+		n.exit = make(chan chan error)
+	}
+	return n
 }
 
 func (n *noopServer) newCodec(contentType string) (codec.Codec, error) {

store/memory.go

@@ -0,0 +1,199 @@
+package store
+
+import (
+	"context"
+	"path/filepath"
+	"sort"
+	"strings"
+	"time"
+
+	"github.com/patrickmn/go-cache"
+)
+
+// NewStore returns a memory store
+func NewStore(opts ...Option) Store {
+	return &memoryStore{
+		opts:  NewOptions(opts...),
+		store: cache.New(cache.NoExpiration, 5*time.Minute),
+	}
+}
+
+func (m *memoryStore) Connect(ctx context.Context) error {
+	return nil
+}
+
+func (m *memoryStore) Disconnect(ctx context.Context) error {
+	m.store.Flush()
+	return nil
+}
+
+type memoryStore struct {
+	opts  Options
+	store *cache.Cache
+}
+
+func (m *memoryStore) key(prefix, key string) string {
+	return filepath.Join(prefix, key)
+}
+
+func (m *memoryStore) prefix(database, table string) string {
+	if len(database) == 0 {
+		database = m.opts.Database
+	}
+	if len(table) == 0 {
+		table = m.opts.Table
+	}
+	return filepath.Join(database, table)
+}
+
+func (m *memoryStore) exists(prefix, key string) error {
+	key = m.key(prefix, key)
+
+	_, found := m.store.Get(key)
+	if !found {
+		return ErrNotFound
+	}
+
+	return nil
+}
+
+func (m *memoryStore) get(prefix, key string, val interface{}) error {
+	key = m.key(prefix, key)
+
+	r, found := m.store.Get(key)
+	if !found {
+		return ErrNotFound
+	}
+
+	buf, ok := r.([]byte)
+	if !ok {
+		return ErrNotFound
+	}
+
+	if err := m.opts.Codec.Unmarshal(buf, val); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (m *memoryStore) delete(prefix, key string) {
+	key = m.key(prefix, key)
+	m.store.Delete(key)
+}
+
+func (m *memoryStore) list(prefix string, limit, offset uint) []string {
+	allItems := m.store.Items()
+	allKeys := make([]string, len(allItems))
+	i := 0
+
+	for k := range allItems {
+		if !strings.HasPrefix(k, prefix+"/") {
+			continue
+		}
+		allKeys[i] = strings.TrimPrefix(k, prefix+"/")
+		i++
+	}
+
+	if limit != 0 || offset != 0 {
+		sort.Slice(allKeys, func(i, j int) bool { return allKeys[i] < allKeys[j] })
+		sort.Slice(allKeys, func(i, j int) bool { return allKeys[i] < allKeys[j] })
+		end := len(allKeys)
+		if limit > 0 {
+			calcLimit := int(offset + limit)
+			if calcLimit < end {
+				end = calcLimit
+			}
+		}
+
+		if int(offset) >= end {
+			return nil
+		}
+		return allKeys[offset:end]
+	}
+
+	return allKeys
+}
+
+func (m *memoryStore) Init(opts ...Option) error {
+	for _, o := range opts {
+		o(&m.opts)
+	}
+	return nil
+}
+
+func (m *memoryStore) String() string {
+	return "memory"
+}
+
+func (m *memoryStore) Name() string {
+	return m.opts.Name
+}
+
+func (m *memoryStore) Exists(ctx context.Context, key string, opts ...ExistsOption) error {
+	prefix := m.prefix(m.opts.Database, m.opts.Table)
+	return m.exists(prefix, key)
+}
+
+func (m *memoryStore) Read(ctx context.Context, key string, val interface{}, opts ...ReadOption) error {
+	readOpts := NewReadOptions(opts...)
+	prefix := m.prefix(readOpts.Database, readOpts.Table)
+	return m.get(prefix, key, val)
+}
+
+func (m *memoryStore) Write(ctx context.Context, key string, val interface{}, opts ...WriteOption) error {
+	writeOpts := NewWriteOptions(opts...)
+
+	prefix := m.prefix(writeOpts.Database, writeOpts.Table)
+
+	key = m.key(prefix, key)
+
+	buf, err := m.opts.Codec.Marshal(val)
+	if err != nil {
+		return err
+	}
+
+	m.store.Set(key, buf, writeOpts.TTL)
+	return nil
+}
+
+func (m *memoryStore) Delete(ctx context.Context, key string, opts ...DeleteOption) error {
+	deleteOptions := NewDeleteOptions(opts...)
+
+	prefix := m.prefix(deleteOptions.Database, deleteOptions.Table)
+	m.delete(prefix, key)
+	return nil
+}
+
+func (m *memoryStore) Options() Options {
+	return m.opts
+}
+
+func (m *memoryStore) List(ctx context.Context, opts ...ListOption) ([]string, error) {
+	listOptions := NewListOptions(opts...)
+
+	prefix := m.prefix(listOptions.Database, listOptions.Table)
+	keys := m.list(prefix, listOptions.Limit, listOptions.Offset)
+
+	if len(listOptions.Prefix) > 0 {
+		var prefixKeys []string
+		for _, k := range keys {
+			if strings.HasPrefix(k, listOptions.Prefix) {
+				prefixKeys = append(prefixKeys, k)
+			}
+		}
+		keys = prefixKeys
+	}
+
+	if len(listOptions.Suffix) > 0 {
+		var suffixKeys []string
+		for _, k := range keys {
+			if strings.HasSuffix(k, listOptions.Suffix) {
+				suffixKeys = append(suffixKeys, k)
+			}
+		}
+		keys = suffixKeys
+	}
+
+	return keys, nil
+}

store/memory_test.go

@@ -0,0 +1,67 @@
+package store_test
+
+import (
+	"context"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/unistack-org/micro/v3/store"
+)
+
+func TestMemoryReInit(t *testing.T) {
+	s := store.NewStore(store.Table("aaa"))
+	s.Init(store.Table(""))
+	if len(s.Options().Table) > 0 {
+		t.Error("Init didn't reinitialise the store")
+	}
+}
+
+func TestMemoryBasic(t *testing.T) {
+	s := store.NewStore()
+	s.Init()
+	basictest(s, t)
+}
+
+func TestMemoryPrefix(t *testing.T) {
+	s := store.NewStore()
+	s.Init(store.Table("some-prefix"))
+	basictest(s, t)
+}
+
+func TestMemoryNamespace(t *testing.T) {
+	s := store.NewStore()
+	s.Init(store.Database("some-namespace"))
+	basictest(s, t)
+}
+
+func TestMemoryNamespacePrefix(t *testing.T) {
+	s := store.NewStore()
+	s.Init(store.Table("some-prefix"), store.Database("some-namespace"))
+	basictest(s, t)
+}
+
+func basictest(s store.Store, t *testing.T) {
+	ctx := context.Background()
+	if len(os.Getenv("IN_TRAVIS_CI")) == 0 {
+		t.Logf("Testing store %s, with options %#+v\n", s.String(), s.Options())
+	}
+	// Read and Write an expiring Record
+	if err := s.Write(ctx, "Hello", "World", store.WriteTTL(time.Millisecond*100)); err != nil {
+		t.Error(err)
+	}
+	var val []byte
+	if err := s.Read(ctx, "Hello", &val); err != nil {
+		t.Error(err)
+	} else {
+		if string(val) != "World" {
+			t.Errorf("Expected %s, got %s", "World", val)
+		}
+	}
+	time.Sleep(time.Millisecond * 200)
+	if err := s.Read(ctx, "Hello", &val); err != store.ErrNotFound {
+		t.Errorf("Expected %# v, got %# v", store.ErrNotFound, err)
+	}
+
+	s.Disconnect(ctx) // reset the store
+}

store/noop.go

@@ -1,69 +0,0 @@
-package store
-
-import "context"
-
-type noopStore struct {
-	opts Options
-}
-
-func NewStore(opts ...Option) Store {
-	return &noopStore{opts: NewOptions(opts...)}
-}
-
-// Init initialize store
-func (n *noopStore) Init(opts ...Option) error {
-	for _, o := range opts {
-		o(&n.opts)
-	}
-	return nil
-}
-
-// Options returns Options struct
-func (n *noopStore) Options() Options {
-	return n.opts
-}
-
-// Name
-func (n *noopStore) Name() string {
-	return n.opts.Name
-}
-
-// String returns string representation
-func (n *noopStore) String() string {
-	return "noop"
-}
-
-// Read reads store value by key
-func (n *noopStore) Exists(ctx context.Context, key string, opts ...ExistsOption) error {
-	return ErrNotFound
-}
-
-// Read reads store value by key
-func (n *noopStore) Read(ctx context.Context, key string, val interface{}, opts ...ReadOption) error {
-	return ErrNotFound
-}
-
-// Write writes store record
-func (n *noopStore) Write(ctx context.Context, key string, val interface{}, opts ...WriteOption) error {
-	return nil
-}
-
-// Delete removes store value by key
-func (n *noopStore) Delete(ctx context.Context, key string, opts ...DeleteOption) error {
-	return nil
-}
-
-// List lists store
-func (n *noopStore) List(ctx context.Context, opts ...ListOption) ([]string, error) {
-	return []string{}, nil
-}
-
-// Connect connects to store
-func (n *noopStore) Connect(ctx context.Context) error {
-	return nil
-}
-
-// Disconnect disconnects from store
-func (n *noopStore) Disconnect(ctx context.Context) error {
-	return nil
-}

sync/memory.go

@@ -0,0 +1,199 @@
+package sync
+
+import (
+	gosync "sync"
+	"time"
+)
+
+type memorySync struct {
+	options Options
+
+	mtx   gosync.RWMutex
+	locks map[string]*memoryLock
+}
+
+type memoryLock struct {
+	id      string
+	time    time.Time
+	ttl     time.Duration
+	release chan bool
+}
+
+type memoryLeader struct {
+	opts   LeaderOptions
+	id     string
+	resign func(id string) error
+	status chan bool
+}
+
+func (m *memoryLeader) Resign() error {
+	return m.resign(m.id)
+}
+
+func (m *memoryLeader) Status() chan bool {
+	return m.status
+}
+
+func (m *memorySync) Leader(id string, opts ...LeaderOption) (Leader, error) {
+	var once gosync.Once
+	var options LeaderOptions
+	for _, o := range opts {
+		o(&options)
+	}
+
+	// acquire a lock for the id
+	if err := m.Lock(id); err != nil {
+		return nil, err
+	}
+
+	// return the leader
+	return &memoryLeader{
+		opts: options,
+		id:   id,
+		resign: func(id string) error {
+			once.Do(func() {
+				m.Unlock(id)
+			})
+			return nil
+		},
+		// TODO: signal when Unlock is called
+		status: make(chan bool, 1),
+	}, nil
+}
+
+func (m *memorySync) Init(opts ...Option) error {
+	for _, o := range opts {
+		o(&m.options)
+	}
+	return nil
+}
+
+func (m *memorySync) Options() Options {
+	return m.options
+}
+
+func (m *memorySync) Lock(id string, opts ...LockOption) error {
+	// lock our access
+	m.mtx.Lock()
+
+	var options LockOptions
+	for _, o := range opts {
+		o(&options)
+	}
+
+	lk, ok := m.locks[id]
+	if !ok {
+		m.locks[id] = &memoryLock{
+			id:      id,
+			time:    time.Now(),
+			ttl:     options.TTL,
+			release: make(chan bool),
+		}
+		// unlock
+		m.mtx.Unlock()
+		return nil
+	}
+
+	m.mtx.Unlock()
+
+	// set wait time
+	var wait <-chan time.Time
+	var ttl <-chan time.Time
+
+	// decide if we should wait
+	if options.Wait > time.Duration(0) {
+		wait = time.After(options.Wait)
+	}
+
+	// check the ttl of the lock
+	if lk.ttl > time.Duration(0) {
+		// time lived for the lock
+		live := time.Since(lk.time)
+
+		// set a timer for the leftover ttl
+		if live > lk.ttl {
+			// release the lock if it expired
+			_ = m.Unlock(id)
+		} else {
+			ttl = time.After(live)
+		}
+	}
+
+lockLoop:
+	for {
+		// wait for the lock to be released
+		select {
+		case <-lk.release:
+			m.mtx.Lock()
+
+			// someone locked before us
+			lk, ok = m.locks[id]
+			if ok {
+				m.mtx.Unlock()
+				continue
+			}
+
+			// got chance to lock
+			m.locks[id] = &memoryLock{
+				id:      id,
+				time:    time.Now(),
+				ttl:     options.TTL,
+				release: make(chan bool),
+			}
+
+			m.mtx.Unlock()
+
+			break lockLoop
+		case <-ttl:
+			// ttl exceeded
+			_ = m.Unlock(id)
+			// TODO: check the ttl again above
+			ttl = nil
+			// try acquire
+			continue
+		case <-wait:
+			return ErrLockTimeout
+		}
+	}
+
+	return nil
+}
+
+func (m *memorySync) Unlock(id string) error {
+	m.mtx.Lock()
+	defer m.mtx.Unlock()
+
+	lk, ok := m.locks[id]
+	// no lock exists
+	if !ok {
+		return nil
+	}
+
+	// delete the lock
+	delete(m.locks, id)
+
+	select {
+	case <-lk.release:
+		return nil
+	default:
+		close(lk.release)
+	}
+
+	return nil
+}
+
+func (m *memorySync) String() string {
+	return "memory"
+}
+
+func NewSync(opts ...Option) Sync {
+	var options Options
+	for _, o := range opts {
+		o(&options)
+	}
+
+	return &memorySync{
+		options: options,
+		locks:   make(map[string]*memoryLock),
+	}
+}

tracer/context.go

@@ -14,11 +14,12 @@ const (
 
 // FromContext returns a span from context
 func FromContext(ctx context.Context) (traceID string, parentSpanID string, isFound bool) {
-	if ctx == nil {
+	md, ok := metadata.FromIncomingContext(ctx)
+	if !ok {
 		return "", "", false
 	}
-	traceID, traceOk := metadata.Get(ctx, traceIDKey)
-	microID, microOk := metadata.Get(ctx, "Micro-Id")
+	traceID, traceOk := md.Get(traceIDKey)
+	microID, microOk := md.Get("Micro-Id")
 	if !traceOk && !microOk {
 		isFound = false
 		return
@@ -26,17 +27,17 @@ func FromContext(ctx context.Context) (traceID string, parentSpanID string, isFo
 	if !traceOk {
 		traceID = microID
 	}
-	parentSpanID, ok := metadata.Get(ctx, spanIDKey)
+	parentSpanID, ok = md.Get(spanIDKey)
 	return traceID, parentSpanID, ok
 }
 
 // NewContext saves the trace and span ids in the context
 func NewContext(ctx context.Context, traceID, parentSpanID string) context.Context {
-	if ctx == nil {
-		ctx = context.Background()
+	md, ok := metadata.FromContext(ctx)
+	if !ok {
+		md = metadata.New(2)
 	}
-	md := metadata.New(2)
 	md.Set(traceIDKey, traceID)
 	md.Set(spanIDKey, parentSpanID)
-	return metadata.MergeContext(ctx, md, true)
+	return metadata.NewContext(ctx, md)
 }

tracer/memory.go

@@ -0,0 +1,98 @@
+package tracer
+
+import (
+	"context"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/unistack-org/micro/v3/util/ring"
+)
+
+type tracer struct {
+	opts Options
+	// ring buffer of traces
+	buffer *ring.Buffer
+}
+
+func (t *tracer) Read(opts ...ReadOption) ([]*Span, error) {
+	var options ReadOptions
+	for _, o := range opts {
+		o(&options)
+	}
+
+	sp := t.buffer.Get(t.buffer.Size())
+
+	spans := make([]*Span, 0, len(sp))
+
+	for _, span := range sp {
+		val := span.Value.(*Span)
+		// skip if trace id is specified and doesn't match
+		if len(options.Trace) > 0 && val.Trace != options.Trace {
+			continue
+		}
+		spans = append(spans, val)
+	}
+
+	return spans, nil
+}
+
+func (t *tracer) Start(ctx context.Context, name string) (context.Context, *Span) {
+	span := &Span{
+		Name:     name,
+		Trace:    uuid.New().String(),
+		Id:       uuid.New().String(),
+		Started:  time.Now(),
+		Metadata: make(map[string]string),
+	}
+
+	// return span if no context
+	if ctx == nil {
+		return NewContext(context.Background(), span.Trace, span.Id), span
+	}
+	traceID, parentSpanID, ok := FromContext(ctx)
+	// If the trace can not be found in the header,
+	// that means this is where the trace is created.
+	if !ok {
+		return NewContext(ctx, span.Trace, span.Id), span
+	}
+
+	// set trace id
+	span.Trace = traceID
+	// set parent
+	span.Parent = parentSpanID
+
+	// return the span
+	return NewContext(ctx, span.Trace, span.Id), span
+}
+
+func (t *tracer) Finish(s *Span) error {
+	// set finished time
+	s.Duration = time.Since(s.Started)
+	// save the span
+	t.buffer.Put(s)
+
+	return nil
+}
+
+func (t *tracer) Init(opts ...Option) error {
+	for _, o := range opts {
+		o(&t.opts)
+	}
+	return nil
+}
+
+func (t *tracer) Lookup(ctx context.Context) (*Span, error) {
+	return nil, nil
+}
+
+func (t *tracer) Name() string {
+	return t.opts.Name
+}
+
+func NewTracer(opts ...Option) Tracer {
+	return &tracer{
+		opts: NewOptions(opts...),
+		// the last 256 requests
+		buffer: ring.New(256),
+	}
+}

tracer/noop.go

@@ -1,44 +0,0 @@
-package tracer
-
-import "context"
-
-type noopTracer struct {
-	opts Options
-}
-
-func (n *noopTracer) Name() string {
-	return n.opts.Name
-}
-
-// Init initilize tracer
-func (n *noopTracer) Init(opts ...Option) error {
-	for _, o := range opts {
-		o(&n.opts)
-	}
-	return nil
-}
-
-// Start starts new span
-func (n *noopTracer) Start(ctx context.Context, name string) (context.Context, *Span) {
-	return nil, nil
-}
-
-// Lookup get span from context
-func (n *noopTracer) Lookup(ctx context.Context) (*Span, error) {
-	return nil, nil
-}
-
-// Finish finishes span
-func (n *noopTracer) Finish(*Span) error {
-	return nil
-}
-
-// Read reads span
-func (n *noopTracer) Read(...ReadOption) ([]*Span, error) {
-	return nil, nil
-}
-
-// NewTracer returns new noop tracer
-func NewTracer(opts ...Option) Tracer {
-	return &noopTracer{opts: NewOptions(opts...)}
-}

util/ctx/ctx.go

@@ -10,10 +10,9 @@ import (
 
 func FromRequest(r *http.Request) context.Context {
 	ctx := r.Context()
-	md, ok := metadata.FromContext(ctx)
+	md, ok := metadata.FromIncomingContext(ctx)
 	if !ok {
-		// create needed map with specific len
-		md = make(metadata.Metadata, len(r.Header)+2)
+		md = metadata.New(len(r.Header) + 2)
 	}
 	for key, val := range r.Header {
 		md.Set(key, strings.Join(val, ","))
@@ -22,5 +21,5 @@ func FromRequest(r *http.Request) context.Context {
 	md["Host"] = r.Host
 	// pass http method
 	md["Method"] = r.Method
-	return metadata.NewContext(ctx, md)
+	return metadata.NewIncomingContext(ctx, md)
 }

util/ctx/ctx_test.go

@@ -28,7 +28,7 @@ func TestRequestToContext(t *testing.T) {
 
 	for _, d := range testData {
 		ctx := FromRequest(d.request)
-		md, ok := metadata.FromContext(ctx)
+		md, ok := metadata.FromIncomingContext(ctx)
 		if !ok {
 			t.Fatalf("Expected metadata for request %+v", d.request)
 		}

util/reflect/reflect.go

@@ -2,24 +2,53 @@ package reflect
 
 import (
 	"errors"
+	"fmt"
+	"net/url"
 	"reflect"
+	"regexp"
+	"strconv"
+	"strings"
+	"unicode"
 )
 
 var (
+	bracketSplitter  = regexp.MustCompile(`\[|\]`)
 	ErrInvalidStruct = errors.New("invalid struct specified")
+	ErrInvalidParam  = errors.New("invalid url query param provided")
 )
 
+func fieldName(name string) string {
+	newstr := make([]rune, 0)
+	upper := false
+	for idx, chr := range name {
+		if idx == 0 {
+			upper = true
+		} else if chr == '_' {
+			upper = true
+			continue
+		}
+		if upper {
+			newstr = append(newstr, unicode.ToUpper(chr))
+		} else {
+			newstr = append(newstr, chr)
+		}
+		upper = false
+	}
+
+	return string(newstr)
+}
+
 func IsEmpty(v reflect.Value) bool {
-	switch v.Kind() {
+	switch getKind(v) {
 	case reflect.Array, reflect.Map, reflect.Slice, reflect.String:
 		return v.Len() == 0
 	case reflect.Bool:
 		return !v.Bool()
-	case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64:
+	case reflect.Int:
 		return v.Int() == 0
-	case reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64, reflect.Uintptr:
+	case reflect.Uint:
 		return v.Uint() == 0
-	case reflect.Float32, reflect.Float64:
+	case reflect.Float32:
 		return v.Float() == 0
 	case reflect.Interface, reflect.Ptr:
 		if v.IsNil() {
@@ -119,3 +148,450 @@ func CopyFrom(a, b interface{}) {
 		}
 	}
 }
+
+func URLMap(query string) (map[string]interface{}, error) {
+	var (
+		mp interface{} = make(map[string]interface{})
+	)
+
+	params := strings.Split(query, "&")
+
+	for _, part := range params {
+		tm, err := queryToMap(part)
+		if err != nil {
+			return nil, err
+		}
+		mp = merge(mp, tm)
+	}
+
+	return mp.(map[string]interface{}), nil
+}
+
+func FlattenMap(a map[string]interface{}) map[string]interface{} {
+	// preprocess map
+	nb := make(map[string]interface{}, len(a))
+	for k, v := range a {
+		ps := strings.Split(k, ".")
+		if len(ps) == 1 {
+			nb[k] = v
+			continue
+		}
+		em := make(map[string]interface{})
+		em[ps[len(ps)-1]] = v
+		for i := len(ps) - 2; i > 0; i-- {
+			nm := make(map[string]interface{})
+			nm[ps[i]] = em
+			em = nm
+		}
+		if vm, ok := nb[ps[0]]; ok {
+			// nested map
+			nm := vm.(map[string]interface{})
+			for vk, vv := range em {
+				nm[vk] = vv
+			}
+			nb[ps[0]] = nm
+		} else {
+			nb[ps[0]] = em
+		}
+	}
+	return nb
+}
+
+func MergeMap(a interface{}, b map[string]interface{}) error {
+	var err error
+
+	ta := reflect.TypeOf(a)
+	if ta.Kind() == reflect.Ptr {
+		ta = ta.Elem()
+	}
+	va := reflect.ValueOf(a)
+	if va.Kind() == reflect.Ptr {
+		va = va.Elem()
+	}
+
+	for mk, mv := range b {
+		vmv := reflect.ValueOf(mv)
+		name := fieldName(mk)
+		fva := va.FieldByName(name)
+		fta, found := ta.FieldByName(name)
+		if !found || !fva.IsValid() || !fva.CanSet() || fta.PkgPath != "" {
+			continue
+		}
+		// fast path via direct assign
+		if vmv.Type().AssignableTo(fta.Type) && !IsEmpty(vmv) {
+			fva.Set(vmv)
+			continue
+		}
+		switch getKind(fva) {
+		case reflect.Bool:
+			err = mergeBool(fva, vmv)
+		case reflect.String:
+			err = mergeString(fva, vmv)
+		case reflect.Int:
+			err = mergeInt(fva, vmv)
+		case reflect.Uint:
+			err = mergeUint(fva, vmv)
+		case reflect.Float32:
+			err = mergeFloat(fva, vmv)
+		case reflect.Array:
+			//fmt.Printf("Array %#+v %#+v\n", fva, vmv)
+		case reflect.Slice:
+			err = mergeSlice(fva, vmv)
+		case reflect.Ptr:
+			if fva.IsNil() {
+				fva.Set(reflect.New(fva.Type().Elem()))
+				if fva.Elem().Type().Kind() == reflect.Struct {
+					for i := 0; i < fva.Elem().NumField(); i++ {
+						field := fva.Elem().Field(i)
+						if field.Type().Kind() == reflect.Ptr && field.IsNil() && fva.Elem().Type().Field(i).Anonymous == true {
+							field.Set(reflect.New(field.Type().Elem()))
+						}
+					}
+				}
+			}
+			if nmp, ok := vmv.Interface().(map[string]interface{}); ok {
+				err = MergeMap(fva.Interface(), nmp)
+			} else {
+				err = fmt.Errorf("cant fill")
+			}
+		case reflect.Struct:
+			if nmp, ok := vmv.Interface().(map[string]interface{}); ok {
+				err = MergeMap(fva.Interface(), nmp)
+			} else {
+				err = fmt.Errorf("cant fill")
+			}
+		case reflect.Map:
+			//fmt.Printf("Map %#+v %#+v\n", fva, vmv)
+		}
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func mergeSlice(va, vb reflect.Value) error {
+	switch getKind(vb) {
+	/*
+		case reflect.Int:
+				if vb.Int() == 1 {
+					va.SetBool(true)
+				}
+			case reflect.Uint:
+				if vb.Uint() == 1 {
+					va.SetBool(true)
+				}
+			case reflect.Float64:
+				if vb.Float() == 1 {
+					va.SetBool(true)
+				}
+	*/
+	case reflect.String:
+		var err error
+		fn := func(c rune) bool { return c == ',' || c == ';' || c == ' ' }
+		slice := strings.FieldsFunc(vb.String(), fn)
+		va.Set(reflect.MakeSlice(va.Type(), len(slice), len(slice)))
+		for idx, sl := range slice {
+			vl := reflect.ValueOf(sl)
+			switch va.Type().Elem().Kind() {
+			case reflect.Bool:
+				err = mergeBool(va.Index(idx), vl)
+			case reflect.String:
+				err = mergeString(va.Index(idx), vl)
+			case reflect.Ptr:
+				if va.Index(idx).IsNil() {
+					va.Index(idx).Set(reflect.New(va.Index(idx).Type().Elem()))
+				}
+				switch va.Type().Elem().String() {
+				case "*wrapperspb.BoolValue":
+					if eva := reflect.Indirect(va.Index(idx)).FieldByName("Value"); eva.IsValid() {
+						err = mergeBool(eva, vl)
+					}
+				case "*wrapperspb.BytesValue":
+					if eva := va.Index(idx).FieldByName("Value"); eva.IsValid() {
+						err = mergeUint(eva, vl)
+					}
+				case "*wrapperspb.DoubleValue", "*wrapperspb.FloatValue":
+					if eva := reflect.Indirect(va.Index(idx)).FieldByName("Value"); eva.IsValid() {
+						err = mergeFloat(eva, vl)
+					}
+				case "*wrapperspb.Int32Value", "*wrapperspb.Int64Value":
+					if eva := reflect.Indirect(va.Index(idx)).FieldByName("Value"); eva.IsValid() {
+						err = mergeInt(eva, vl)
+					}
+				case "*wrapperspb.StringValue":
+					if eva := reflect.Indirect(va.Index(idx)).FieldByName("Value"); eva.IsValid() {
+						err = mergeString(eva, vl)
+					}
+				case "*wrapperspb.UInt32Value", "*wrapperspb.UInt64Value":
+					if eva := reflect.Indirect(va.Index(idx)).FieldByName("Value"); eva.IsValid() {
+						err = mergeUint(eva, vl)
+					}
+				}
+			}
+			if err != nil {
+				return err
+			}
+		}
+	default:
+		return fmt.Errorf("cant merge %v %s with %v %s", va, va.Kind(), vb, vb.Kind())
+	}
+	return nil
+}
+
+func mergeBool(va, vb reflect.Value) error {
+	switch getKind(vb) {
+	case reflect.Int:
+		if vb.Int() == 1 {
+			va.SetBool(true)
+		}
+	case reflect.Uint:
+		if vb.Uint() == 1 {
+			va.SetBool(true)
+		}
+	case reflect.Float32:
+		if vb.Float() == 1 {
+			va.SetBool(true)
+		}
+	case reflect.String:
+		if b, err := strconv.ParseBool(vb.String()); err != nil {
+			return err
+		} else {
+			va.SetBool(b)
+		}
+	default:
+		return fmt.Errorf("cant merge %v %s with %v %s", va, va.Kind(), vb, vb.Kind())
+	}
+	return nil
+}
+
+func mergeString(va, vb reflect.Value) error {
+	switch getKind(vb) {
+	case reflect.Int:
+		va.SetString(fmt.Sprintf("%d", vb.Int()))
+	case reflect.Uint:
+		va.SetString(fmt.Sprintf("%d", vb.Uint()))
+	case reflect.Float32:
+		va.SetString(fmt.Sprintf("%f", vb.Float()))
+	case reflect.String:
+		va.Set(vb)
+	default:
+		return fmt.Errorf("cant merge %v %s with %v %s", va, va.Kind(), vb, vb.Kind())
+	}
+	return nil
+}
+
+func mergeInt(va, vb reflect.Value) error {
+	switch getKind(vb) {
+	case reflect.Int:
+		va.Set(vb)
+	case reflect.Uint:
+		va.SetInt(int64(vb.Uint()))
+	case reflect.Float32:
+		va.SetInt(int64(vb.Float()))
+	case reflect.String:
+		if f, err := strconv.ParseInt(vb.String(), 10, va.Type().Bits()); err != nil {
+			return err
+		} else {
+			va.SetInt(f)
+		}
+	default:
+		return fmt.Errorf("cant merge %v %s with %v %s", va, va.Kind(), vb, vb.Kind())
+	}
+	return nil
+}
+
+func mergeUint(va, vb reflect.Value) error {
+	switch getKind(vb) {
+	case reflect.Int:
+		va.SetUint(uint64(vb.Int()))
+	case reflect.Uint:
+		va.Set(vb)
+	case reflect.Float32:
+		va.SetUint(uint64(vb.Float()))
+	case reflect.String:
+		if f, err := strconv.ParseUint(vb.String(), 10, va.Type().Bits()); err != nil {
+			return err
+		} else {
+			va.SetUint(f)
+		}
+	default:
+		return fmt.Errorf("cant merge %v %s with %v %s", va, va.Kind(), vb, vb.Kind())
+	}
+	return nil
+}
+
+func mergeFloat(va, vb reflect.Value) error {
+	switch getKind(vb) {
+	case reflect.Int:
+		va.SetFloat(float64(vb.Int()))
+	case reflect.Uint:
+		va.SetFloat(float64(vb.Uint()))
+	case reflect.Float32:
+		va.Set(vb)
+	case reflect.String:
+		if f, err := strconv.ParseFloat(vb.String(), va.Type().Bits()); err != nil {
+			return err
+		} else {
+			va.SetFloat(f)
+		}
+	default:
+		return fmt.Errorf("cant merge %v %s with %v %s", va, va.Kind(), vb, vb.Kind())
+	}
+
+	return nil
+}
+
+func getKind(val reflect.Value) reflect.Kind {
+	kind := val.Kind()
+	switch {
+	case kind >= reflect.Int && kind <= reflect.Int64:
+		return reflect.Int
+	case kind >= reflect.Uint && kind <= reflect.Uint64:
+		return reflect.Uint
+	case kind >= reflect.Float32 && kind <= reflect.Float64:
+		return reflect.Float32
+	}
+	return kind
+}
+
+func btSplitter(str string) []string {
+	r := bracketSplitter.Split(str, -1)
+	for idx, s := range r {
+		if len(s) == 0 {
+			if len(r) > idx+1 {
+				copy(r[idx:], r[idx+1:])
+				r = r[:len(r)-1]
+			}
+		}
+	}
+	return r
+}
+
+// queryToMap turns something like a[b][c]=4 into
+//   map[string]interface{}{
+//     "a": map[string]interface{}{
+// 		  "b": map[string]interface{}{
+// 			  "c": 4,
+// 		  },
+// 	  },
+//   }
+func queryToMap(param string) (map[string]interface{}, error) {
+	rawKey, rawValue, err := splitKeyAndValue(param)
+	if err != nil {
+		return nil, err
+	}
+	rawValue, err = url.QueryUnescape(rawValue)
+	if err != nil {
+		return nil, err
+	}
+	rawKey, err = url.QueryUnescape(rawKey)
+	if err != nil {
+		return nil, err
+	}
+
+	pieces := btSplitter(rawKey)
+	key := pieces[0]
+
+	// If len==1 then rawKey has no [] chars and we can just
+	// decode this as key=value into {key: value}
+	if len(pieces) == 1 {
+		return map[string]interface{}{
+			key: rawValue,
+		}, nil
+	}
+
+	// If len > 1 then we have something like a[b][c]=2
+	// so we need to turn this into {"a": {"b": {"c": 2}}}
+	// To do this we break our key into two pieces:
+	//   a and b[c]
+	// and then we set {"a": queryToMap("b[c]", value)}
+	ret := make(map[string]interface{})
+	ret[key], err = queryToMap(buildNewKey(rawKey) + "=" + rawValue)
+	if err != nil {
+		return nil, err
+	}
+
+	// When URL params have a set of empty brackets (eg a[]=1)
+	// it is assumed to be an array. This will get us the
+	// correct value for the array item and return it as an
+	// []interface{} so that it can be merged properly.
+	if pieces[1] == "" {
+		temp := ret[key].(map[string]interface{})
+		ret[key] = []interface{}{temp[""]}
+	}
+	return ret, nil
+}
+
+// buildNewKey will take something like:
+// origKey = "bar[one][two]"
+// pieces = [bar one two ]
+// and return "one[two]"
+func buildNewKey(origKey string) string {
+	pieces := btSplitter(origKey)
+
+	ret := origKey[len(pieces[0])+1:]
+	ret = ret[:len(pieces[1])] + ret[len(pieces[1])+1:]
+	return ret
+}
+
+// splitKeyAndValue splits a URL param at the last equal
+// sign and returns the two strings. If no equal sign is
+// found, the ErrInvalidParam error is returned.
+func splitKeyAndValue(param string) (string, string, error) {
+	li := strings.LastIndex(param, "=")
+	if li == -1 {
+		return "", "", ErrInvalidParam
+	}
+	return param[:li], param[li+1:], nil
+}
+
+// merge merges a with b if they are either both slices
+// or map[string]interface{} types. Otherwise it returns b.
+func merge(a interface{}, b interface{}) interface{} {
+	if av, aok := a.(map[string]interface{}); aok {
+		if bv, bok := b.(map[string]interface{}); bok {
+			return mergeMapIface(av, bv)
+		}
+	}
+	if av, aok := a.([]interface{}); aok {
+		if bv, bok := b.([]interface{}); bok {
+			return mergeSliceIface(av, bv)
+		}
+	}
+
+	va := reflect.ValueOf(a)
+	vb := reflect.ValueOf(b)
+	if (va.Type().Kind() == reflect.Slice) && (va.Type().Elem().Kind() == vb.Type().Kind() || vb.Type().ConvertibleTo(va.Type().Elem())) {
+		va = reflect.Append(va, vb.Convert(va.Type().Elem()))
+		return va.Interface()
+	}
+
+	return b
+}
+
+// mergeMap merges a with b, attempting to merge any nested
+// values in nested maps but eventually overwriting anything
+// in a that can't be merged with whatever is in b.
+func mergeMapIface(a map[string]interface{}, b map[string]interface{}) map[string]interface{} {
+	for bK, bV := range b {
+		if aV, ok := a[bK]; ok {
+			if (reflect.ValueOf(aV).Type().Kind() == reflect.ValueOf(bV).Type().Kind()) ||
+				((reflect.ValueOf(aV).Type().Kind() == reflect.Slice) && reflect.ValueOf(aV).Type().Elem().Kind() == reflect.ValueOf(bV).Type().Kind()) {
+				nV := []interface{}{aV, bV}
+				a[bK] = nV
+			} else {
+				a[bK] = merge(a[bK], bV)
+			}
+		} else {
+			a[bK] = bV
+		}
+	}
+	return a
+}
+
+// mergeSlice merges a with b and returns the result.
+func mergeSliceIface(a []interface{}, b []interface{}) []interface{} {
+	a = append(a, b...)
+	return a
+}

util/reflect/reflect_test.go

@@ -0,0 +1,45 @@
+package reflect
+
+import (
+	"net/url"
+	"testing"
+)
+
+func TestURLSliceVars(t *testing.T) {
+	u, err := url.Parse("http://localhost/v1/test/call/my_name?key=arg1&key=arg2&key=arg3")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	mp, err := URLMap(u.RawQuery)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	v, ok := mp["key"]
+	if !ok {
+		t.Fatalf("key not exists: %#+v", mp)
+	}
+
+	vm, ok := v.([]interface{})
+	if !ok {
+		t.Fatalf("invalid key value")
+	}
+
+	if len(vm) != 3 {
+		t.Fatalf("missing key value: %#+v", mp)
+	}
+}
+
+func TestURLVars(t *testing.T) {
+	u, err := url.Parse("http://localhost/v1/test/call/my_name?req=key&arg1=arg1&arg2=12345&nested.string_args=str1&nested.string_args=str2&arg2=54321")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	mp, err := URLMap(u.RawQuery)
+	if err != nil {
+		t.Fatal(err)
+	}
+	_ = mp
+}