proxy publish

fix race in http transport

README.md

@@ -1,6 +1,6 @@
 # Go Micro [![License](https://img.shields.io/:license-apache-blue.svg)](https://opensource.org/licenses/Apache-2.0) [![GoDoc](https://godoc.org/github.com/micro/go-micro?status.svg)](https://godoc.org/github.com/micro/go-micro) [![Travis CI](https://api.travis-ci.org/micro/go-micro.svg?branch=master)](https://travis-ci.org/micro/go-micro) [![Go Report Card](https://goreportcard.com/badge/micro/go-micro)](https://goreportcard.com/report/github.com/micro/go-micro)
 
-Go Micro is a pluggable framework for micro service development.
+Go Micro is a framework for micro service development.
 
 ## Overview
 
@@ -8,7 +8,7 @@ Go Micro provides the core requirements for distributed systems development incl
 The **micro** philosophy is sane defaults with a pluggable architecture. We provide defaults to get you started quickly 
 but everything can be easily swapped out. 
 
-<img src="https://micro.mu/docs/images/go-micro.png" />
+<img src="https://micro.mu/docs/images/go-micro.svg" />
 
 Plugins are available at [github.com/micro/go-plugins](https://github.com/micro/go-plugins).
 
@@ -31,7 +31,7 @@ across the services and retry a different node if there's a problem.
 to seamlessly encode and decode Go types for you. Any variety of messages could be encoded and sent from different clients. The client 
 and server handle this by default. This includes protobuf and json by default.
 
-- **Sync Streaming** - RPC based request/response with support for bidirectional streaming. We provide an abstraction for synchronous 
+- **Request/Response** - RPC based request/response with support for bidirectional streaming. We provide an abstraction for synchronous 
 communication. A request made to a service will be automatically resolved, load balanced, dialled and streamed. The default 
 transport is http/1.1 or http2 when tls is enabled.
 
@@ -52,5 +52,3 @@ See the [docs](https://micro.mu/docs/go-micro.html) for detailed information on
 Sixt is an Enterprise Sponsor of Micro
 
 <a href="https://micro.mu/blog/2016/04/25/announcing-sixt-sponsorship.html"><img src="https://micro.mu/sixt_logo.png" width=150px height="auto" /></a>
-
-Become a sponsor by backing micro on [Patreon](https://www.patreon.com/microhq)

broker/http_broker.go

@@ -19,7 +19,6 @@ import (
 	"time"
 
 	"github.com/google/uuid"
-	"github.com/micro/go-log"
 	"github.com/micro/go-micro/codec/json"
 	merr "github.com/micro/go-micro/errors"
 	"github.com/micro/go-micro/registry"
@@ -397,7 +396,6 @@ func (h *httpBroker) Connect() error {
 		return err
 	}
 
-	log.Logf("Broker Listening on %s", l.Addr().String())
 	addr := h.address
 	h.address = l.Addr().String()
 

broker/options.go

@@ -111,3 +111,10 @@ func TLSConfig(t *tls.Config) Option {
 		o.TLSConfig = t
 	}
 }
+
+// SubscribeContext set context
+func SubscribeContext(ctx context.Context) SubscribeOption {
+	return func(o *SubscribeOptions) {
+		o.Context = ctx
+	}
+}

client/options.go

@@ -65,6 +65,8 @@ type CallOptions struct {
 }
 
 type PublishOptions struct {
+	// Exchange is the routing exchange for the message
+	Exchange string
 	// Other options for implementations of the interface
 	// can be stored in a context
 	Context context.Context
@@ -236,6 +238,13 @@ func DialTimeout(d time.Duration) Option {
 
 // Call Options
 
+// WithExchange sets the exchange to route a message through
+func WithExchange(e string) PublishOption {
+	return func(o *PublishOptions) {
+		o.Exchange = e
+	}
+}
+
 // WithAddress sets the remote address to use rather than using service discovery
 func WithAddress(a string) CallOption {
 	return func(o *CallOptions) {

client/rpc_client.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"fmt"
 	"net"
+	"os"
 	"strconv"
 	"sync"
 	"sync/atomic"
@@ -213,12 +214,17 @@ func (r *rpcClient) stream(ctx context.Context, node *registry.Node, req Request
 		codec:  codec,
 	}
 
+	// set request codec
+	if r, ok := req.(*rpcRequest); ok {
+		r.codec = codec
+	}
+
 	stream := &rpcStream{
 		context:  ctx,
 		request:  req,
 		response: rsp,
 		closed:   make(chan bool),
-		codec:    newRpcCodec(msg, c, cf),
+		codec:    codec,
 	}
 
 	ch := make(chan error, 1)
@@ -268,6 +274,18 @@ func (r *rpcClient) Options() Options {
 }
 
 func (r *rpcClient) next(request Request, opts CallOptions) (selector.Next, error) {
+	service := request.Service()
+
+	// get proxy
+	if prx := os.Getenv("MICRO_PROXY"); len(prx) > 0 {
+		service = prx
+	}
+
+	// get proxy address
+	if prx := os.Getenv("MICRO_PROXY_ADDRESS"); len(prx) > 0 {
+		opts.Address = prx
+	}
+
 	// return remote address
 	if len(opts.Address) > 0 {
 		address := opts.Address
@@ -288,11 +306,11 @@ func (r *rpcClient) next(request Request, opts CallOptions) (selector.Next, erro
 	}
 
 	// get next nodes from the selector
-	next, err := r.opts.Selector.Select(request.Service(), opts.SelectOptions...)
+	next, err := r.opts.Selector.Select(service, opts.SelectOptions...)
 	if err != nil && err == selector.ErrNotFound {
-		return nil, errors.NotFound("go.micro.client", "service %s: %v", request.Service(), err.Error())
+		return nil, errors.NotFound("go.micro.client", "service %s: %v", service, err.Error())
 	} else if err != nil {
-		return nil, errors.InternalServerError("go.micro.client", "error selecting %s node: %v", request.Service(), err.Error())
+		return nil, errors.InternalServerError("go.micro.client", "error selecting %s node: %v", service, err.Error())
 	}
 
 	return next, nil
@@ -480,6 +498,13 @@ func (r *rpcClient) Stream(ctx context.Context, request Request, opts ...CallOpt
 }
 
 func (r *rpcClient) Publish(ctx context.Context, msg Message, opts ...PublishOption) error {
+	options := PublishOptions{
+		Context: context.Background(),
+	}
+	for _, o := range opts {
+		o(&options)
+	}
+
 	md, ok := metadata.FromContext(ctx)
 	if !ok {
 		md = make(map[string]string)
@@ -487,8 +512,21 @@ func (r *rpcClient) Publish(ctx context.Context, msg Message, opts ...PublishOpt
 
 	id := uuid.New().String()
 	md["Content-Type"] = msg.ContentType()
-	md["X-Micro-Topic"] = msg.Topic()
-	md["X-Micro-Id"] = id
+	md["Micro-Topic"] = msg.Topic()
+	md["Micro-Id"] = id
+
+	// set the topic
+	topic := msg.Topic()
+
+	// get proxy
+	if prx := os.Getenv("MICRO_PROXY"); len(prx) > 0 {
+		options.Exchange = prx
+	}
+
+	// get the exchange
+	if len(options.Exchange) > 0 {
+		topic = options.Exchange
+	}
 
 	// encode message body
 	cf, err := r.newCodec(msg.ContentType())
@@ -497,11 +535,11 @@ func (r *rpcClient) Publish(ctx context.Context, msg Message, opts ...PublishOpt
 	}
 	b := &buffer{bytes.NewBuffer(nil)}
 	if err := cf(b).Write(&codec.Message{
-		Target: msg.Topic(),
+		Target: topic,
 		Type:   codec.Publication,
 		Header: map[string]string{
-			"X-Micro-Id":    id,
-			"X-Micro-Topic": msg.Topic(),
+			"Micro-Id":    id,
+			"Micro-Topic": msg.Topic(),
 		},
 	}, msg.Payload()); err != nil {
 		return errors.InternalServerError("go.micro.client", err.Error())
@@ -510,7 +548,7 @@ func (r *rpcClient) Publish(ctx context.Context, msg Message, opts ...PublishOpt
 		r.opts.Broker.Connect()
 	})
 
-	return r.opts.Broker.Publish(msg.Topic(), &broker.Message{
+	return r.opts.Broker.Publish(topic, &broker.Message{
 		Header: md,
 		Body:   b.Bytes(),
 	})

client/rpc_codec.go

@@ -84,6 +84,50 @@ func (rwc *readWriteCloser) Close() error {
 	return nil
 }
 
+func getHeaders(m *codec.Message) {
+	get := func(hdr string) string {
+		if hd := m.Header[hdr]; len(hd) > 0 {
+			return hd
+		}
+		// old
+		return m.Header["X-"+hdr]
+	}
+
+	// check error in header
+	if len(m.Error) == 0 {
+		m.Error = get("Micro-Error")
+	}
+
+	// check endpoint in header
+	if len(m.Endpoint) == 0 {
+		m.Endpoint = get("Micro-Endpoint")
+	}
+
+	// check method in header
+	if len(m.Method) == 0 {
+		m.Method = get("Micro-Method")
+	}
+
+	if len(m.Id) == 0 {
+		m.Id = get("Micro-Id")
+	}
+}
+
+func setHeaders(m *codec.Message) {
+	set := func(hdr, v string) {
+		if len(v) == 0 {
+			return
+		}
+		m.Header[hdr] = v
+		m.Header["X-"+hdr] = v
+	}
+
+	set("Micro-Id", m.Id)
+	set("Micro-Service", m.Target)
+	set("Micro-Method", m.Method)
+	set("Micro-Endpoint", m.Endpoint)
+}
+
 // setupProtocol sets up the old protocol
 func setupProtocol(msg *transport.Message, node *registry.Node) codec.NewCodec {
 	protocol := node.Metadata["protocol"]
@@ -133,10 +177,7 @@ func (c *rpcCodec) Write(m *codec.Message, body interface{}) error {
 	}
 
 	// set the mucp headers
-	m.Header["X-Micro-Id"] = m.Id
-	m.Header["X-Micro-Service"] = m.Target
-	m.Header["X-Micro-Method"] = m.Method
-	m.Header["X-Micro-Endpoint"] = m.Endpoint
+	setHeaders(m)
 
 	// if body is bytes Frame don't encode
 	if body != nil {
@@ -162,52 +203,32 @@ func (c *rpcCodec) Write(m *codec.Message, body interface{}) error {
 		Header: m.Header,
 		Body:   m.Body,
 	}
-
 	// send the request
 	if err := c.client.Send(&msg); err != nil {
 		return errors.InternalServerError("go.micro.client.transport", err.Error())
 	}
-
 	return nil
 }
 
-func (c *rpcCodec) ReadHeader(wm *codec.Message, r codec.MessageType) error {
-	var m transport.Message
-	if err := c.client.Recv(&m); err != nil {
+func (c *rpcCodec) ReadHeader(m *codec.Message, r codec.MessageType) error {
+	var tm transport.Message
+
+	// read message from transport
+	if err := c.client.Recv(&tm); err != nil {
 		return errors.InternalServerError("go.micro.client.transport", err.Error())
 	}
-	c.buf.rbuf.Reset()
-	c.buf.rbuf.Write(m.Body)
 
-	var me codec.Message
-	// set headers
-	me.Header = m.Header
+	c.buf.rbuf.Reset()
+	c.buf.rbuf.Write(tm.Body)
+
+	// set headers from transport
+	m.Header = tm.Header
 
 	// read header
-	err := c.codec.ReadHeader(&me, r)
-	wm.Endpoint = me.Endpoint
-	wm.Method = me.Method
-	wm.Id = me.Id
-	wm.Error = me.Error
+	err := c.codec.ReadHeader(m, r)
 
-	// check error in header
-	if len(me.Error) == 0 {
-		wm.Error = me.Header["X-Micro-Error"]
-	}
-
-	// check endpoint in header
-	if len(me.Endpoint) == 0 {
-		wm.Endpoint = me.Header["X-Micro-Endpoint"]
-	}
-
-	// check method in header
-	if len(me.Method) == 0 {
-		wm.Method = me.Header["X-Micro-Method"]
-	}
-
-	if len(me.Id) == 0 {
-		wm.Id = me.Header["X-Micro-Id"]
-	}
+	// get headers
+	getHeaders(m)
 
 	// return header error
 	if err != nil {

cmd/cmd.go

@@ -404,6 +404,10 @@ func (c *cmd) Before(ctx *cli.Context) error {
 		serverOpts = append(serverOpts, server.RegisterTTL(ttl*time.Second))
 	}
 
+	if val := time.Duration(ctx.GlobalInt("register_interval")); val > 0 {
+		serverOpts = append(serverOpts, server.RegisterInterval(val*time.Second))
+	}
+
 	// client opts
 	if r := ctx.Int("client_retries"); r >= 0 {
 		clientOpts = append(clientOpts, client.Retries(r))

codec/grpc/grpc.go

@@ -29,8 +29,8 @@ func (c *Codec) ReadHeader(m *codec.Message, t codec.MessageType) error {
 	// service method
 	path := m.Header[":path"]
 	if len(path) == 0 || path[0] != '/' {
-		m.Target = m.Header["X-Micro-Service"]
-		m.Endpoint = m.Header["X-Micro-Endpoint"]
+		m.Target = m.Header["Micro-Service"]
+		m.Endpoint = m.Header["Micro-Endpoint"]
 	} else {
 		// [ , a.package.Foo, Bar]
 		parts := strings.Split(path, "/")

go.mod

@@ -0,0 +1,22 @@
+module github.com/micro/go-micro
+
+require (
+	github.com/armon/circbuf v0.0.0-20190214190532-5111143e8da2 // indirect
+	github.com/go-log/log v0.1.0
+	github.com/golang/protobuf v1.2.0
+	github.com/google/uuid v1.1.0
+	github.com/hashicorp/consul v1.4.2
+	github.com/hashicorp/memberlist v0.1.3
+	github.com/mattn/go-colorable v0.1.1 // indirect
+	github.com/micro/cli v0.1.0
+	github.com/micro/go-log v0.1.0
+	github.com/micro/go-rcache v0.2.1
+	github.com/micro/h2c v1.0.0
+	github.com/micro/mdns v0.1.0
+	github.com/micro/util v0.2.0
+	github.com/mitchellh/hashstructure v1.0.0
+	github.com/pkg/errors v0.8.1
+	golang.org/x/crypto v0.0.0-20190222235706-ffb98f73852f // indirect
+	golang.org/x/net v0.0.0-20190213061140-3a22650c66bd
+	golang.org/x/sys v0.0.0-20190222171317-cd391775e71e // indirect
+)

go.sum

@@ -0,0 +1,136 @@
+github.com/armon/circbuf v0.0.0-20150827004946-bbbad097214e/go.mod h1:3U/XgcO3hCbHZ8TKRvWD2dDTCfh9M9ya+I9JpbB7O8o=
+github.com/armon/circbuf v0.0.0-20190214190532-5111143e8da2/go.mod h1:3U/XgcO3hCbHZ8TKRvWD2dDTCfh9M9ya+I9JpbB7O8o=
+github.com/armon/go-metrics v0.0.0-20180917152333-f0300d1749da h1:8GUt8eRujhVEGZFFEjBj46YV4rDjvGrNxb0KMWYkL2I=
+github.com/armon/go-metrics v0.0.0-20180917152333-f0300d1749da/go.mod h1:Q73ZrmVTwzkszR9V5SSuryQ31EELlFMUz1kKyl939pY=
+github.com/armon/go-radix v0.0.0-20180808171621-7fddfc383310/go.mod h1:ufUuZ+zHj4x4TnLV4JWEpy2hxWSpsRywHrMgIH9cCH8=
+github.com/armon/go-radix v1.0.0/go.mod h1:ufUuZ+zHj4x4TnLV4JWEpy2hxWSpsRywHrMgIH9cCH8=
+github.com/bgentry/speakeasy v0.1.0/go.mod h1:+zsyZBPWlz7T6j88CTgSN5bM796AkVf0kBD4zp0CCIs=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4=
+github.com/go-log/log v0.1.0 h1:wudGTNsiGzrD5ZjgIkVZ517ugi2XRe9Q/xRCzwEO4/U=
+github.com/go-log/log v0.1.0/go.mod h1:4mBwpdRMFLiuXZDCwU2lKQFsoSCo72j3HqBK9d81N2M=
+github.com/golang/protobuf v1.2.0 h1:P3YflyNX/ehuJFLhxviNdFxQPkGK5cDcApsge1SqnvM=
+github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
+github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c h1:964Od4U6p2jUkFxvCydnIczKteheJEzHRToSGK3Bnlw=
+github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
+github.com/google/uuid v1.1.0 h1:Jf4mxPC/ziBnoPIdpQdPJ9OeiomAUHLvxmPRSPH9m4s=
+github.com/google/uuid v1.1.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/hashicorp/consul v1.4.2 h1:D9iJoJb8Ehe/Zmr+UEE3U3FjOLZ4LUxqFMl4O43BM1U=
+github.com/hashicorp/consul v1.4.2/go.mod h1:mFrjN1mfidgJfYP1xrJCF+AfRhr6Eaqhb2+sfyn/OOI=
+github.com/hashicorp/errwrap v1.0.0 h1:hLrqtEDnRye3+sgx6z4qVLNuviH3MR5aQ0ykNJa/UYA=
+github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
+github.com/hashicorp/go-cleanhttp v0.5.0 h1:wvCrVc9TjDls6+YGAF2hAifE1E5U1+b4tH6KdvN3Gig=
+github.com/hashicorp/go-cleanhttp v0.5.0/go.mod h1:JpRdi6/HCYpAwUzNwuwqhbovhLtngrth3wmdIIUrZ80=
+github.com/hashicorp/go-immutable-radix v1.0.0 h1:AKDB1HM5PWEA7i4nhcpwOrO2byshxBjXVn/J/3+z5/0=
+github.com/hashicorp/go-immutable-radix v1.0.0/go.mod h1:0y9vanUI8NX6FsYoO3zeMjhV/C5i9g4Q3DwcSNZ4P60=
+github.com/hashicorp/go-msgpack v0.5.3 h1:zKjpN5BK/P5lMYrLmBHdBULWbJ0XpYR+7NGzqkZzoD4=
+github.com/hashicorp/go-msgpack v0.5.3/go.mod h1:ahLV/dePpqEmjfWmKiqvPkv/twdG7iPBM1vqhUKIvfM=
+github.com/hashicorp/go-multierror v1.0.0 h1:iVjPR7a6H0tWELX5NxNe7bYopibicUzc7uPribsnS6o=
+github.com/hashicorp/go-multierror v1.0.0/go.mod h1:dHtQlpGsu+cZNNAkkCN/P3hoUDHhCYQXV3UM06sGGrk=
+github.com/hashicorp/go-rootcerts v1.0.0 h1:Rqb66Oo1X/eSV1x66xbDccZjhJigjg0+e82kpwzSwCI=
+github.com/hashicorp/go-rootcerts v1.0.0/go.mod h1:K6zTfqpRlCUIjkwsN4Z+hiSfzSTQa6eBIzfwKfwNnHU=
+github.com/hashicorp/go-sockaddr v1.0.0/go.mod h1:7Xibr9yA9JjQq1JpNB2Vw7kxv8xerXegt+ozgdvDeDU=
+github.com/hashicorp/go-sockaddr v1.0.1 h1:eCkkJ5KOOktDvwbsE9KPyiBWaOfp1ZNy2gLHgL8PSBM=
+github.com/hashicorp/go-sockaddr v1.0.1/go.mod h1:rB4wwRAUzs07qva3c5SdrY/NEtAUjGlgmH/UkBUC97A=
+github.com/hashicorp/go-syslog v1.0.0/go.mod h1:qPfqrKkXGihmCqbJM2mZgkZGvKG1dFdvsLplgctolz4=
+github.com/hashicorp/go-uuid v1.0.0/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
+github.com/hashicorp/go-uuid v1.0.1/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
+github.com/hashicorp/go.net v0.0.1/go.mod h1:hjKkEWcCURg++eb33jQU7oqQcI9XDCnUzHA0oac0k90=
+github.com/hashicorp/golang-lru v0.5.0 h1:CL2msUPvZTLb5O648aiLNJw3hnBxN2+1Jq8rCOH9wdo=
+github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
+github.com/hashicorp/logutils v1.0.0/go.mod h1:QIAnNjmIWmVIIkWDTG1z5v++HQmx9WQRO+LraFDTW64=
+github.com/hashicorp/mdns v1.0.0/go.mod h1:tL+uN++7HEJ6SQLQ2/p+z2pH24WQKWjBPkE0mNTz8vQ=
+github.com/hashicorp/memberlist v0.1.3 h1:EmmoJme1matNzb+hMpDuR/0sbJSUisxyqBGG676r31M=
+github.com/hashicorp/memberlist v0.1.3/go.mod h1:ajVTdAv/9Im8oMAAj5G31PhhMCZJV2pPBoIllUwCN7I=
+github.com/hashicorp/serf v0.8.2 h1:YZ7UKsJv+hKjqGVUUbtE3HNj79Eln2oQ75tniF6iPt0=
+github.com/hashicorp/serf v0.8.2/go.mod h1:6hOLApaqBFA1NXqRQAsxw9QxuDEvNxSQRwA/JwenrHc=
+github.com/mattn/go-colorable v0.0.9/go.mod h1:9vuHe8Xs5qXnSaW/c/ABM9alt+Vo+STaOChaDxuIBZU=
+github.com/mattn/go-colorable v0.1.0/go.mod h1:9vuHe8Xs5qXnSaW/c/ABM9alt+Vo+STaOChaDxuIBZU=
+github.com/mattn/go-colorable v0.1.1/go.mod h1:FuOcm+DKB9mbwrcAfNl7/TZVBZ6rcnceauSikq3lYCQ=
+github.com/mattn/go-isatty v0.0.3/go.mod h1:M+lRXTBqGeGNdLjl/ufCoiOlB5xdOkqRJdNxMWT7Zi4=
+github.com/mattn/go-isatty v0.0.4/go.mod h1:M+lRXTBqGeGNdLjl/ufCoiOlB5xdOkqRJdNxMWT7Zi4=
+github.com/mattn/go-isatty v0.0.5/go.mod h1:Iq45c/XA43vh69/j3iqttzPXn0bhXyGjM0Hdxcsrc5s=
+github.com/micro/cli v0.0.0-20181223203424-1b0c9793c300/go.mod h1:x9x6qy+tXv17jzYWQup462+j3SIUgDa6vVTzU4IXy/w=
+github.com/micro/cli v0.1.0 h1:5DT+QdbAPPQvB3gYTgwze7tFO1m+7DU1sz9XfQczbsc=
+github.com/micro/cli v0.1.0/go.mod h1:jRT9gmfVKWSS6pkKcXQ8YhUyj6bzwxK8Fp5b0Y7qNnk=
+github.com/micro/go-log v0.1.0 h1:szYSR+yyTsomZM2jyinJC5562DlqffSjHmTZFaeZ2vY=
+github.com/micro/go-log v0.1.0/go.mod h1:qaFBF6d6Jk01Gz4cbMkCA2vVuLk3FSaLLjmEGrMCreA=
+github.com/micro/go-micro v0.23.0/go.mod h1:3z3lfMkNU9Sr1L/CxL++8pVJmQapRo0N6kNjwYDtOVs=
+github.com/micro/go-micro v0.26.0/go.mod h1:CweCFO/pq8dCSIOdzVZ4ooIpUrKlyJ0AcFB269M7PgU=
+github.com/micro/go-micro v0.26.1/go.mod h1:Jgc5gPEmDiG1TWE5Qnzzx5qyXnU9VTXKT1FkXkfvt8g=
+github.com/micro/go-rcache v0.1.0 h1:YTIgANVHgBe1XOQ/yLICL+s2gbZCAdW+c2ckhekjkuc=
+github.com/micro/go-rcache v0.1.0/go.mod h1:INzyZjXO5M+PmN2A33YxD4TaOY61xjFIM4CfSHv+At8=
+github.com/micro/go-rcache v0.2.0/go.mod h1:EoiTwbY2ubQ6lc3ScV+SnmKbelDzeFezDxPDvF8XDxw=
+github.com/micro/go-rcache v0.2.1 h1:hx24BZuhW4UVCyLE8p6fDrUetXrYDlOYSn5DSmqcjms=
+github.com/micro/go-rcache v0.2.1/go.mod h1:aPCNY3RbjBdyd6ShLENl4MDSgpAiWIU4LyNLE9+TOEo=
+github.com/micro/h2c v1.0.0/go.mod h1:54sOOQW/GRlHhH43vKwOhUb+kHaXhVxR0d3CJhn9alE=
+github.com/micro/mdns v0.0.0-20181201230301-9c3770d4057a/go.mod h1:SQG6o/94RinohLuB5noHSevg2Iqg2wXLDUn4lj2LWWo=
+github.com/micro/mdns v0.1.0 h1:fuLybUsfynbigJmCot/54i+gwe0hpc/vtCMvWt2WfDI=
+github.com/micro/mdns v0.1.0/go.mod h1:KJ0dW7KmicXU2BV++qkLlmHYcVv7/hHnbtguSWt9Aoc=
+github.com/micro/util v0.1.0 h1:ghhF5KKRNlKMexzK+cWo6W6uRAZdKy1UKG/9O74NCYc=
+github.com/micro/util v0.1.0/go.mod h1:MZgOs0nwxzv9k4xQo4fpF9IwZGF2O96F5/phP9X4/Sw=
+github.com/micro/util v0.2.0 h1:6u0cPj1TeixEk5cAR9jbcVRUWDQsmCaZvDBiM3zFZuA=
+github.com/micro/util v0.2.0/go.mod h1:SgRDkxJJluC2ZNiPfINY42ObEaCAFjL3jP5a+u+qRLU=
+github.com/miekg/dns v1.0.14/go.mod h1:W1PPwlIAgtquWBMBEV9nkV9Cazfe8ScdGz/Lj7v3Nrg=
+github.com/miekg/dns v1.1.3/go.mod h1:W1PPwlIAgtquWBMBEV9nkV9Cazfe8ScdGz/Lj7v3Nrg=
+github.com/miekg/dns v1.1.4 h1:rCMZsU2ScVSYcAsOXgmC6+AKOK+6pmQTOcw03nfwYV0=
+github.com/miekg/dns v1.1.4/go.mod h1:W1PPwlIAgtquWBMBEV9nkV9Cazfe8ScdGz/Lj7v3Nrg=
+github.com/mitchellh/cli v1.0.0/go.mod h1:hNIlj7HEI86fIcpObd7a0FcrxTWetlwJDGcceTlRvqc=
+github.com/mitchellh/go-homedir v1.0.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=
+github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=
+github.com/mitchellh/go-wordwrap v1.0.0/go.mod h1:ZXFpozHsX6DPmq2I0TCekCxypsnAUbP2oI0UX1GXzOo=
+github.com/mitchellh/gox v0.4.0/go.mod h1:Sd9lOJ0+aimLBi73mGofS1ycjY8lL3uZM3JPS42BGNg=
+github.com/mitchellh/hashstructure v1.0.0 h1:ZkRJX1CyOoTkar7p/mLS5TZU4nJ1Rn/F8u9dGS02Q3Y=
+github.com/mitchellh/hashstructure v1.0.0/go.mod h1:QjSHrPWS+BGUVBYkbTZWEnOh3G1DutKwClXU/ABz6AQ=
+github.com/mitchellh/iochan v1.0.0/go.mod h1:JwYml1nuB7xOzsp52dPpHFffvOCDupsG0QubkSMEySY=
+github.com/mitchellh/mapstructure v0.0.0-20160808181253-ca63d7c062ee/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh9fWfEaFds41c1Y=
+github.com/mitchellh/mapstructure v1.1.2 h1:fmNYVwqnSfB9mZU6OS2O6GsXM+wcskZDuKQzvN1EDeE=
+github.com/mitchellh/mapstructure v1.1.2/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh9fWfEaFds41c1Y=
+github.com/pascaldekloe/goe v0.0.0-20180627143212-57f6aae5913c/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc=
+github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I=
+github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/posener/complete v1.1.1/go.mod h1:em0nMJCgc9GFtwrmVmEMR/ZL6WyhyjMBndrE9hABlRI=
+github.com/posener/complete v1.2.1/go.mod h1:6gapUrK/U1TAN7ciCoNRIdVC5sbdBTUh1DKN0g6uH7E=
+github.com/ryanuber/columnize v0.0.0-20160712163229-9b3edd62028f/go.mod h1:sm1tb6uqfes/u+d4ooFouqFdy9/2g9QGwK3SQygK0Ts=
+github.com/ryanuber/columnize v2.1.0+incompatible/go.mod h1:sm1tb6uqfes/u+d4ooFouqFdy9/2g9QGwK3SQygK0Ts=
+github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529 h1:nn5Wsu0esKSJiIVhscUtVbo7ada43DJhG55ua/hjS5I=
+github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg65j358z/aeFdxmN0P9QXhEzd20vsDc=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
+github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+golang.org/x/crypto v0.0.0-20181029021203-45a5f77698d3/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
+golang.org/x/crypto v0.0.0-20190130090550-b01c7a725664/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
+golang.org/x/crypto v0.0.0-20190131182504-b8fe1690c613 h1:MQ/ZZiDsUapFFiMS+vzwXkCTeEKaum+Do5rINYJDmxc=
+golang.org/x/crypto v0.0.0-20190131182504-b8fe1690c613/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
+golang.org/x/crypto v0.0.0-20190211182817-74369b46fc67 h1:ng3VDlRp5/DHpSWl02R4rM9I+8M2rhmsuLwAMmkLQWE=
+golang.org/x/crypto v0.0.0-20190211182817-74369b46fc67/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
+golang.org/x/crypto v0.0.0-20190222235706-ffb98f73852f h1:qWFY9ZxP3tfI37wYIs/MnIAqK0vlXp1xnYEa5HxFSSY=
+golang.org/x/crypto v0.0.0-20190222235706-ffb98f73852f/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
+golang.org/x/net v0.0.0-20181023162649-9b4f9f5ad519/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
+golang.org/x/net v0.0.0-20181201002055-351d144fa1fc/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
+golang.org/x/net v0.0.0-20190125091013-d26f9f9a57f3/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
+golang.org/x/net v0.0.0-20190206173232-65e2d4e15006 h1:bfLnR+k0tq5Lqt6dflRLcZiz6UaXCMt3vhYJ1l4FQ80=
+golang.org/x/net v0.0.0-20190206173232-65e2d4e15006/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
+golang.org/x/net v0.0.0-20190213061140-3a22650c66bd h1:HuTn7WObtcDo9uEEU7rEqL0jYthdXAmZ6PP+meazmaU=
+golang.org/x/net v0.0.0-20190213061140-3a22650c66bd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
+golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sys v0.0.0-20180823144017-11551d06cbcc/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20181026203630-95b1ffbd15a5/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20190129075346-302c3dd5f1cc/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20190204203706-41f3e6584952 h1:FDfvYgoVsA7TTZSbgiqjAbfPbK47CNHdWl3h/PJtii0=
+golang.org/x/sys v0.0.0-20190204203706-41f3e6584952/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20190209173611-3b5209105503 h1:5SvYFrOM3W8Mexn9/oA44Ji7vhXAZQ9hiP+1Q/DMrWg=
+golang.org/x/sys v0.0.0-20190209173611-3b5209105503/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20190213121743-983097b1a8a3 h1:+KlxhGbYkFs8lMfwKn+2ojry1ID5eBSMXprS2u/wqCE=
+golang.org/x/sys v0.0.0-20190213121743-983097b1a8a3/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20190222072716-a9d3bda3a223/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20190222171317-cd391775e71e h1:oF7qaQxUH6KzFdKN4ww7NpPdo53SZi4UlcksLrb2y/o=
+golang.org/x/sys v0.0.0-20190222171317-cd391775e71e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/text v0.3.0 h1:g61tztE5qeGQ89tm6NTjjM9VPIm088od1l6aSorWRWg=
+golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=

micro.go

@@ -1,4 +1,4 @@
-// Package micro is a pluggable RPC framework for microservices
+// Package micro is a pluggable framework for microservices
 package micro
 
 import (
@@ -42,7 +42,7 @@ type Publisher interface {
 type Option func(*Options)
 
 var (
-	HeaderPrefix = "X-Micro-"
+	HeaderPrefix = "Micro-"
 )
 
 // NewService creates and returns a new Service based on the packages within.

options.go

@@ -22,9 +22,6 @@ type Options struct {
 	Registry  registry.Registry
 	Transport transport.Transport
 
-	// Register loop interval
-	RegisterInterval time.Duration
-
 	// Before and After funcs
 	BeforeStart []func() error
 	BeforeStop  []func() error
@@ -125,6 +122,13 @@ func Transport(t transport.Transport) Option {
 
 // Convenience options
 
+// Address sets the address of the server
+func Address(addr string) Option {
+	return func(o *Options) {
+		o.Server.Init(server.Address(addr))
+	}
+}
+
 // Name of the service
 func Name(n string) Option {
 	return func(o *Options) {
@@ -168,7 +172,7 @@ func RegisterTTL(t time.Duration) Option {
 // RegisterInterval specifies the interval on which to re-register
 func RegisterInterval(t time.Duration) Option {
 	return func(o *Options) {
-		o.RegisterInterval = t
+		o.Server.Init(server.RegisterInterval(t))
 	}
 }
 

registry/gossip/README.md

@@ -1,6 +1,6 @@
 # Gossip Registry
 
-Gossip is a zero dependency registry which uses hashicorp/memberlist to broadcast registry information 
+Gossip is a zero dependency registry which uses github.com/hashicorp/memberlist to broadcast registry information 
 via the SWIM protocol. 
 
 ## Usage
@@ -20,5 +20,5 @@ On startup you'll see something like
 To join this gossip ring set the registry address using flag or env var
 
 ```bash
-MICRO_REGISTRY_ADDRESS= 192.168.1.65:56390 
+MICRO_REGISTRY_ADDRESS=192.168.1.65:56390
 ```

registry/gossip/gossip.go

@@ -1,9 +1,10 @@
-// Package Gossip provides a gossip registry based on hashicorp/memberlist
+// Package gossip provides a gossip registry based on hashicorp/memberlist
 package gossip
 
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"io/ioutil"
 	"net"
 	"os"
@@ -21,10 +22,42 @@ import (
 	"github.com/mitchellh/hashstructure"
 )
 
+// use registry.Result int32 values after it switches from string to int32 types
+// type actionType int32
+// type updateType int32
+
 const (
-	addAction  = "update"
-	delAction  = "delete"
-	syncAction = "sync"
+	actionTypeInvalid int32 = iota
+	actionTypeCreate
+	actionTypeDelete
+	actionTypeUpdate
+	actionTypeSync
+)
+
+const (
+	nodeActionUnknown int32 = iota
+	nodeActionJoin
+	nodeActionLeave
+	nodeActionUpdate
+)
+
+func actionTypeString(t int32) string {
+	switch t {
+	case actionTypeCreate:
+		return "create"
+	case actionTypeDelete:
+		return "delete"
+	case actionTypeUpdate:
+		return "update"
+	case actionTypeSync:
+		return "sync"
+	}
+	return "invalid"
+}
+
+const (
+	updateTypeInvalid int32 = iota
+	updateTypeService
 )
 
 type broadcast struct {
@@ -37,18 +70,45 @@ type delegate struct {
 	updates chan *update
 }
 
-type gossipRegistry struct {
-	queue    *memberlist.TransmitLimitedQueue
-	updates  chan *update
-	options  registry.Options
-	member   *memberlist.Memberlist
-	interval time.Duration
+type event struct {
+	action int32
+	node   string
+}
 
+type eventDelegate struct {
+	events chan *event
+}
+
+func (ed *eventDelegate) NotifyJoin(n *memberlist.Node) {
+	ed.events <- &event{action: nodeActionJoin, node: n.Address()}
+}
+func (ed *eventDelegate) NotifyLeave(n *memberlist.Node) {
+	ed.events <- &event{action: nodeActionLeave, node: n.Address()}
+}
+func (ed *eventDelegate) NotifyUpdate(n *memberlist.Node) {
+	ed.events <- &event{action: nodeActionUpdate, node: n.Address()}
+}
+
+type gossipRegistry struct {
+	queue       *memberlist.TransmitLimitedQueue
+	updates     chan *update
+	events      chan *event
+	options     registry.Options
+	member      *memberlist.Memberlist
+	interval    time.Duration
+	tcpInterval time.Duration
+
+	connectRetry   bool
+	connectTimeout time.Duration
 	sync.RWMutex
 	services map[string][]*registry.Service
 
-	s        sync.RWMutex
 	watchers map[string]chan *registry.Result
+
+	mtu     int
+	addrs   []string
+	members map[string]int32
+	done    chan bool
 }
 
 type update struct {
@@ -57,10 +117,16 @@ type update struct {
 	sync    chan *registry.Service
 }
 
+type updates struct {
+	sync.RWMutex
+	services map[uint64]*update
+}
+
 var (
 	// You should change this if using secure
 	DefaultSecret = []byte("micro-gossip-key") // exactly 16 bytes
-	ExpiryTick    = time.Second * 5
+	ExpiryTick    = time.Second * 1            // needs to be smaller than registry.RegisterTTL
+	MaxPacketSize = 512
 )
 
 func configure(g *gossipRegistry, opts ...registry.Option) error {
@@ -92,34 +158,33 @@ func configure(g *gossipRegistry, opts ...registry.Option) error {
 	}
 
 	// shutdown old member
-	if g.member != nil {
-		g.member.Shutdown()
-	}
+	g.Stop()
+	// new done chan
+	g.done = make(chan bool)
 
 	// replace addresses
 	curAddrs = newAddrs
 
-	// create a queue
-	queue := &memberlist.TransmitLimitedQueue{
-		NumNodes: func() int {
-			return len(curAddrs)
-		},
-		RetransmitMult: 3,
-	}
-
 	// create a new default config
 	c := memberlist.DefaultLocalConfig()
 
-	if optConfig, ok := g.options.Context.Value(contextConfig{}).(*memberlist.Config); ok && optConfig != nil {
-		c = optConfig
+	// sane good default options
+	c.LogOutput = ioutil.Discard // log to /dev/null
+	c.PushPullInterval = 0       // disable expensive tcp push/pull
+	c.ProtocolVersion = 4        // suport latest stable features
+
+	// set config from options
+	if config, ok := g.options.Context.Value(configKey{}).(*memberlist.Config); ok && config != nil {
+		c = config
 	}
 
-	if hostport, ok := g.options.Context.Value(contextAddress{}).(string); ok {
-		host, port, err := net.SplitHostPort(hostport)
+	// set address
+	if address, ok := g.options.Context.Value(addressKey{}).(string); ok {
+		host, port, err := net.SplitHostPort(address)
 		if err == nil {
-			pn, err := strconv.Atoi(port)
+			p, err := strconv.Atoi(port)
 			if err == nil {
-				c.BindPort = pn
+				c.BindPort = p
 			}
 			c.BindAddr = host
 		}
@@ -128,12 +193,13 @@ func configure(g *gossipRegistry, opts ...registry.Option) error {
 		c.BindPort = 0
 	}
 
-	if hostport, ok := g.options.Context.Value(contextAdvertise{}).(string); ok {
-		host, port, err := net.SplitHostPort(hostport)
+	// set the advertise address
+	if advertise, ok := g.options.Context.Value(advertiseKey{}).(string); ok {
+		host, port, err := net.SplitHostPort(advertise)
 		if err == nil {
-			pn, err := strconv.Atoi(port)
+			p, err := strconv.Atoi(port)
 			if err == nil {
-				c.AdvertisePort = pn
+				c.AdvertisePort = p
 			}
 			c.AdvertiseAddr = host
 		}
@@ -145,23 +211,44 @@ func configure(g *gossipRegistry, opts ...registry.Option) error {
 	// set the name
 	c.Name = strings.Join([]string{"micro", hostname, uuid.New().String()}, "-")
 
+	// set a secret key if secure
+	if g.options.Secure {
+		k, ok := g.options.Context.Value(secretKey{}).([]byte)
+		if !ok {
+			// use the default secret
+			k = DefaultSecret
+		}
+		c.SecretKey = k
+	}
+
+	// set connect retry
+	if v, ok := g.options.Context.Value(connectRetryKey{}).(bool); ok && v {
+		g.connectRetry = true
+	}
+
+	// set connect timeout
+	if td, ok := g.options.Context.Value(connectTimeoutKey{}).(time.Duration); ok {
+		g.connectTimeout = td
+	}
+
+	// create a queue
+	queue := &memberlist.TransmitLimitedQueue{
+		NumNodes: func() int {
+			return len(curAddrs)
+		},
+		RetransmitMult: 3,
+	}
+
 	// set the delegate
 	c.Delegate = &delegate{
 		updates: g.updates,
 		queue:   queue,
 	}
 
-	// log to dev null
-	c.LogOutput = ioutil.Discard
-
-	// set a secret key if secure
-	if g.options.Secure {
-		k, ok := g.options.Context.Value(contextSecretKey{}).([]byte)
-		if !ok {
-			// use the default secret
-			k = DefaultSecret
+	if g.connectRetry {
+		c.Events = &eventDelegate{
+			events: g.events,
 		}
-		c.SecretKey = k
 	}
 
 	// create the memberlist
@@ -170,46 +257,33 @@ func configure(g *gossipRegistry, opts ...registry.Option) error {
 		return err
 	}
 
-	// join the memberlist
+	// set internals
+	g.Lock()
+
 	if len(curAddrs) > 0 {
-		_, err := m.Join(curAddrs)
-		if err != nil {
-			return err
+		for _, addr := range curAddrs {
+			g.members[addr] = nodeActionUnknown
 		}
 	}
 
-	// set internals
+	g.tcpInterval = c.PushPullInterval
+	g.addrs = curAddrs
 	g.queue = queue
 	g.member = m
 	g.interval = c.GossipInterval
 
-	log.Logf("Registry Listening on %s", m.LocalNode().Address())
-	return nil
+	g.Unlock()
+
+	log.Logf("[gossip] Registry Listening on %s", m.LocalNode().Address())
+
+	// try connect
+	return g.connect(curAddrs)
 }
 
+func (*broadcast) UniqueBroadcast() {}
+
 func (b *broadcast) Invalidates(other memberlist.Broadcast) bool {
-	up := new(pb.Update)
-	if err := proto.Unmarshal(other.Message(), up); err != nil {
-		return false
-	}
-
-	// ids do not match
-	if b.update.Id == up.Id {
-		return false
-	}
-
-	// timestamps do not match
-	if b.update.Timestamp != up.Timestamp {
-		return false
-	}
-
-	// type does not match
-	if b.update.Type != up.Type {
-		return false
-	}
-
-	// invalidates
-	return true
+	return false
 }
 
 func (b *broadcast) Message() []byte {
@@ -217,6 +291,9 @@ func (b *broadcast) Message() []byte {
 	if err != nil {
 		return nil
 	}
+	if l := len(up); l > MaxPacketSize {
+		log.Logf("[gossip] broadcast message size %d bigger then MaxPacketSize %d", l, MaxPacketSize)
+	}
 	return up
 }
 
@@ -242,7 +319,7 @@ func (d *delegate) NotifyMsg(b []byte) {
 		}
 
 		// only process service action
-		if up.Type != "service" {
+		if up.Type != updateTypeService {
 			return
 		}
 
@@ -280,7 +357,7 @@ func (d *delegate) LocalState(join bool) []byte {
 
 	d.updates <- &update{
 		Update: &pb.Update{
-			Action: syncAction,
+			Action: actionTypeSync,
 		},
 		sync: syncCh,
 	}
@@ -305,11 +382,10 @@ func (d *delegate) MergeRemoteState(buf []byte, join bool) {
 	if err := json.Unmarshal(buf, &services); err != nil {
 		return
 	}
-
 	for _, service := range services {
 		for _, srv := range service {
 			d.updates <- &update{
-				Update:  &pb.Update{Action: addAction},
+				Update:  &pb.Update{Action: actionTypeCreate},
 				Service: srv,
 				sync:    nil,
 			}
@@ -317,8 +393,57 @@ func (d *delegate) MergeRemoteState(buf []byte, join bool) {
 	}
 }
 
+func (g *gossipRegistry) connect(addrs []string) error {
+	if len(addrs) == 0 {
+		return nil
+	}
+
+	timeout := make(<-chan time.Time)
+
+	if g.connectTimeout > 0 {
+		timeout = time.After(g.connectTimeout)
+	}
+
+	ticker := time.NewTicker(1 * time.Second)
+	defer ticker.Stop()
+
+	fn := func() (int, error) {
+		return g.member.Join(addrs)
+	}
+
+	// don't wait for first try
+	if _, err := fn(); err == nil {
+		return nil
+	}
+
+	// wait loop
+	for {
+		select {
+		// context closed
+		case <-g.options.Context.Done():
+			return nil
+		// call close, don't wait anymore
+		case <-g.done:
+			return nil
+		//  in case of timeout fail with a timeout error
+		case <-timeout:
+			return fmt.Errorf("[gossip] connect timeout %v", g.addrs)
+		// got a tick, try to connect
+		case <-ticker.C:
+			if _, err := fn(); err == nil {
+				log.Logf("[gossip] connect success for %v", g.addrs)
+				return nil
+			} else {
+				log.Logf("[gossip] connect failed for %v", g.addrs)
+			}
+		}
+	}
+
+	return nil
+}
+
 func (g *gossipRegistry) publish(action string, services []*registry.Service) {
-	g.s.RLock()
+	g.RLock()
 	for _, sub := range g.watchers {
 		go func(sub chan *registry.Result) {
 			for _, service := range services {
@@ -326,7 +451,7 @@ func (g *gossipRegistry) publish(action string, services []*registry.Service) {
 			}
 		}(sub)
 	}
-	g.s.RUnlock()
+	g.RUnlock()
 }
 
 func (g *gossipRegistry) subscribe() (chan *registry.Result, chan bool) {
@@ -335,56 +460,158 @@ func (g *gossipRegistry) subscribe() (chan *registry.Result, chan bool) {
 
 	id := uuid.New().String()
 
-	g.s.Lock()
+	g.Lock()
 	g.watchers[id] = next
-	g.s.Unlock()
+	g.Unlock()
 
 	go func() {
 		<-exit
-		g.s.Lock()
+		g.Lock()
 		delete(g.watchers, id)
 		close(next)
-		g.s.Unlock()
+		g.Unlock()
 	}()
 
 	return next, exit
 }
 
-func (g *gossipRegistry) run() {
-	var mtx sync.Mutex
-	updates := map[uint64]*update{}
+func (g *gossipRegistry) Stop() error {
+	select {
+	case <-g.done:
+		return nil
+	default:
+		close(g.done)
+		g.Lock()
+		if g.member != nil {
+			g.member.Leave(g.interval * 2)
+			g.member.Shutdown()
+			g.member = nil
+		}
+		g.Unlock()
+	}
+	return nil
+}
 
-	// expiry loop
-	go func() {
-		t := time.NewTicker(ExpiryTick)
-		defer t.Stop()
+// connectLoop attempts to reconnect to the memberlist
+func (g *gossipRegistry) connectLoop() {
+	// try every second
+	ticker := time.NewTicker(1 * time.Second)
+	defer ticker.Stop()
 
-		for _ = range t.C {
+	for {
+		select {
+		case <-g.done:
+			return
+		case <-g.options.Context.Done():
+			g.Stop()
+			return
+		case <-ticker.C:
+			var addrs []string
+
+			g.RLock()
+
+			// only process if we have a memberlist
+			if g.member == nil {
+				g.RUnlock()
+				continue
+			}
+
+			// self
+			local := g.member.LocalNode().Address()
+
+			// operate on each member
+			for node, action := range g.members {
+				switch action {
+				// process leave event
+				case nodeActionLeave:
+					// don't process self
+					if node == local {
+						continue
+					}
+					addrs = append(addrs, node)
+				}
+			}
+
+			g.RUnlock()
+
+			// connect to all the members
+			// TODO: only connect to new members
+			if len(addrs) > 0 {
+				g.connect(addrs)
+			}
+		}
+	}
+}
+
+func (g *gossipRegistry) expiryLoop(updates *updates) {
+	ticker := time.NewTicker(ExpiryTick)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-g.done:
+			return
+		case <-ticker.C:
 			now := uint64(time.Now().UnixNano())
 
-			mtx.Lock()
+			updates.Lock()
 
 			// process all the updates
-			for k, v := range updates {
+			for k, v := range updates.services {
 				// check if expiry time has passed
-				if d := (v.Update.Timestamp + v.Update.Expires); d < now {
+				if d := (v.Update.Expires); d < now {
 					// delete from records
-					delete(updates, k)
+					delete(updates.services, k)
 					// set to delete
-					v.Update.Action = delAction
+					v.Update.Action = actionTypeDelete
 					// fire a new update
 					g.updates <- v
 				}
 			}
 
-			mtx.Unlock()
+			updates.Unlock()
 		}
-	}()
+	}
+}
+
+// process member events
+func (g *gossipRegistry) eventLoop() {
+	for {
+		select {
+		// return when done
+		case <-g.done:
+			return
+		case ev := <-g.events:
+			// TODO: nonblocking update
+			g.Lock()
+			if _, ok := g.members[ev.node]; ok {
+				g.members[ev.node] = ev.action
+			}
+			g.Unlock()
+		}
+	}
+}
+
+func (g *gossipRegistry) run() {
+	updates := &updates{
+		services: make(map[uint64]*update),
+	}
+
+	// expiry loop
+	go g.expiryLoop(updates)
+
+	// event loop
+	go g.eventLoop()
+
+	// connect loop
+	if g.connectRetry {
+		go g.connectLoop()
+	}
 
 	// process the updates
 	for u := range g.updates {
 		switch u.Update.Action {
-		case addAction:
+		case actionTypeCreate:
 			g.Lock()
 			if service, ok := g.services[u.Service.Name]; !ok {
 				g.services[u.Service.Name] = []*registry.Service{u.Service}
@@ -395,18 +622,18 @@ func (g *gossipRegistry) run() {
 			g.Unlock()
 
 			// publish update to watchers
-			go g.publish(addAction, []*registry.Service{u.Service})
+			go g.publish(actionTypeString(actionTypeCreate), []*registry.Service{u.Service})
 
 			// we need to expire the node at some point in the future
 			if u.Update.Expires > 0 {
 				// create a hash of this service
 				if hash, err := hashstructure.Hash(u.Service, nil); err == nil {
-					mtx.Lock()
-					updates[hash] = u
-					mtx.Unlock()
+					updates.Lock()
+					updates.services[hash] = u
+					updates.Unlock()
 				}
 			}
-		case delAction:
+		case actionTypeDelete:
 			g.Lock()
 			if service, ok := g.services[u.Service.Name]; ok {
 				if services := delServices(service, []*registry.Service{u.Service}); len(services) == 0 {
@@ -418,15 +645,15 @@ func (g *gossipRegistry) run() {
 			g.Unlock()
 
 			// publish update to watchers
-			go g.publish(delAction, []*registry.Service{u.Service})
+			go g.publish(actionTypeString(actionTypeDelete), []*registry.Service{u.Service})
 
 			// delete from expiry checks
 			if hash, err := hashstructure.Hash(u.Service, nil); err == nil {
-				mtx.Lock()
-				delete(updates, hash)
-				mtx.Unlock()
+				updates.Lock()
+				delete(updates.services, hash)
+				updates.Unlock()
 			}
-		case syncAction:
+		case actionTypeSync:
 			// no sync channel provided
 			if u.sync == nil {
 				continue
@@ -441,7 +668,7 @@ func (g *gossipRegistry) run() {
 				}
 
 				// publish to watchers
-				go g.publish(addAction, service)
+				go g.publish(actionTypeString(actionTypeCreate), service)
 			}
 
 			g.RUnlock()
@@ -479,12 +706,14 @@ func (g *gossipRegistry) Register(s *registry.Service, opts ...registry.Register
 		o(&options)
 	}
 
+	if options.TTL == 0 && g.tcpInterval == 0 {
+		return fmt.Errorf("Require register TTL or interval for memberlist.Config")
+	}
+
 	up := &pb.Update{
-		Id:        uuid.New().String(),
-		Timestamp: uint64(time.Now().UnixNano()),
-		Expires:   uint64(options.TTL.Nanoseconds()),
-		Action:    "update",
-		Type:      "service",
+		Expires: uint64(time.Now().Add(options.TTL).UnixNano()),
+		Action:  actionTypeCreate,
+		Type:    updateTypeService,
 		Metadata: map[string]string{
 			"Content-Type": "application/json",
 		},
@@ -519,10 +748,8 @@ func (g *gossipRegistry) Deregister(s *registry.Service) error {
 	g.Unlock()
 
 	up := &pb.Update{
-		Id:        uuid.New().String(),
-		Timestamp: uint64(time.Now().UnixNano()),
-		Action:    "delete",
-		Type:      "service",
+		Action: actionTypeDelete,
+		Type:   updateTypeService,
 		Metadata: map[string]string{
 			"Content-Type": "application/json",
 		},
@@ -570,25 +797,28 @@ func (g *gossipRegistry) String() string {
 }
 
 func NewRegistry(opts ...registry.Option) registry.Registry {
-	gossip := &gossipRegistry{
+	g := &gossipRegistry{
 		options: registry.Options{
 			Context: context.Background(),
 		},
+		done:     make(chan bool),
+		events:   make(chan *event, 100),
 		updates:  make(chan *update, 100),
 		services: make(map[string][]*registry.Service),
 		watchers: make(map[string]chan *registry.Result),
+		members:  make(map[string]int32),
 	}
 
 	// run the updater
-	go gossip.run()
+	go g.run()
 
 	// configure the gossiper
-	if err := configure(gossip, opts...); err != nil {
-		log.Fatalf("Error configuring registry: %v", err)
+	if err := configure(g, opts...); err != nil {
+		log.Fatalf("[gossip] Error configuring registry: %v", err)
 	}
 
 	// wait for setup
-	<-time.After(gossip.interval * 2)
+	<-time.After(g.interval * 2)
 
-	return gossip
+	return g
 }

registry/gossip/gossip_test.go

@@ -0,0 +1,214 @@
+package gossip
+
+import (
+	"os"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/hashicorp/memberlist"
+	"github.com/micro/go-micro/registry"
+)
+
+func newMemberlistConfig() *memberlist.Config {
+	mc := memberlist.DefaultLANConfig()
+	mc.DisableTcpPings = false
+	mc.GossipVerifyIncoming = false
+	mc.GossipVerifyOutgoing = false
+	mc.EnableCompression = false
+	mc.PushPullInterval = 3 * time.Second
+	mc.LogOutput = os.Stderr
+	mc.ProtocolVersion = 4
+	mc.Name = uuid.New().String()
+	return mc
+}
+
+func newRegistry(opts ...registry.Option) registry.Registry {
+	options := []registry.Option{
+		ConnectRetry(true),
+		ConnectTimeout(60 * time.Second),
+	}
+
+	options = append(options, opts...)
+	r := NewRegistry(options...)
+	return r
+}
+
+func TestGossipRegistryBroadcast(t *testing.T) {
+	mc1 := newMemberlistConfig()
+	r1 := newRegistry(Config(mc1), Address("127.0.0.1:54321"))
+
+	mc2 := newMemberlistConfig()
+	r2 := newRegistry(Config(mc2), Address("127.0.0.2:54321"), registry.Addrs("127.0.0.1:54321"))
+
+	defer r1.(*gossipRegistry).Stop()
+	defer r2.(*gossipRegistry).Stop()
+
+	svc1 := &registry.Service{Name: "service.1", Version: "0.0.0.1"}
+	svc2 := &registry.Service{Name: "service.2", Version: "0.0.0.2"}
+
+	if err := r1.Register(svc1, registry.RegisterTTL(10*time.Second)); err != nil {
+		t.Fatal(err)
+	}
+	if err := r2.Register(svc2, registry.RegisterTTL(10*time.Second)); err != nil {
+		t.Fatal(err)
+	}
+
+	var found bool
+	svcs, err := r1.ListServices()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	for _, svc := range svcs {
+		if svc.Name == "service.2" {
+			found = true
+		}
+	}
+	if !found {
+		t.Fatalf("[gossip registry] service.2 not found in r1, broadcast not work")
+	}
+
+	found = false
+
+	svcs, err = r2.ListServices()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	for _, svc := range svcs {
+		if svc.Name == "service.1" {
+			found = true
+		}
+	}
+
+	if !found {
+		t.Fatalf("[gossip registry] broadcast failed: service.1 not found in r2")
+	}
+
+	if err := r1.Deregister(svc1); err != nil {
+		t.Fatal(err)
+	}
+	if err := r2.Deregister(svc2); err != nil {
+		t.Fatal(err)
+	}
+
+}
+func TestGossipRegistryRetry(t *testing.T) {
+	mc1 := newMemberlistConfig()
+	r1 := newRegistry(Config(mc1), Address("127.0.0.1:54321"))
+
+	mc2 := newMemberlistConfig()
+	r2 := newRegistry(Config(mc2), Address("127.0.0.2:54321"), registry.Addrs("127.0.0.1:54321"))
+
+	defer r1.(*gossipRegistry).Stop()
+	defer r2.(*gossipRegistry).Stop()
+
+	svc1 := &registry.Service{Name: "service.1", Version: "0.0.0.1"}
+	svc2 := &registry.Service{Name: "service.2", Version: "0.0.0.2"}
+
+	var mu sync.Mutex
+	ch := make(chan struct{})
+	ticker := time.NewTicker(1 * time.Second)
+	defer ticker.Stop()
+
+	go func() {
+		for {
+			select {
+			case <-ticker.C:
+				mu.Lock()
+				if r1 != nil {
+					r1.Register(svc1, registry.RegisterTTL(2*time.Second))
+				}
+				if r2 != nil {
+					r2.Register(svc2, registry.RegisterTTL(2*time.Second))
+				}
+				if ch != nil {
+					close(ch)
+					ch = nil
+				}
+				mu.Unlock()
+			}
+		}
+	}()
+
+	<-ch
+	var found bool
+
+	svcs, err := r2.ListServices()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	for _, svc := range svcs {
+		if svc.Name == "service.1" {
+			found = true
+		}
+	}
+
+	if !found {
+		t.Fatalf("[gossip registry] broadcast failed: service.1 not found in r2")
+	}
+
+	if err = r1.(*gossipRegistry).Stop(); err != nil {
+		t.Fatalf("[gossip registry] failed to stop registry: %v", err)
+	}
+
+	mu.Lock()
+	r1 = nil
+	mu.Unlock()
+
+	<-time.After(3 * time.Second)
+
+	found = false
+	svcs, err = r2.ListServices()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	for _, svc := range svcs {
+		if svc.Name == "service.1" {
+			found = true
+		}
+	}
+
+	if found {
+		t.Fatalf("[gossip registry] service.1 found in r2")
+	}
+
+	if tr := os.Getenv("TRAVIS"); len(tr) > 0 {
+		t.Logf("[gossip registry] skip test on travis")
+		t.Skip()
+		return
+	}
+
+	r1 = newRegistry(Config(mc1), Address("127.0.0.1:54321"))
+	<-time.After(2 * time.Second)
+
+	found = false
+	svcs, err = r2.ListServices()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	for _, svc := range svcs {
+		if svc.Name == "service.1" {
+			found = true
+		}
+	}
+
+	if !found {
+		t.Fatalf("[gossip registry] connect retry failed: service.1 not found in r2")
+	}
+
+	if err := r1.Deregister(svc1); err != nil {
+		t.Fatal(err)
+	}
+	if err := r2.Deregister(svc2); err != nil {
+		t.Fatal(err)
+	}
+
+	r1.(*gossipRegistry).Stop()
+	r2.(*gossipRegistry).Stop()
+}

registry/gossip/options.go

@@ -2,44 +2,57 @@ package gossip
 
 import (
 	"context"
+	"time"
 
 	"github.com/hashicorp/memberlist"
 	"github.com/micro/go-micro/registry"
 )
 
-type contextSecretKey struct{}
+type secretKey struct{}
+type addressKey struct{}
+type configKey struct{}
+type advertiseKey struct{}
+type connectTimeoutKey struct{}
+type connectRetryKey struct{}
+
+// helper for setting registry options
+func setRegistryOption(k, v interface{}) registry.Option {
+	return func(o *registry.Options) {
+		if o.Context == nil {
+			o.Context = context.Background()
+		}
+		o.Context = context.WithValue(o.Context, k, v)
+	}
+}
 
 // Secret specifies an encryption key. The value should be either
 // 16, 24, or 32 bytes to select AES-128, AES-192, or AES-256.
 func Secret(k []byte) registry.Option {
-	return func(o *registry.Options) {
-		o.Context = context.WithValue(o.Context, contextSecretKey{}, k)
-	}
+	return setRegistryOption(secretKey{}, k)
 }
 
-type contextAddress struct{}
-
 // Address to bind to - host:port
 func Address(a string) registry.Option {
-	return func(o *registry.Options) {
-		o.Context = context.WithValue(o.Context, contextAddress{}, a)
-	}
+	return setRegistryOption(addressKey{}, a)
 }
 
-type contextConfig struct{}
-
-// Config allow to inject a *memberlist.Config struct for configuring gossip
+// Config sets *memberlist.Config for configuring gossip
 func Config(c *memberlist.Config) registry.Option {
-	return func(o *registry.Options) {
-		o.Context = context.WithValue(o.Context, contextConfig{}, c)
-	}
+	return setRegistryOption(configKey{}, c)
 }
 
-type contextAdvertise struct{}
-
-// The address to advertise for other gossip members - host:port
+// The address to advertise for other gossip members to connect to - host:port
 func Advertise(a string) registry.Option {
-	return func(o *registry.Options) {
-		o.Context = context.WithValue(o.Context, contextAdvertise{}, a)
-	}
+	return setRegistryOption(advertiseKey{}, a)
+}
+
+// ConnectTimeout sets the registry connect timeout. Use -1 to specify infinite timeout
+func ConnectTimeout(td time.Duration) registry.Option {
+	return setRegistryOption(connectTimeoutKey{}, td)
+}
+
+// ConnectRetry enables reconnect to registry then connection closed,
+// use with ConnectTimeout to specify how long retry
+func ConnectRetry(v bool) registry.Option {
+	return setRegistryOption(connectRetryKey{}, v)
 }

registry/gossip/proto/gossip.pb.go

@@ -3,9 +3,12 @@
 
 package gossip
 
-import proto "github.com/golang/protobuf/proto"
-import fmt "fmt"
-import math "math"
+import (
+	fmt "fmt"
+	math "math"
+
+	proto "github.com/golang/protobuf/proto"
+)
 
 // Reference imports to suppress errors if they are not otherwise used.
 var _ = proto.Marshal
@@ -20,16 +23,12 @@ const _ = proto.ProtoPackageIsVersion2 // please upgrade the proto package
 
 // Update is the message broadcast
 type Update struct {
-	// unique id of update
-	Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"`
-	// unix nano timestamp of update
-	Timestamp uint64 `protobuf:"varint,2,opt,name=timestamp,proto3" json:"timestamp,omitempty"`
 	// time to live for entry
-	Expires uint64 `protobuf:"varint,3,opt,name=expires,proto3" json:"expires,omitempty"`
-	// type of update; service
-	Type string `protobuf:"bytes,4,opt,name=type,proto3" json:"type,omitempty"`
-	// what action is taken; add, del, put
-	Action string `protobuf:"bytes,5,opt,name=action,proto3" json:"action,omitempty"`
+	Expires uint64 `protobuf:"varint,1,opt,name=expires,proto3" json:"expires,omitempty"`
+	// type of update
+	Type int32 `protobuf:"varint,2,opt,name=type,proto3" json:"type,omitempty"`
+	// what action is taken
+	Action int32 `protobuf:"varint,3,opt,name=action,proto3" json:"action,omitempty"`
 	// any other associated metadata about the data
 	Metadata map[string]string `protobuf:"bytes,6,rep,name=metadata,proto3" json:"metadata,omitempty" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"`
 	// the payload data;
@@ -43,16 +42,17 @@ func (m *Update) Reset()         { *m = Update{} }
 func (m *Update) String() string { return proto.CompactTextString(m) }
 func (*Update) ProtoMessage()    {}
 func (*Update) Descriptor() ([]byte, []int) {
-	return fileDescriptor_gossip_fd1eb378131a5d12, []int{0}
+	return fileDescriptor_18cba623e76e57f3, []int{0}
 }
+
 func (m *Update) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_Update.Unmarshal(m, b)
 }
 func (m *Update) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
 	return xxx_messageInfo_Update.Marshal(b, m, deterministic)
 }
-func (dst *Update) XXX_Merge(src proto.Message) {
-	xxx_messageInfo_Update.Merge(dst, src)
+func (m *Update) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_Update.Merge(m, src)
 }
 func (m *Update) XXX_Size() int {
 	return xxx_messageInfo_Update.Size(m)
@@ -63,20 +63,6 @@ func (m *Update) XXX_DiscardUnknown() {
 
 var xxx_messageInfo_Update proto.InternalMessageInfo
 
-func (m *Update) GetId() string {
-	if m != nil {
-		return m.Id
-	}
-	return ""
-}
-
-func (m *Update) GetTimestamp() uint64 {
-	if m != nil {
-		return m.Timestamp
-	}
-	return 0
-}
-
 func (m *Update) GetExpires() uint64 {
 	if m != nil {
 		return m.Expires
@@ -84,18 +70,18 @@ func (m *Update) GetExpires() uint64 {
 	return 0
 }
 
-func (m *Update) GetType() string {
+func (m *Update) GetType() int32 {
 	if m != nil {
 		return m.Type
 	}
-	return ""
+	return 0
 }
 
-func (m *Update) GetAction() string {
+func (m *Update) GetAction() int32 {
 	if m != nil {
 		return m.Action
 	}
-	return ""
+	return 0
 }
 
 func (m *Update) GetMetadata() map[string]string {
@@ -118,25 +104,24 @@ func init() {
 }
 
 func init() {
-	proto.RegisterFile("github.com/micro/go-micro/registry/gossip/proto/gossip.proto", fileDescriptor_gossip_fd1eb378131a5d12)
+	proto.RegisterFile("github.com/micro/go-micro/registry/gossip/proto/gossip.proto", fileDescriptor_18cba623e76e57f3)
 }
 
-var fileDescriptor_gossip_fd1eb378131a5d12 = []byte{
-	// 251 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x54, 0x90, 0xcf, 0x4a, 0xc4, 0x30,
-	0x10, 0x87, 0x69, 0xb6, 0x9b, 0xb5, 0xe3, 0x1f, 0x64, 0x10, 0x09, 0xb2, 0x87, 0xe2, 0xa9, 0x17,
-	0x5b, 0xd0, 0xcb, 0xa2, 0x5e, 0x3d, 0x7a, 0x09, 0xf8, 0x00, 0xd9, 0x36, 0xd4, 0xa0, 0xd9, 0x84,
-	0x64, 0x56, 0xec, 0x13, 0xf8, 0xda, 0xb2, 0x69, 0x54, 0xbc, 0x7d, 0xdf, 0xcc, 0x24, 0x99, 0x5f,
-	0xe0, 0x71, 0x34, 0xf4, 0xba, 0xdf, 0xb6, 0xbd, 0xb3, 0x9d, 0x35, 0x7d, 0x70, 0xdd, 0xe8, 0x6e,
-	0x66, 0x08, 0x7a, 0x34, 0x91, 0xc2, 0xd4, 0x8d, 0x2e, 0x46, 0xe3, 0x3b, 0x1f, 0x1c, 0xb9, 0x2c,
-	0x6d, 0x12, 0xe4, 0xb3, 0x5d, 0x7f, 0x31, 0xe0, 0x2f, 0x7e, 0x50, 0xa4, 0xf1, 0x0c, 0x98, 0x19,
-	0x44, 0x51, 0x17, 0x4d, 0x25, 0x99, 0x19, 0x70, 0x0d, 0x15, 0x19, 0xab, 0x23, 0x29, 0xeb, 0x05,
-	0xab, 0x8b, 0xa6, 0x94, 0x7f, 0x05, 0x14, 0xb0, 0xd2, 0x9f, 0xde, 0x04, 0x1d, 0xc5, 0x22, 0xf5,
-	0x7e, 0x14, 0x11, 0x4a, 0x9a, 0xbc, 0x16, 0x65, 0xba, 0x29, 0x31, 0x5e, 0x02, 0x57, 0x3d, 0x19,
-	0xb7, 0x13, 0xcb, 0x54, 0xcd, 0x86, 0x1b, 0x38, 0xb2, 0x9a, 0xd4, 0xa0, 0x48, 0x09, 0x5e, 0x2f,
-	0x9a, 0xe3, 0xdb, 0x75, 0x9b, 0xf7, 0x9c, 0xb7, 0x6a, 0x9f, 0x73, 0xfb, 0x69, 0x47, 0x61, 0x92,
-	0xbf, 0xd3, 0x87, 0x57, 0xd2, 0xa9, 0x55, 0x5d, 0x34, 0x27, 0x32, 0xf1, 0xd5, 0x03, 0x9c, 0xfe,
-	0x1b, 0xc7, 0x73, 0x58, 0xbc, 0xe9, 0x29, 0x67, 0x3a, 0x20, 0x5e, 0xc0, 0xf2, 0x43, 0xbd, 0xef,
-	0x75, 0x0a, 0x54, 0xc9, 0x59, 0xee, 0xd9, 0xa6, 0xd8, 0xf2, 0xf4, 0x31, 0x77, 0xdf, 0x01, 0x00,
-	0x00, 0xff, 0xff, 0x06, 0x6e, 0x00, 0x3c, 0x58, 0x01, 0x00, 0x00,
+var fileDescriptor_18cba623e76e57f3 = []byte{
+	// 227 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x54, 0x8f, 0xc1, 0x4a, 0x03, 0x31,
+	0x14, 0x45, 0x49, 0xa7, 0x4d, 0xed, 0x53, 0x41, 0x1e, 0x22, 0x41, 0x5c, 0x0c, 0xae, 0x66, 0xe3,
+	0x0c, 0xe8, 0xa6, 0xa8, 0x5b, 0x97, 0x6e, 0x02, 0x7e, 0x40, 0x3a, 0x0d, 0x31, 0xe8, 0x34, 0x21,
+	0x79, 0x15, 0xf3, 0xa9, 0xfe, 0x8d, 0x34, 0x89, 0x42, 0x77, 0xe7, 0x24, 0x37, 0xdc, 0x1b, 0x78,
+	0x36, 0x96, 0xde, 0xf7, 0x9b, 0x7e, 0x74, 0xd3, 0x30, 0xd9, 0x31, 0xb8, 0xc1, 0xb8, 0xbb, 0x02,
+	0x41, 0x1b, 0x1b, 0x29, 0xa4, 0xc1, 0xb8, 0x18, 0xad, 0x1f, 0x7c, 0x70, 0xe4, 0xaa, 0xf4, 0x59,
+	0x90, 0x17, 0xbb, 0xfd, 0x61, 0xc0, 0xdf, 0xfc, 0x56, 0x91, 0x46, 0x01, 0x4b, 0xfd, 0xed, 0x6d,
+	0xd0, 0x51, 0xb0, 0x96, 0x75, 0x73, 0xf9, 0xa7, 0x88, 0x30, 0xa7, 0xe4, 0xb5, 0x98, 0xb5, 0xac,
+	0x5b, 0xc8, 0xcc, 0x78, 0x05, 0x5c, 0x8d, 0x64, 0xdd, 0x4e, 0x34, 0xf9, 0xb4, 0x1a, 0xae, 0xe1,
+	0x64, 0xd2, 0xa4, 0xb6, 0x8a, 0x94, 0xe0, 0x6d, 0xd3, 0x9d, 0xde, 0xdf, 0xf4, 0xb5, 0xb9, 0xf4,
+	0xf4, 0xaf, 0xf5, 0xfa, 0x65, 0x47, 0x21, 0xc9, 0xff, 0xf4, 0xa1, 0x25, 0xbf, 0x5a, 0xb6, 0xac,
+	0x3b, 0x93, 0x99, 0xaf, 0x9f, 0xe0, 0xfc, 0x28, 0x8e, 0x17, 0xd0, 0x7c, 0xe8, 0x94, 0x07, 0xae,
+	0xe4, 0x01, 0xf1, 0x12, 0x16, 0x5f, 0xea, 0x73, 0x5f, 0xd6, 0xad, 0x64, 0x91, 0xc7, 0xd9, 0x9a,
+	0x6d, 0x78, 0xfe, 0xea, 0xc3, 0x6f, 0x00, 0x00, 0x00, 0xff, 0xff, 0xd6, 0x63, 0x7b, 0x1b, 0x2a,
+	0x01, 0x00, 0x00,
 }

registry/gossip/proto/gossip.proto

@@ -4,16 +4,12 @@ package gossip;
 
 // Update is the message broadcast
 message Update {
-	// unique id of update
-	string id = 1;
-	// unix nano timestamp of update
-	uint64 timestamp = 2;
 	// time to live for entry
-	uint64 expires = 3;
-	// type of update; service
-	string type = 4;
-	// what action is taken; add, del, put
-	string action = 5;
+	uint64 expires = 1;
+	// type of update
+	int32 type = 2;
+	// what action is taken
+	int32 action = 3;
 	// any other associated metadata about the data
 	map<string, string> metadata = 6;
 	// the payload data;

registry/gossip/util.go

@@ -70,21 +70,30 @@ func addNodes(old, neu []*registry.Node) []*registry.Node {
 }
 
 func addServices(old, neu []*registry.Service) []*registry.Service {
+	var srv []*registry.Service
+
 	for _, s := range neu {
 		var seen bool
-		for i, o := range old {
+		for _, o := range old {
 			if o.Version == s.Version {
-				s.Nodes = addNodes(o.Nodes, s.Nodes)
+				sp := new(registry.Service)
+				// make copy
+				*sp = *o
+				// set nodes
+				sp.Nodes = addNodes(o.Nodes, s.Nodes)
+
+				// mark as seen
 				seen = true
-				old[i] = s
+				srv = append(srv, sp)
 				break
 			}
 		}
 		if !seen {
-			old = append(old, s)
+			srv = append(srv, cp([]*registry.Service{s})...)
 		}
 	}
-	return old
+
+	return srv
 }
 
 func delNodes(old, del []*registry.Node) []*registry.Node {

registry/mdns_registry.go

@@ -2,6 +2,7 @@
 package registry
 
 import (
+	"context"
 	"net"
 	"strings"
 	"sync"
@@ -194,20 +195,20 @@ func (m *mdnsRegistry) Deregister(service *Service) error {
 }
 
 func (m *mdnsRegistry) GetService(service string) ([]*Service, error) {
-	p := mdns.DefaultParams(service)
-	p.Timeout = m.opts.Timeout
-	entryCh := make(chan *mdns.ServiceEntry, 10)
-	p.Entries = entryCh
-
-	exit := make(chan bool)
-	defer close(exit)
-
 	serviceMap := make(map[string]*Service)
+	entries := make(chan *mdns.ServiceEntry, 10)
+	done := make(chan bool)
+
+	p := mdns.DefaultParams(service)
+	// set context with timeout
+	p.Context, _ = context.WithTimeout(context.Background(), m.opts.Timeout)
+	// set entries channel
+	p.Entries = entries
 
 	go func() {
 		for {
 			select {
-			case e := <-entryCh:
+			case e := <-entries:
 				// list record so skip
 				if p.Service == "_services" {
 					continue
@@ -243,16 +244,21 @@ func (m *mdnsRegistry) GetService(service string) ([]*Service, error) {
 				})
 
 				serviceMap[txt.Version] = s
-			case <-exit:
+			case <-p.Context.Done():
+				close(done)
 				return
 			}
 		}
 	}()
 
+	// execute the query
 	if err := mdns.Query(p); err != nil {
 		return nil, err
 	}
 
+	// wait for completion
+	<-done
+
 	// create list and return
 	var services []*Service
 
@@ -264,21 +270,22 @@ func (m *mdnsRegistry) GetService(service string) ([]*Service, error) {
 }
 
 func (m *mdnsRegistry) ListServices() ([]*Service, error) {
-	p := mdns.DefaultParams("_services")
-	p.Timeout = m.opts.Timeout
-	entryCh := make(chan *mdns.ServiceEntry, 10)
-	p.Entries = entryCh
-
-	exit := make(chan bool)
-	defer close(exit)
-
 	serviceMap := make(map[string]bool)
+	entries := make(chan *mdns.ServiceEntry, 10)
+	done := make(chan bool)
+
+	p := mdns.DefaultParams("_services")
+	// set context with timeout
+	p.Context, _ = context.WithTimeout(context.Background(), m.opts.Timeout)
+	// set entries channel
+	p.Entries = entries
+
 	var services []*Service
 
 	go func() {
 		for {
 			select {
-			case e := <-entryCh:
+			case e := <-entries:
 				if e.TTL == 0 {
 					continue
 				}
@@ -288,16 +295,21 @@ func (m *mdnsRegistry) ListServices() ([]*Service, error) {
 					serviceMap[name] = true
 					services = append(services, &Service{Name: name})
 				}
-			case <-exit:
+			case <-p.Context.Done():
+				close(done)
 				return
 			}
 		}
 	}()
 
+	// execute query
 	if err := mdns.Query(p); err != nil {
 		return nil, err
 	}
 
+	// wait till done
+	<-done
+
 	return services, nil
 }
 

registry/memory/data.go

@@ -0,0 +1,51 @@
+package memory
+
+import (
+	"github.com/micro/go-micro/registry"
+)
+
+var (
+	// mock data
+	Data = map[string][]*registry.Service{
+		"foo": []*registry.Service{
+			{
+				Name:    "foo",
+				Version: "1.0.0",
+				Nodes: []*registry.Node{
+					{
+						Id:      "foo-1.0.0-123",
+						Address: "localhost",
+						Port:    9999,
+					},
+					{
+						Id:      "foo-1.0.0-321",
+						Address: "localhost",
+						Port:    9999,
+					},
+				},
+			},
+			{
+				Name:    "foo",
+				Version: "1.0.1",
+				Nodes: []*registry.Node{
+					{
+						Id:      "foo-1.0.1-321",
+						Address: "localhost",
+						Port:    6666,
+					},
+				},
+			},
+			{
+				Name:    "foo",
+				Version: "1.0.3",
+				Nodes: []*registry.Node{
+					{
+						Id:      "foo-1.0.3-345",
+						Address: "localhost",
+						Port:    8888,
+					},
+				},
+			},
+		},
+	}
+)

registry/memory/memory.go

@@ -2,60 +2,24 @@
 package memory
 
 import (
+	"context"
 	"sync"
+	"time"
 
+	"github.com/google/uuid"
 	"github.com/micro/go-micro/registry"
 )
 
 type Registry struct {
+	options registry.Options
+
 	sync.RWMutex
 	Services map[string][]*registry.Service
+	Watchers map[string]*Watcher
 }
 
 var (
-	// mock data
-	Data = map[string][]*registry.Service{
-		"foo": []*registry.Service{
-			{
-				Name:    "foo",
-				Version: "1.0.0",
-				Nodes: []*registry.Node{
-					{
-						Id:      "foo-1.0.0-123",
-						Address: "localhost",
-						Port:    9999,
-					},
-					{
-						Id:      "foo-1.0.0-321",
-						Address: "localhost",
-						Port:    9999,
-					},
-				},
-			},
-			{
-				Name:    "foo",
-				Version: "1.0.1",
-				Nodes: []*registry.Node{
-					{
-						Id:      "foo-1.0.1-321",
-						Address: "localhost",
-						Port:    6666,
-					},
-				},
-			},
-			{
-				Name:    "foo",
-				Version: "1.0.3",
-				Nodes: []*registry.Node{
-					{
-						Id:      "foo-1.0.3-345",
-						Address: "localhost",
-						Port:    8888,
-					},
-				},
-			},
-		},
-	}
+	timeout = time.Millisecond * 10
 )
 
 // Setup sets mock data
@@ -67,69 +31,130 @@ func (m *Registry) Setup() {
 	m.Services = Data
 }
 
-func (m *Registry) GetService(service string) ([]*registry.Service, error) {
-	m.Lock()
-	defer m.Unlock()
+func (m *Registry) watch(r *registry.Result) {
+	var watchers []*Watcher
 
+	m.RLock()
+	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(timeout):
+			}
+		}
+	}
+}
+
+func (m *Registry) Init(opts ...registry.Option) error {
+	for _, o := range opts {
+		o(&m.options)
+	}
+
+	// add services
+	m.Lock()
+	for k, v := range getServices(m.options.Context) {
+		s := m.Services[k]
+		m.Services[k] = addServices(s, v)
+	}
+	m.Unlock()
+	return nil
+}
+
+func (m *Registry) Options() registry.Options {
+	return m.options
+}
+
+func (m *Registry) GetService(service string) ([]*registry.Service, error) {
+	m.RLock()
 	s, ok := m.Services[service]
 	if !ok || len(s) == 0 {
+		m.RUnlock()
 		return nil, registry.ErrNotFound
 	}
+	m.RUnlock()
 	return s, nil
-
 }
 
 func (m *Registry) ListServices() ([]*registry.Service, error) {
-	m.Lock()
-	defer m.Unlock()
-
+	m.RLock()
 	var services []*registry.Service
 	for _, service := range m.Services {
 		services = append(services, service...)
 	}
+	m.RUnlock()
 	return services, nil
 }
 
 func (m *Registry) Register(s *registry.Service, opts ...registry.RegisterOption) error {
-	m.Lock()
-	defer m.Unlock()
+	go m.watch(&registry.Result{Action: "update", Service: s})
 
+	m.Lock()
 	services := addServices(m.Services[s.Name], []*registry.Service{s})
 	m.Services[s.Name] = services
+	m.Unlock()
 	return nil
 }
 
 func (m *Registry) Deregister(s *registry.Service) error {
-	m.Lock()
-	defer m.Unlock()
+	go m.watch(&registry.Result{Action: "delete", Service: s})
 
+	m.Lock()
 	services := delServices(m.Services[s.Name], []*registry.Service{s})
 	m.Services[s.Name] = services
+	m.Unlock()
 	return nil
 }
 
 func (m *Registry) Watch(opts ...registry.WatchOption) (registry.Watcher, error) {
-	var wopts registry.WatchOptions
+	var wo registry.WatchOptions
 	for _, o := range opts {
-		o(&wopts)
+		o(&wo)
 	}
-	return &memoryWatcher{exit: make(chan bool), opts: wopts}, nil
+
+	w := &Watcher{
+		exit: make(chan bool),
+		res:  make(chan *registry.Result),
+		id:   uuid.New().String(),
+		wo:   wo,
+	}
+
+	m.Lock()
+	m.Watchers[w.id] = w
+	m.Unlock()
+	return w, nil
 }
 
 func (m *Registry) String() string {
 	return "memory"
 }
 
-func (m *Registry) Init(opts ...registry.Option) error {
-	return nil
-}
-
-func (m *Registry) Options() registry.Options {
-	return registry.Options{}
-}
-
 func NewRegistry(opts ...registry.Option) registry.Registry {
+	options := registry.Options{
+		Context: context.Background(),
+	}
+
+	for _, o := range opts {
+		o(&options)
+	}
+
+	services := getServices(options.Context)
+	if services == nil {
+		services = make(map[string][]*registry.Service)
+	}
+
 	return &Registry{
-		Services: make(map[string][]*registry.Service),
+		options:  options,
+		Services: services,
+		Watchers: make(map[string]*Watcher),
 	}
 }

registry/memory/memory_test.go

@@ -80,9 +80,8 @@ var (
 	}
 )
 
-func TestMockRegistry(t *testing.T) {
+func TestMemoryRegistry(t *testing.T) {
 	m := NewRegistry()
-	m.(*Registry).Setup()
 
 	fn := func(k string, v []*registry.Service) {
 		services, err := m.GetService(k)
@@ -108,11 +107,6 @@ func TestMockRegistry(t *testing.T) {
 		}
 	}
 
-	// test existing memory data
-	for k, v := range Data {
-		fn(k, v)
-	}
-
 	// register data
 	for _, v := range testData {
 		for _, service := range v {
@@ -124,7 +118,6 @@ func TestMockRegistry(t *testing.T) {
 
 	// using test data
 	for k, v := range testData {
-
 		fn(k, v)
 	}
 

registry/memory/options.go

@@ -0,0 +1,27 @@
+package memory
+
+import (
+	"context"
+
+	"github.com/micro/go-micro/registry"
+)
+
+type servicesKey struct{}
+
+func getServices(ctx context.Context) map[string][]*registry.Service {
+	s, ok := ctx.Value(servicesKey{}).(map[string][]*registry.Service)
+	if !ok {
+		return nil
+	}
+	return s
+}
+
+// Services is an option that preloads service data
+func Services(s map[string][]*registry.Service) registry.Option {
+	return func(o *registry.Options) {
+		if o.Context == nil {
+			o.Context = context.Background()
+		}
+		o.Context = context.WithValue(o.Context, servicesKey{}, s)
+	}
+}

registry/memory/watcher.go

@@ -0,0 +1,37 @@
+package memory
+
+import (
+	"errors"
+
+	"github.com/micro/go-micro/registry"
+)
+
+type Watcher struct {
+	id   string
+	wo   registry.WatchOptions
+	res  chan *registry.Result
+	exit chan bool
+}
+
+func (m *Watcher) Next() (*registry.Result, error) {
+	for {
+		select {
+		case r := <-m.res:
+			if len(m.wo.Service) > 0 && m.wo.Service != r.Service.Name {
+				continue
+			}
+			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)
+	}
+}

registry/memory/watcher_test.go

@@ -0,0 +1,30 @@
+package memory
+
+import (
+	"testing"
+
+	"github.com/micro/go-micro/registry"
+)
+
+func TestWatcher(t *testing.T) {
+	w := &Watcher{
+		id:   "test",
+		res:  make(chan *registry.Result),
+		exit: make(chan bool),
+	}
+
+	go func() {
+		w.res <- &registry.Result{}
+	}()
+
+	_, 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()")
+	}
+}

registry/options.go

@@ -11,7 +11,6 @@ type Options struct {
 	Timeout   time.Duration
 	Secure    bool
 	TLSConfig *tls.Config
-
 	// Other options for implementations of the interface
 	// can be stored in a context
 	Context context.Context

selector/default.go

@@ -1,350 +1,25 @@
 package selector
 
 import (
-	"sync"
 	"time"
 
-	"github.com/micro/go-log"
 	"github.com/micro/go-micro/registry"
+	"github.com/micro/go-rcache"
 )
 
 type registrySelector struct {
-	so  Options
-	ttl time.Duration
-
-	// registry cache
-	sync.RWMutex
-	cache map[string][]*registry.Service
-	ttls  map[string]time.Time
-
-	watched map[string]bool
-
-	// used to close or reload watcher
-	reload chan bool
-	exit   chan bool
+	so Options
+	rc rcache.Cache
 }
 
-var (
-	DefaultTTL = time.Minute
-)
-
-// isValid checks if the service is valid
-func (c *registrySelector) isValid(services []*registry.Service, ttl time.Time) bool {
-	// no services exist
-	if len(services) == 0 {
-		return false
-	}
-
-	// ttl is invalid
-	if ttl.IsZero() {
-		return false
-	}
-
-	// time since ttl is longer than timeout
-	if time.Since(ttl) > c.ttl {
-		return false
-	}
-
-	// ok
-	return true
-}
-
-func (c *registrySelector) quit() bool {
-	select {
-	case <-c.exit:
-		return true
-	default:
-		return false
-	}
-}
-
-// cp copies a service. Because we're caching handing back pointers would
-// create a race condition, so we do this instead
-// its fast enough
-func (c *registrySelector) cp(current []*registry.Service) []*registry.Service {
-	var services []*registry.Service
-
-	for _, service := range current {
-		// copy service
-		s := new(registry.Service)
-		*s = *service
-
-		// copy nodes
-		var nodes []*registry.Node
-		for _, node := range service.Nodes {
-			n := new(registry.Node)
-			*n = *node
-			nodes = append(nodes, n)
-		}
-		s.Nodes = nodes
-
-		// copy endpoints
-		var eps []*registry.Endpoint
-		for _, ep := range service.Endpoints {
-			e := new(registry.Endpoint)
-			*e = *ep
-			eps = append(eps, e)
-		}
-		s.Endpoints = eps
-
-		// append service
-		services = append(services, s)
-	}
-
-	return services
-}
-
-func (c *registrySelector) del(service string) {
-	delete(c.cache, service)
-	delete(c.ttls, service)
-}
-
-func (c *registrySelector) get(service string) ([]*registry.Service, error) {
-	// read lock
-	c.RLock()
-
-	// check the cache first
-	services := c.cache[service]
-	// get cache ttl
-	ttl := c.ttls[service]
-
-	// got services && within ttl so return cache
-	if c.isValid(services, ttl) {
-		// make a copy
-		cp := c.cp(services)
-		// unlock the read
-		c.RUnlock()
-		// return servics
-		return cp, nil
-	}
-
-	// get does the actual request for a service and cache it
-	get := func(service string) ([]*registry.Service, error) {
-		// ask the registry
-		services, err := c.so.Registry.GetService(service)
-		if err != nil {
-			return nil, err
-		}
-
-		// cache results
-		c.Lock()
-		c.set(service, c.cp(services))
-		c.Unlock()
-
-		return services, nil
-	}
-
-	// watch service if not watched
-	if _, ok := c.watched[service]; !ok {
-		go c.run(service)
-	}
-
-	// unlock the read lock
-	c.RUnlock()
-
-	// get and return services
-	return get(service)
-}
-
-func (c *registrySelector) set(service string, services []*registry.Service) {
-	c.cache[service] = services
-	c.ttls[service] = time.Now().Add(c.ttl)
-}
-
-func (c *registrySelector) update(res *registry.Result) {
-	if res == nil || res.Service == nil {
-		return
-	}
-
-	c.Lock()
-	defer c.Unlock()
-
-	services, ok := c.cache[res.Service.Name]
-	if !ok {
-		// we're not going to cache anything
-		// unless there was already a lookup
-		return
-	}
-
-	if len(res.Service.Nodes) == 0 {
-		switch res.Action {
-		case "delete":
-			c.del(res.Service.Name)
-		}
-		return
-	}
-
-	// existing service found
-	var service *registry.Service
-	var index int
-	for i, s := range services {
-		if s.Version == res.Service.Version {
-			service = s
-			index = i
+func (c *registrySelector) newRCache() rcache.Cache {
+	ropts := []rcache.Option{}
+	if c.so.Context != nil {
+		if t, ok := c.so.Context.Value("selector_ttl").(time.Duration); ok {
+			ropts = append(ropts, rcache.WithTTL(t))
 		}
 	}
-
-	switch res.Action {
-	case "create", "update":
-		if service == nil {
-			c.set(res.Service.Name, append(services, res.Service))
-			return
-		}
-
-		// append old nodes to new service
-		for _, cur := range service.Nodes {
-			var seen bool
-			for _, node := range res.Service.Nodes {
-				if cur.Id == node.Id {
-					seen = true
-					break
-				}
-			}
-			if !seen {
-				res.Service.Nodes = append(res.Service.Nodes, cur)
-			}
-		}
-
-		services[index] = res.Service
-		c.set(res.Service.Name, services)
-	case "delete":
-		if service == nil {
-			return
-		}
-
-		var nodes []*registry.Node
-
-		// filter cur nodes to remove the dead one
-		for _, cur := range service.Nodes {
-			var seen bool
-			for _, del := range res.Service.Nodes {
-				if del.Id == cur.Id {
-					seen = true
-					break
-				}
-			}
-			if !seen {
-				nodes = append(nodes, cur)
-			}
-		}
-
-		// still got nodes, save and return
-		if len(nodes) > 0 {
-			service.Nodes = nodes
-			services[index] = service
-			c.set(service.Name, services)
-			return
-		}
-
-		// zero nodes left
-
-		// only have one thing to delete
-		// nuke the thing
-		if len(services) == 1 {
-			c.del(service.Name)
-			return
-		}
-
-		// still have more than 1 service
-		// check the version and keep what we know
-		var srvs []*registry.Service
-		for _, s := range services {
-			if s.Version != service.Version {
-				srvs = append(srvs, s)
-			}
-		}
-
-		// save
-		c.set(service.Name, srvs)
-	}
-}
-
-// run starts the cache watcher loop
-// it creates a new watcher if there's a problem
-// reloads the watcher if Init is called
-// and returns when Close is called
-func (c *registrySelector) run(name string) {
-	// set watcher
-	c.Lock()
-	c.watched[name] = true
-	c.Unlock()
-
-	// delete watcher on exit
-	defer func() {
-		c.Lock()
-		delete(c.watched, name)
-		c.Unlock()
-	}()
-
-	// error counter
-	var cerr int
-
-	for {
-		// exit early if already dead
-		if c.quit() {
-			return
-		}
-
-		// create new watcher
-		w, err := c.so.Registry.Watch(
-			registry.WatchService(name),
-		)
-
-		if err != nil {
-			if c.quit() {
-				return
-			}
-			cerr++
-			if cerr > 3 {
-				log.Log(err)
-				cerr = 0
-			}
-			time.Sleep(time.Second)
-			continue
-		}
-
-		// watch for events
-		if err := c.watch(w); err != nil {
-			if c.quit() {
-				return
-			}
-			cerr++
-			if cerr > 3 {
-				cerr = 0
-				log.Log(err)
-			}
-			continue
-		}
-
-		// reset err counter
-		cerr = 0
-	}
-}
-
-// watch loops the next event and calls update
-// it returns if there's an error
-func (c *registrySelector) watch(w registry.Watcher) error {
-	defer w.Stop()
-
-	// manage this loop
-	go func() {
-		// wait for exit or reload signal
-		select {
-		case <-c.exit:
-		case <-c.reload:
-		}
-
-		// stop the watcher
-		w.Stop()
-	}()
-
-	for {
-		res, err := w.Next()
-		if err != nil {
-			return err
-		}
-		c.update(res)
-	}
+	return rcache.New(c.so.Registry, ropts...)
 }
 
 func (c *registrySelector) Init(opts ...Option) error {
@@ -352,15 +27,8 @@ func (c *registrySelector) Init(opts ...Option) error {
 		o(&c.so)
 	}
 
-	// reload the watcher
-	go func() {
-		select {
-		case <-c.exit:
-			return
-		default:
-			c.reload <- true
-		}
-	}()
+	c.rc.Stop()
+	c.rc = c.newRCache()
 
 	return nil
 }
@@ -381,7 +49,7 @@ func (c *registrySelector) Select(service string, opts ...SelectOption) (Next, e
 	// get the service
 	// try the cache first
 	// if that fails go directly to the registry
-	services, err := c.get(service)
+	services, err := c.rc.GetService(service)
 	if err != nil {
 		return nil, err
 	}
@@ -407,17 +75,8 @@ func (c *registrySelector) Reset(service string) {
 
 // Close stops the watcher and destroys the cache
 func (c *registrySelector) Close() error {
-	c.Lock()
-	c.cache = make(map[string][]*registry.Service)
-	c.watched = make(map[string]bool)
-	c.Unlock()
+	c.rc.Stop()
 
-	select {
-	case <-c.exit:
-		return nil
-	default:
-		close(c.exit)
-	}
 	return nil
 }
 
@@ -438,21 +97,10 @@ func NewSelector(opts ...Option) Selector {
 		sopts.Registry = registry.DefaultRegistry
 	}
 
-	ttl := DefaultTTL
-
-	if sopts.Context != nil {
-		if t, ok := sopts.Context.Value("selector_ttl").(time.Duration); ok {
-			ttl = t
-		}
+	s := &registrySelector{
+		so: sopts,
 	}
+	s.rc = s.newRCache()
 
-	return &registrySelector{
-		so:      sopts,
-		ttl:     ttl,
-		watched: make(map[string]bool),
-		cache:   make(map[string][]*registry.Service),
-		ttls:    make(map[string]time.Time),
-		reload:  make(chan bool, 1),
-		exit:    make(chan bool),
-	}
+	return s
 }

server/handler.go

@@ -1,13 +1,20 @@
 package server
 
+import "context"
+
+type HandlerOption func(*HandlerOptions)
+
 type HandlerOptions struct {
 	Internal bool
 	Metadata map[string]map[string]string
 }
 
+type SubscriberOption func(*SubscriberOptions)
+
 type SubscriberOptions struct {
 	Queue    string
 	Internal bool
+	Context  context.Context
 }
 
 // EndpointMetadata is a Handler option that allows metadata to be added to
@@ -34,6 +41,17 @@ func InternalSubscriber(b bool) SubscriberOption {
 		o.Internal = b
 	}
 }
+func NewSubscriberOptions(opts ...SubscriberOption) SubscriberOptions {
+	opt := SubscriberOptions{
+		Context: context.Background(),
+	}
+
+	for _, o := range opts {
+		o(&opt)
+	}
+
+	return opt
+}
 
 // Shared queue name distributed messages across subscribers
 func SubscriberQueue(n string) SubscriberOption {
@@ -41,3 +59,10 @@ func SubscriberQueue(n string) SubscriberOption {
 		o.Queue = n
 	}
 }
+
+// SubscriberContext set context options to allow broker SubscriberOption passed
+func SubscriberContext(ctx context.Context) SubscriberOption {
+	return func(o *SubscriberOptions) {
+		o.Context = ctx
+	}
+}

server/options.go

@@ -27,6 +27,8 @@ type Options struct {
 
 	// The register expiry time
 	RegisterTTL time.Duration
+	// The interval on which to register
+	RegisterInterval time.Duration
 
 	// The router for requests
 	Router Router
@@ -168,6 +170,13 @@ func RegisterTTL(t time.Duration) Option {
 	}
 }
 
+// Register the service with at interval
+func RegisterInterval(t time.Duration) Option {
+	return func(o *Options) {
+		o.RegisterInterval = t
+	}
+}
+
 // WithRouter sets the request router
 func WithRouter(r Router) Option {
 	return func(o *Options) {

server/rpc_codec.go

@@ -66,13 +66,63 @@ func (rwc *readWriteCloser) Close() error {
 	return nil
 }
 
+func getHeader(hdr string, md map[string]string) string {
+	if hd := md[hdr]; len(hd) > 0 {
+		return hd
+	}
+	return md["X-"+hdr]
+}
+
+func getHeaders(m *codec.Message) {
+	get := func(hdr, v string) string {
+		if len(v) > 0 {
+			return v
+		}
+
+		if hd := m.Header[hdr]; len(hd) > 0 {
+			return hd
+		}
+
+		// old
+		return m.Header["X-"+hdr]
+	}
+
+	m.Id = get("Micro-Id", m.Id)
+	m.Error = get("Micro-Error", m.Error)
+	m.Endpoint = get("Micro-Endpoint", m.Endpoint)
+	m.Method = get("Micro-Method", m.Method)
+	m.Target = get("Micro-Service", m.Target)
+
+	// TODO: remove this cruft
+	if len(m.Endpoint) == 0 {
+		m.Endpoint = m.Method
+	}
+}
+
+func setHeaders(m, r *codec.Message) {
+	set := func(hdr, v string) {
+		if len(v) == 0 {
+			return
+		}
+		m.Header[hdr] = v
+		m.Header["X-"+hdr] = v
+	}
+
+	// set headers
+	set("Micro-Id", r.Id)
+	set("Micro-Service", r.Target)
+	set("Micro-Method", r.Method)
+	set("Micro-Endpoint", r.Endpoint)
+	set("Micro-Error", r.Error)
+}
+
 // setupProtocol sets up the old protocol
 func setupProtocol(msg *transport.Message) codec.NewCodec {
-	service := msg.Header["X-Micro-Service"]
-	method := msg.Header["X-Micro-Method"]
-	endpoint := msg.Header["X-Micro-Endpoint"]
-	protocol := msg.Header["X-Micro-Protocol"]
-	target := msg.Header["X-Micro-Target"]
+	service := getHeader("Micro-Service", msg.Header)
+	method := getHeader("Micro-Method", msg.Header)
+	endpoint := getHeader("Micro-Endpoint", msg.Header)
+	protocol := getHeader("Micro-Protocol", msg.Header)
+	target := getHeader("Micro-Target", msg.Header)
 
 	// if the protocol exists (mucp) do nothing
 	if len(protocol) > 0 {
@@ -91,12 +141,12 @@ func setupProtocol(msg *transport.Message) codec.NewCodec {
 
 	// no method then set to endpoint
 	if len(method) == 0 {
-		msg.Header["X-Micro-Method"] = method
+		msg.Header["Micro-Method"] = endpoint
 	}
 
 	// no endpoint then set to method
 	if len(endpoint) == 0 {
-		msg.Header["X-Micro-Endpoint"] = method
+		msg.Header["Micro-Endpoint"] = method
 	}
 
 	return nil
@@ -118,7 +168,7 @@ func newRpcCodec(req *transport.Message, socket transport.Socket, c codec.NewCod
 }
 
 func (c *rpcCodec) ReadHeader(r *codec.Message, t codec.MessageType) error {
-	// the initieal message
+	// the initial message
 	m := codec.Message{
 		Header: c.req.Header,
 		Body:   c.req.Body,
@@ -153,25 +203,22 @@ func (c *rpcCodec) ReadHeader(r *codec.Message, t codec.MessageType) error {
 	c.first = false
 
 	// set some internal things
-	m.Target = m.Header["X-Micro-Service"]
-	m.Method = m.Header["X-Micro-Method"]
-	m.Endpoint = m.Header["X-Micro-Endpoint"]
-	m.Id = m.Header["X-Micro-Id"]
+	getHeaders(&m)
 
 	// read header via codec
-	err := c.codec.ReadHeader(&m, codec.Request)
-
-	// set the method/id
-	r.Method = m.Method
-	r.Endpoint = m.Endpoint
-	r.Id = m.Id
-
-	// TODO: remove the old legacy cruft
-	if len(r.Endpoint) == 0 {
-		r.Endpoint = r.Method
+	if err := c.codec.ReadHeader(&m, codec.Request); err != nil {
+		return err
 	}
 
-	return err
+	// fallback for 0.14 and older
+	if len(m.Endpoint) == 0 {
+		m.Endpoint = m.Method
+	}
+
+	// set message
+	*r = m
+
+	return nil
 }
 
 func (c *rpcCodec) ReadBody(b interface{}) error {
@@ -206,29 +253,7 @@ func (c *rpcCodec) Write(r *codec.Message, b interface{}) error {
 		m.Header = map[string]string{}
 	}
 
-	// set request id
-	if len(r.Id) > 0 {
-		m.Header["X-Micro-Id"] = r.Id
-	}
-
-	// set target
-	if len(r.Target) > 0 {
-		m.Header["X-Micro-Service"] = r.Target
-	}
-
-	// set request method
-	if len(r.Method) > 0 {
-		m.Header["X-Micro-Method"] = r.Method
-	}
-
-	// set request endpoint
-	if len(r.Endpoint) > 0 {
-		m.Header["X-Micro-Endpoint"] = r.Endpoint
-	}
-
-	if len(r.Error) > 0 {
-		m.Header["X-Micro-Error"] = r.Error
-	}
+	setHeaders(m, r)
 
 	// the body being sent
 	var body []byte
@@ -246,6 +271,7 @@ func (c *rpcCodec) Write(r *codec.Message, b interface{}) error {
 		// write an error if it failed
 		m.Error = errors.Wrapf(err, "Unable to encode body").Error()
 		m.Header["X-Micro-Error"] = m.Error
+		m.Header["Micro-Error"] = m.Error
 		// no body to write
 		if err := c.codec.Write(m, nil); err != nil {
 			return err

server/rpc_request.go

@@ -47,6 +47,11 @@ func (r *rpcRequest) Header() map[string]string {
 	return r.header
 }
 
+func (r *rpcRequest) Body() interface{} {
+	// TODO: convert to interface value
+	return r.body
+}
+
 func (r *rpcRequest) Read() ([]byte, error) {
 	// got a body
 	if r.body != nil {

server/rpc_router.go

@@ -162,27 +162,23 @@ func prepareMethod(method reflect.Method) *methodType {
 	return &methodType{method: method, ArgType: argType, ReplyType: replyType, ContextType: contextType, stream: stream}
 }
 
-func (router *router) sendResponse(sending sync.Locker, req *request, reply interface{}, cc codec.Writer, errmsg string, last bool) (err error) {
+func (router *router) sendResponse(sending sync.Locker, req *request, reply interface{}, cc codec.Writer, last bool) error {
 	msg := new(codec.Message)
 	msg.Type = codec.Response
 	resp := router.getResponse()
 	resp.msg = msg
 
-	// Encode the response header
-	resp.msg.Endpoint = req.msg.Endpoint
-	if errmsg != "" {
-		resp.msg.Error = errmsg
-		reply = invalidRequest
-	}
 	resp.msg.Id = req.msg.Id
 	sending.Lock()
-	err = cc.Write(resp.msg, reply)
+	err := cc.Write(resp.msg, reply)
 	sending.Unlock()
 	router.freeResponse(resp)
 	return err
 }
 
-func (s *service) call(ctx context.Context, router *router, sending *sync.Mutex, mtype *methodType, req *request, argv, replyv reflect.Value, cc codec.Writer) {
+func (s *service) call(ctx context.Context, router *router, sending *sync.Mutex, mtype *methodType, req *request, argv, replyv reflect.Value, cc codec.Writer) error {
+	defer router.freeRequest(req)
+
 	function := mtype.method.Func
 	var returnValues []reflect.Value
 
@@ -206,18 +202,13 @@ func (s *service) call(ctx context.Context, router *router, sending *sync.Mutex,
 			return nil
 		}
 
-		errmsg := ""
-		err := fn(ctx, r, replyv.Interface())
-		if err != nil {
-			errmsg = err.Error()
+		// execute handler
+		if err := fn(ctx, r, replyv.Interface()); err != nil {
+			return err
 		}
 
-		err = router.sendResponse(sending, req, replyv.Interface(), cc, errmsg, true)
-		if err != nil {
-			log.Log("rpc call: unable to send response: ", err)
-		}
-		router.freeRequest(req)
-		return
+		// send response
+		return router.sendResponse(sending, req, replyv.Interface(), cc, true)
 	}
 
 	// declare a local error to see if we errored out already
@@ -250,16 +241,15 @@ func (s *service) call(ctx context.Context, router *router, sending *sync.Mutex,
 	// client.Stream request
 	r.stream = true
 
-	errmsg := ""
+	// execute handler
 	if err := fn(ctx, r, stream); err != nil {
-		errmsg = err.Error()
+		return err
 	}
 
 	// this is the last packet, we don't do anything with
 	// the error here (well sendStreamResponse will log it
 	// already)
-	router.sendResponse(sending, req, nil, cc, errmsg, true)
-	router.freeRequest(req)
+	return router.sendResponse(sending, req, nil, cc, true)
 }
 
 func (m *methodType) prepareContext(ctx context.Context) reflect.Value {
@@ -448,11 +438,9 @@ func (router *router) ServeRequest(ctx context.Context, r Request, rsp Response)
 		}
 		// send a response if we actually managed to read a header.
 		if req != nil {
-			router.sendResponse(sending, req, invalidRequest, rsp.Codec(), err.Error(), true)
 			router.freeRequest(req)
 		}
 		return err
 	}
-	service.call(ctx, router, sending, mtype, req, argv, replyv, rsp.Codec())
-	return nil
+	return service.call(ctx, router, sending, mtype, req, argv, replyv, rsp.Codec())
 }

server/rpc_server.go

@@ -10,7 +10,7 @@ import (
 	"sync"
 	"time"
 
-	"github.com/micro/go-log"
+	log "github.com/micro/go-log"
 	"github.com/micro/go-micro/broker"
 	"github.com/micro/go-micro/codec"
 	"github.com/micro/go-micro/metadata"
@@ -120,9 +120,9 @@ func (s *rpcServer) ServeConn(sock transport.Socket) {
 
 		// internal request
 		request := &rpcRequest{
-			service:     msg.Header["X-Micro-Service"],
-			method:      msg.Header["X-Micro-Method"],
-			endpoint:    msg.Header["X-Micro-Endpoint"],
+			service:     getHeader("Micro-Service", msg.Header),
+			method:      getHeader("Micro-Method", msg.Header),
+			endpoint:    getHeader("Micro-Endpoint", msg.Header),
 			contentType: ct,
 			codec:       rcodec,
 			header:      msg.Header,
@@ -157,13 +157,18 @@ func (s *rpcServer) ServeConn(sock transport.Socket) {
 
 		// TODO: handle error better
 		if err := handler(ctx, request, response); err != nil {
-			// write an error response
-			rcodec.Write(&codec.Message{
-				Header: msg.Header,
-				Error:  err.Error(),
-				Type:   codec.Error,
-			}, nil)
-
+			if err != lastStreamResponseError {
+				// write an error response
+				err = rcodec.Write(&codec.Message{
+					Header: msg.Header,
+					Error:  err.Error(),
+					Type:   codec.Error,
+				}, nil)
+				// could not write the error response
+				if err != nil {
+					log.Logf("rpc: unable to write error response: %v", err)
+				}
+			}
 			s.wg.Done()
 			return
 		}
@@ -271,12 +276,18 @@ func (s *rpcServer) Register() error {
 		return err
 	}
 
+	// make copy of metadata
+	md := make(metadata.Metadata)
+	for k, v := range config.Metadata {
+		md[k] = v
+	}
+
 	// register service
 	node := &registry.Node{
 		Id:       config.Name + "-" + config.Id,
 		Address:  addr,
 		Port:     port,
-		Metadata: config.Metadata,
+		Metadata: md,
 	}
 
 	node.Metadata["transport"] = config.Transport.String()
@@ -328,7 +339,7 @@ func (s *rpcServer) Register() error {
 	s.Unlock()
 
 	if !registered {
-		log.Logf("Registering node: %s", node.Id)
+		log.Logf("Registry [%s] Registering node: %s", config.Registry.String(), node.Id)
 	}
 
 	// create registry options
@@ -354,6 +365,9 @@ func (s *rpcServer) Register() error {
 		if queue := sb.Options().Queue; len(queue) > 0 {
 			opts = append(opts, broker.Queue(queue))
 		}
+		if cx := sb.Options().Context; cx != nil {
+			opts = append(opts, broker.SubscribeContext(cx))
+		}
 		sub, err := config.Broker.Subscribe(sb.Topic(), handler, opts...)
 		if err != nil {
 			return err
@@ -403,7 +417,7 @@ func (s *rpcServer) Deregister() error {
 		Nodes:   []*registry.Node{node},
 	}
 
-	log.Logf("Deregistering node: %s", node.Id)
+	log.Logf("Registry [%s] Deregistering node: %s", config.Registry.String(), node.Id)
 	if err := config.Registry.Deregister(service); err != nil {
 		return err
 	}
@@ -433,36 +447,53 @@ func (s *rpcServer) Start() error {
 	registerDebugHandler(s)
 	config := s.Options()
 
+	// start listening on the transport
 	ts, err := config.Transport.Listen(config.Address)
 	if err != nil {
 		return err
 	}
 
-	log.Logf("Transport Listening on %s", ts.Addr())
-	s.Lock()
+	log.Logf("Transport [%s] Listening on %s", config.Transport.String(), ts.Addr())
+
 	// swap address
+	s.Lock()
 	addr := s.opts.Address
 	s.opts.Address = ts.Addr()
 	s.Unlock()
 
-	exit := make(chan bool, 1)
+	// connect to the broker
+	if err := config.Broker.Connect(); err != nil {
+		return err
+	}
+
+	log.Logf("Broker [%s] Connected to %s", config.Broker.String(), config.Broker.Address())
+
+	// announce self to the world
+	if err := s.Register(); err != nil {
+		log.Log("Server register error: ", err)
+	}
+
+	exit := make(chan bool)
 
 	go func() {
 		for {
+			// listen for connections
 			err := ts.Accept(s.ServeConn)
 
-			// check if we're supposed to exit
+			// TODO: listen for messages
+			// msg := broker.Exchange(service).Consume()
+
 			select {
+			// check if we're supposed to exit
 			case <-exit:
 				return
-			default:
-			}
-
 			// check the error and backoff
-			if err != nil {
-				log.Logf("Accept error: %v", err)
-				time.Sleep(time.Second)
-				continue
+			default:
+				if err != nil {
+					log.Logf("Accept error: %v", err)
+					time.Sleep(time.Second)
+					continue
+				}
 			}
 
 			// no error just exit
@@ -471,9 +502,37 @@ func (s *rpcServer) Start() error {
 	}()
 
 	go func() {
-		// wait for exit
-		ch := <-s.exit
-		exit <- true
+		t := new(time.Ticker)
+
+		// only process if it exists
+		if s.opts.RegisterInterval > time.Duration(0) {
+			// new ticker
+			t = time.NewTicker(s.opts.RegisterInterval)
+		}
+
+		// return error chan
+		var ch chan error
+
+	Loop:
+		for {
+			select {
+			// register self on interval
+			case <-t.C:
+				if err := s.Register(); err != nil {
+					log.Log("Server register error: ", err)
+				}
+			// wait for exit
+			case ch = <-s.exit:
+				t.Stop()
+				close(exit)
+				break Loop
+			}
+		}
+
+		// deregister self
+		if err := s.Deregister(); err != nil {
+			log.Log("Server deregister error: ", err)
+		}
 
 		// wait for requests to finish
 		if wait(s.opts.Context) {
@@ -486,14 +545,13 @@ func (s *rpcServer) Start() error {
 		// disconnect the broker
 		config.Broker.Disconnect()
 
-		s.Lock()
 		// swap back address
+		s.Lock()
 		s.opts.Address = addr
 		s.Unlock()
 	}()
 
-	// TODO: subscribe to cruft
-	return config.Broker.Connect()
+	return nil
 }
 
 func (s *rpcServer) Stop() error {

server/server.go

@@ -21,8 +21,6 @@ type Server interface {
 	NewHandler(interface{}, ...HandlerOption) Handler
 	NewSubscriber(string, interface{}, ...SubscriberOption) Subscriber
 	Subscribe(Subscriber) error
-	Register() error
-	Deregister() error
 	Start() error
 	Stop() error
 	String() string
@@ -53,6 +51,8 @@ type Request interface {
 	ContentType() string
 	// Header of the request
 	Header() map[string]string
+	// Body is the initial decoded value
+	Body() interface{}
 	// Read the undecoded request body
 	Read() ([]byte, error)
 	// The encoded message stream
@@ -114,14 +114,10 @@ type Subscriber interface {
 
 type Option func(*Options)
 
-type HandlerOption func(*HandlerOptions)
-
-type SubscriberOption func(*SubscriberOptions)
-
 var (
 	DefaultAddress        = ":0"
-	DefaultName           = "go-server"
-	DefaultVersion        = "1.0.0"
+	DefaultName           = "server"
+	DefaultVersion        = "latest"
 	DefaultId             = uuid.New().String()
 	DefaultServer  Server = newRpcServer()
 	DefaultRouter         = newRpcRouter()
@@ -177,16 +173,6 @@ func Subscribe(s Subscriber) error {
 	return DefaultServer.Subscribe(s)
 }
 
-// Register registers the default server with the discovery system
-func Register() error {
-	return DefaultServer.Register()
-}
-
-// Deregister deregisters the default server from the discovery system
-func Deregister() error {
-	return DefaultServer.Deregister()
-}
-
 // Run starts the default server and waits for a kill
 // signal before exiting. Also registers/deregisters the server
 func Run() error {
@@ -194,18 +180,10 @@ func Run() error {
 		return err
 	}
 
-	if err := DefaultServer.Register(); err != nil {
-		return err
-	}
-
 	ch := make(chan os.Signal, 1)
 	signal.Notify(ch, syscall.SIGTERM, syscall.SIGINT, syscall.SIGKILL)
 	log.Logf("Received signal %s", <-ch)
 
-	if err := DefaultServer.Deregister(); err != nil {
-		return err
-	}
-
 	return Stop()
 }
 

service.go

@@ -5,10 +5,7 @@ import (
 	"os/signal"
 	"sync"
 	"syscall"
-	"time"
 
-	"github.com/micro/cli"
-	"github.com/micro/go-log"
 	"github.com/micro/go-micro/client"
 	"github.com/micro/go-micro/cmd"
 	"github.com/micro/go-micro/metadata"
@@ -36,27 +33,6 @@ func newService(opts ...Option) Service {
 	}
 }
 
-func (s *service) run(exit chan bool) {
-	if s.opts.RegisterInterval <= time.Duration(0) {
-		return
-	}
-
-	t := time.NewTicker(s.opts.RegisterInterval)
-
-	for {
-		select {
-		case <-t.C:
-			err := s.opts.Server.Register()
-			if err != nil {
-				log.Log("service run Server.Register error: ", err)
-			}
-		case <-exit:
-			t.Stop()
-			return
-		}
-	}
-}
-
 // Init initialises options. Additionally it calls cmd.Init
 // which parses command line flags. cmd.Init is only called
 // on first Init.
@@ -67,20 +43,6 @@ func (s *service) Init(opts ...Option) {
 	}
 
 	s.once.Do(func() {
-		// save user action
-		action := s.opts.Cmd.App().Action
-
-		// set service action
-		s.opts.Cmd.App().Action = func(c *cli.Context) {
-			// set register interval
-			if i := time.Duration(c.GlobalInt("register_interval")); i > 0 {
-				s.opts.RegisterInterval = i * time.Second
-			}
-
-			// user action
-			action(c)
-		}
-
 		// Initialise the command flags, overriding new service
 		_ = s.opts.Cmd.Init(
 			cmd.Broker(&s.opts.Broker),
@@ -105,7 +67,7 @@ func (s *service) Server() server.Server {
 }
 
 func (s *service) String() string {
-	return "go-micro"
+	return "micro"
 }
 
 func (s *service) Start() error {
@@ -119,10 +81,6 @@ func (s *service) Start() error {
 		return err
 	}
 
-	if err := s.opts.Server.Register(); err != nil {
-		return err
-	}
-
 	for _, fn := range s.opts.AfterStart {
 		if err := fn(); err != nil {
 			return err
@@ -141,10 +99,6 @@ func (s *service) Stop() error {
 		}
 	}
 
-	if err := s.opts.Server.Deregister(); err != nil {
-		return err
-	}
-
 	if err := s.opts.Server.Stop(); err != nil {
 		return err
 	}
@@ -163,10 +117,6 @@ func (s *service) Run() error {
 		return err
 	}
 
-	// start reg loop
-	ex := make(chan bool)
-	go s.run(ex)
-
 	ch := make(chan os.Signal, 1)
 	signal.Notify(ch, syscall.SIGTERM, syscall.SIGINT, syscall.SIGQUIT)
 
@@ -177,8 +127,5 @@ func (s *service) Run() error {
 	case <-s.opts.Context.Done():
 	}
 
-	// exit reg loop
-	close(ex)
-
 	return s.Stop()
 }

transport/http_transport.go

@@ -13,11 +13,11 @@ import (
 	"sync"
 	"time"
 
-	"github.com/micro/h2c"
 	maddr "github.com/micro/util/go/lib/addr"
 	mnet "github.com/micro/util/go/lib/net"
 	mls "github.com/micro/util/go/lib/tls"
 	"golang.org/x/net/http2"
+	"golang.org/x/net/http2/h2c"
 )
 
 type buffer struct {
@@ -35,7 +35,7 @@ type httpTransportClient struct {
 	dialOpts DialOptions
 	once     sync.Once
 
-	sync.Mutex
+	sync.RWMutex
 	r    chan *http.Request
 	bl   []*http.Request
 	buff *bufio.Reader
@@ -133,12 +133,6 @@ func (h *httpTransportClient) Recv(m *Message) error {
 		r = rc
 	}
 
-	h.Lock()
-	defer h.Unlock()
-	if h.buff == nil {
-		return io.EOF
-	}
-
 	// set timeout if its greater than 0
 	if h.ht.opts.Timeout > time.Duration(0) {
 		h.conn.SetDeadline(time.Now().Add(h.ht.opts.Timeout))
@@ -181,7 +175,6 @@ func (h *httpTransportClient) Close() error {
 	h.once.Do(func() {
 		h.Lock()
 		h.buff.Reset(nil)
-		h.buff = nil
 		h.Unlock()
 		close(h.r)
 	})
@@ -424,10 +417,7 @@ func (h *httpTransportListener) Accept(fn func(Socket)) error {
 
 	// insecure connection use h2c
 	if !(h.ht.opts.Secure || h.ht.opts.TLSConfig != nil) {
-		srv.Handler = &h2c.HandlerH2C{
-			Handler:  mux,
-			H2Server: &http2.Server{},
-		}
+		srv.Handler = h2c.NewHandler(mux, &http2.Server{})
 	}
 
 	// begin serving