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)
+	}
+}

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 {

go.mod

@@ -13,6 +13,7 @@ require (
 	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-20210119194325-5f4716e94777

go.sum

@@ -45,6 +45,8 @@ github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e h1:fD57ERR4JtEqsWb
 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=

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")
+	}
+}

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()")
+	}
+}

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
+}

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/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),
+	}
+}