Merge pull request #248 from micro/rework

Rework Interfaces

.travis.yml

@@ -1,7 +1,7 @@
 language: go
 go:
-- 1.8.5
-- 1.9.2
+- 1.9.5
+- 1.10.x
 notifications:
   slack:
     secure: aEvhLbhujaGaKSrOokiG3//PaVHTIrc3fBpoRbCRqfZpyq6WREoapJJhF+tIpWWOwaC9GmChbD6aHo/jMUgwKXVyPSaNjiEL87YzUUpL8B2zslNp1rgfTg/LrzthOx3Q1TYwpaAl3to0fuHUVFX4yMeC2vuThq7WSXgMMxFCtbc=

README.md

@@ -2,9 +2,9 @@
 
 Go Micro is a pluggable RPC framework for distributed systems development.
 
-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. It comes with built in support for {json,proto}-rpc encoding, consul or multicast dns for service discovery, http for communication and random hashed client side load balancing.
+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. It comes with built in support for {json,proto}-rpc encoding, consul or multicast dns for service discovery, http for communication and random hashed client side load balancing.
 
-Everything in go-micro is **pluggable**. You can find and contribute to plugins at [github.com/micro/go-plugins](https://github.com/micro/go-plugins).
+Plugins are available at [github.com/micro/go-plugins](https://github.com/micro/go-plugins).
 
 Follow us on [Twitter](https://twitter.com/microhq) or join the [Slack](http://slack.micro.mu/) community.
 
@@ -14,10 +14,9 @@ Go Micro abstracts away the details of distributed systems. Here are the main fe
 
 - **Service Discovery** - Automatic service registration and name resolution
 - **Load Balancing** - Client side load balancing built on discovery
-- **Sync Comms** - RPC based communication with support for bidirectional streaming
-- **Async Comms** - Native PubSub messaging built in for event driven architectures
-- **Message Encoding** - Dynamic encoding based on content-type with protobuf and json out of the box
-- **Service Interface** - All features are wrapped up in a simple high level interface
+- **Message Encoding** - Dynamic encoding based on content-type with protobuf and json support
+- **Sync Streaming** - RPC based communication with support for bidirectional streaming
+- **Async Messaging** - Native PubSub messaging built in for event driven architectures
 
 Go Micro supports both the Service and Function programming models. Read on to learn more.
 
@@ -35,15 +34,25 @@ Watch the [Golang UK Conf 2016](https://www.youtube.com/watch?v=xspaDovwk34) vid
 
 ## Getting started
 
+- [Install Protobuf](#install-protobuf)
 - [Service Discovery](#service-discovery)
 - [Writing a Service](#writing-a-service)
 - [Writing a Function](#writing-a-function)
+- [Publish & Subscribe](#publish--subscribe)
 - [Plugins](#plugins)
 - [Wrappers](#wrappers)
 
+## Install Protobuf
+
+Protobuf is required for code generation
+
+You'll need to install:
+
+- [protoc-gen-micro](https://github.com/micro/protoc-gen-micro)
+
 ## Service Discovery
 
-Service discovery is used to resolve service names to addresses. It's the only dependency of go-micro.
+Service discovery is used to resolve service names to addresses. 
 
 ### Consul
 
@@ -60,7 +69,7 @@ Discovery is pluggable. Find plugins for etcd, kubernetes, zookeeper and more in
 Pass `--registry=mdns` to any command or the enviroment variable `MICRO_REGISTRY=mdns`
 
 ```
-go run main.go --registry=mdns
+MICRO_REGISTRY=mdns go run main.go
 ```
 
 ## Writing a service
@@ -91,23 +100,12 @@ message HelloResponse {
 }
 ```
 
-### Install protobuf
-
-Install [protobuf](https://developers.google.com/protocol-buffers/)
-
-Now install the micro fork of protoc-gen-go. The protobuf compiler for Go. 
-
-```shell
-go get github.com/micro/protobuf/{proto,protoc-gen-go}
-```
-
 ### Generate the proto
 
 After writing the proto definition we must compile it using protoc with the micro plugin.
 
 ```shell
-protoc -I$GOPATH/src --go_out=plugins=micro:$GOPATH/src \
-	$GOPATH/src/github.com/micro/examples/service/proto/greeter.proto
+protoc --proto_path=$GOPATH/src:. --micro_out=. --go_out=. path/to/greeter.proto
 ```
 
 ### Write the service
@@ -125,11 +123,11 @@ It does the following:
 package main
 
 import (
+	"context"
 	"fmt"
 
 	micro "github.com/micro/go-micro"
 	proto "github.com/micro/examples/service/proto"
-	"golang.org/x/net/context"
 )
 
 type Greeter struct{}
@@ -180,20 +178,21 @@ The generated proto includes a greeter client to reduce boilerplate code.
 package main
 
 import (
+	"context"
 	"fmt"
 
 	micro "github.com/micro/go-micro"
 	proto "github.com/micro/examples/service/proto"
-	"golang.org/x/net/context"
 )
 
 
 func main() {
 	// Create a new service. Optionally include some options here.
 	service := micro.NewService(micro.Name("greeter.client"))
+	service.Init()
 
 	// Create new greeter client
-	greeter := proto.NewGreeterClient("greeter", service.Client())
+	greeter := proto.GreeterServiceClient("greeter", service.Client())
 
 	// Call the greeter
 	rsp, err := greeter.Hello(context.TODO(), &proto.HelloRequest{Name: "John"})
@@ -229,9 +228,10 @@ A Function is a one time executing Service which exits after completing a reques
 package main
 
 import (
+	"context"
+
 	proto "github.com/micro/examples/function/proto"
 	"github.com/micro/go-micro"
-	"golang.org/x/net/context"
 )
 
 type Greeter struct{}
@@ -244,7 +244,7 @@ func (g *Greeter) Hello(ctx context.Context, req *proto.HelloRequest, rsp *proto
 func main() {
 	// create a new function
 	fnc := micro.NewFunction(
-		micro.Name("go.micro.fnc.greeter"),
+		micro.Name("greeter"),
 	)
 
 	// init the command line
@@ -260,6 +260,47 @@ func main() {
 
 It's that simple.
 
+## Publish & Subscribe
+
+Go-micro has a built in message broker interface for event driven architectures.
+
+PubSub operates on the same protobuf generated messages as RPC. They are encoded/decoded automatically and sent via the broker. 
+By default go-micro includes a point-to-point http broker but this can be swapped out via go-plugins.
+
+### Publish
+
+
+Create a new publisher with a `topic` name and service client
+
+```go
+p := micro.NewPublisher("events", service.Client())
+```
+
+Publish a proto message
+
+```go
+p.Publish(context.TODO(), &proto.Event{Name: "event"})
+```
+
+### Subscribe
+
+Create a message handler. It's signature should be `func(context.Context, v interface{}) error`.
+
+```go
+func ProcessEvent(ctx context.Context, event *proto.Event) error {
+	fmt.Printf("Got event %+v\n", event)
+	return nil
+}
+```
+
+Register the message handler with a `topic`
+
+```go
+micro.RegisterSubscriber("events", ProcessEvent)
+```
+
+See [examples/pubsub](https://github.com/micro/examples/tree/master/pubsub) for a complete example.
+
 ## Plugins
 
 By default go-micro only provides a few implementation of each interface at the core but it's completely pluggable. There's already dozens of plugins which are available at [github.com/micro/go-plugins](https://github.com/micro/go-plugins). Contributions are welcome!
@@ -411,6 +452,8 @@ Check out [ja-micro](https://github.com/Sixt/ja-micro) to write services in Java
 
 ## Sponsors
 
-Open source development of Micro is sponsored by Sixt
+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

@@ -2,6 +2,7 @@ package broker
 
 import (
 	"bytes"
+	"context"
 	"crypto/tls"
 	"errors"
 	"fmt"
@@ -22,12 +23,10 @@ import (
 	merr "github.com/micro/go-micro/errors"
 	"github.com/micro/go-micro/registry"
 	"github.com/micro/go-rcache"
-	maddr "github.com/micro/misc/lib/addr"
-	mnet "github.com/micro/misc/lib/net"
-	mls "github.com/micro/misc/lib/tls"
+	maddr "github.com/micro/util/go/lib/addr"
+	mnet "github.com/micro/util/go/lib/net"
+	mls "github.com/micro/util/go/lib/tls"
 	"github.com/pborman/uuid"
-
-	"golang.org/x/net/context"
 )
 
 // HTTP Broker is a point to point async broker

broker/options.go

@@ -1,11 +1,11 @@
 package broker
 
 import (
+	"context"
 	"crypto/tls"
 
 	"github.com/micro/go-micro/broker/codec"
 	"github.com/micro/go-micro/registry"
-	"golang.org/x/net/context"
 )
 
 type Options struct {

client/backoff.go

@@ -1,10 +1,9 @@
 package client
 
 import (
+	"context"
 	"math"
 	"time"
-
-	"golang.org/x/net/context"
 )
 
 type BackoffFunc func(ctx context.Context, req Request, attempts int) (time.Duration, error)

client/backoff_test.go

@@ -1,18 +1,19 @@
 package client
 
 import (
+	"context"
 	"math"
 	"testing"
 	"time"
-
-	"golang.org/x/net/context"
 )
 
 func TestBackoff(t *testing.T) {
 	delta := time.Duration(0)
 
+	c := NewClient()
+
 	for i := 0; i < 5; i++ {
-		d, err := exponentialBackoff(context.TODO(), NewJsonRequest("test", "test", nil), i)
+		d, err := exponentialBackoff(context.TODO(), c.NewRequest("test", "test", nil), i)
 		if err != nil {
 			t.Fatal(err)
 		}

client/client.go

@@ -2,9 +2,8 @@
 package client
 
 import (
+	"context"
 	"time"
-
-	"golang.org/x/net/context"
 )
 
 // Client is the interface used to make requests to services.
@@ -13,22 +12,18 @@ import (
 type Client interface {
 	Init(...Option) error
 	Options() Options
-	NewPublication(topic string, msg interface{}) Publication
+	NewMessage(topic string, msg interface{}) Message
 	NewRequest(service, method string, req interface{}, reqOpts ...RequestOption) Request
-	NewProtoRequest(service, method string, req interface{}, reqOpts ...RequestOption) Request
-	NewJsonRequest(service, method string, req interface{}, reqOpts ...RequestOption) Request
 	Call(ctx context.Context, req Request, rsp interface{}, opts ...CallOption) error
-	CallRemote(ctx context.Context, addr string, req Request, rsp interface{}, opts ...CallOption) error
-	Stream(ctx context.Context, req Request, opts ...CallOption) (Streamer, error)
-	StreamRemote(ctx context.Context, addr string, req Request, opts ...CallOption) (Streamer, error)
-	Publish(ctx context.Context, p Publication, opts ...PublishOption) error
+	Stream(ctx context.Context, req Request, opts ...CallOption) (Stream, error)
+	Publish(ctx context.Context, msg Message, opts ...PublishOption) error
 	String() string
 }
 
-// Publication is the interface for a message published asynchronously
-type Publication interface {
+// Message is the interface for publishing asynchronously
+type Message interface {
 	Topic() string
-	Message() interface{}
+	Payload() interface{}
 	ContentType() string
 }
 
@@ -42,8 +37,8 @@ type Request interface {
 	Stream() bool
 }
 
-// Streamer is the inteface for a bidirectional synchronous stream
-type Streamer interface {
+// Stream is the inteface for a bidirectional synchronous stream
+type Stream interface {
 	Context() context.Context
 	Request() Request
 	Send(interface{}) error
@@ -86,26 +81,15 @@ func Call(ctx context.Context, request Request, response interface{}, opts ...Ca
 	return DefaultClient.Call(ctx, request, response, opts...)
 }
 
-// Makes a synchronous call to the specified address using the default client
-func CallRemote(ctx context.Context, address string, request Request, response interface{}, opts ...CallOption) error {
-	return DefaultClient.CallRemote(ctx, address, request, response, opts...)
-}
-
-// Creates a streaming connection with a service and returns responses on the
-// channel passed in. It's up to the user to close the streamer.
-func Stream(ctx context.Context, request Request, opts ...CallOption) (Streamer, error) {
-	return DefaultClient.Stream(ctx, request, opts...)
-}
-
-// Creates a streaming connection to the address specified.
-func StreamRemote(ctx context.Context, address string, request Request, opts ...CallOption) (Streamer, error) {
-	return DefaultClient.StreamRemote(ctx, address, request, opts...)
-}
-
 // Publishes a publication using the default client. Using the underlying broker
 // set within the options.
-func Publish(ctx context.Context, p Publication) error {
-	return DefaultClient.Publish(ctx, p)
+func Publish(ctx context.Context, msg Message) error {
+	return DefaultClient.Publish(ctx, msg)
+}
+
+// Creates a new message using the default client
+func NewMessage(topic string, payload interface{}) Message {
+	return DefaultClient.NewMessage(topic, payload)
 }
 
 // Creates a new client with the options passed in
@@ -113,25 +97,16 @@ func NewClient(opt ...Option) Client {
 	return newRpcClient(opt...)
 }
 
-// Creates a new publication using the default client
-func NewPublication(topic string, message interface{}) Publication {
-	return DefaultClient.NewPublication(topic, message)
-}
-
 // Creates a new request using the default client. Content Type will
 // be set to the default within options and use the appropriate codec
 func NewRequest(service, method string, request interface{}, reqOpts ...RequestOption) Request {
 	return DefaultClient.NewRequest(service, method, request, reqOpts...)
 }
 
-// Creates a new protobuf request using the default client
-func NewProtoRequest(service, method string, request interface{}, reqOpts ...RequestOption) Request {
-	return DefaultClient.NewProtoRequest(service, method, request, reqOpts...)
-}
-
-// Creates a new json request using the default client
-func NewJsonRequest(service, method string, request interface{}, reqOpts ...RequestOption) Request {
-	return DefaultClient.NewJsonRequest(service, method, request, reqOpts...)
+// Creates a streaming connection with a service and returns responses on the
+// channel passed in. It's up to the user to close the streamer.
+func NewStream(ctx context.Context, request Request, opts ...CallOption) (Stream, error) {
+	return DefaultClient.Stream(ctx, request, opts...)
 }
 
 func String() string {

client/client_wrapper.go

@@ -1,51 +0,0 @@
-package client
-
-/*
-Wrapper is a type of middleware for the go-micro client. It allows
-the client to be "wrapped" so that requests and responses can be intercepted
-to perform extra requirements such as auth, tracing, monitoring, logging, etc.
-
-Example usage:
-
-	import (
-		"log"
-		"github.com/micro/go-micro/client"
-
-	)
-
-	type LogWrapper struct {
-		client.Client
-	}
-
-	func (l *LogWrapper) Call(ctx context.Context, req client.Request, rsp interface{}, opts ...client.CallOption) error {
-		log.Println("Making request to service " + req.Service() + " method " + req.Method())
-		return w.Client.Call(ctx, req, rsp)
-	}
-
-	func Wrapper(c client.Client) client.Client {
-		return &LogWrapper{c}
-	}
-
-	func main() {
-		c := client.NewClient(client.Wrap(Wrapper))
-
-	}
-
-
-*/
-
-import (
-	"golang.org/x/net/context"
-)
-
-// CallFunc represents the individual call func
-type CallFunc func(ctx context.Context, address string, req Request, rsp interface{}, opts CallOptions) error
-
-// CallWrapper is a low level wrapper for the CallFunc
-type CallWrapper func(CallFunc) CallFunc
-
-// Wrapper wraps a client and returns a client
-type Wrapper func(Client) Client
-
-// StreamWrapper wraps a Stream and returns the equivalent
-type StreamWrapper func(Streamer) Streamer

client/context.go

@@ -1,7 +1,7 @@
 package client
 
 import (
-	"golang.org/x/net/context"
+	"context"
 )
 
 type clientKey struct{}

client/mock/context.go

@@ -1,7 +1,7 @@
 package mock
 
 import (
-	"golang.org/x/net/context"
+	"context"
 )
 
 type responseKey struct{}

client/mock/mock.go

@@ -1,14 +1,13 @@
 package mock
 
 import (
+	"context"
 	"fmt"
 	"reflect"
 	"sync"
 
 	"github.com/micro/go-micro/client"
 	"github.com/micro/go-micro/errors"
-
-	"golang.org/x/net/context"
 )
 
 var (
@@ -50,22 +49,14 @@ func (m *MockClient) Options() client.Options {
 	return m.Opts
 }
 
-func (m *MockClient) NewPublication(topic string, msg interface{}) client.Publication {
-	return m.Client.NewPublication(topic, msg)
+func (m *MockClient) NewMessage(topic string, msg interface{}) client.Message {
+	return m.Client.NewMessage(topic, msg)
 }
 
 func (m *MockClient) NewRequest(service, method string, req interface{}, reqOpts ...client.RequestOption) client.Request {
 	return m.Client.NewRequest(service, method, req, reqOpts...)
 }
 
-func (m *MockClient) NewProtoRequest(service, method string, req interface{}, reqOpts ...client.RequestOption) client.Request {
-	return m.Client.NewProtoRequest(service, method, req, reqOpts...)
-}
-
-func (m *MockClient) NewJsonRequest(service, method string, req interface{}, reqOpts ...client.RequestOption) client.Request {
-	return m.Client.NewJsonRequest(service, method, req, reqOpts...)
-}
-
 func (m *MockClient) Call(ctx context.Context, req client.Request, rsp interface{}, opts ...client.CallOption) error {
 	m.Lock()
 	defer m.Unlock()
@@ -98,39 +89,7 @@ func (m *MockClient) Call(ctx context.Context, req client.Request, rsp interface
 	return fmt.Errorf("rpc: can't find service %s", req.Method())
 }
 
-func (m *MockClient) CallRemote(ctx context.Context, addr string, req client.Request, rsp interface{}, opts ...client.CallOption) error {
-	m.Lock()
-	defer m.Unlock()
-
-	response, ok := m.Response[req.Service()]
-	if !ok {
-		return errors.NotFound("go.micro.client.mock", "service not found")
-	}
-
-	for _, r := range response {
-		if r.Method != req.Method() {
-			continue
-		}
-
-		if r.Error != nil {
-			return r.Error
-		}
-
-		v := reflect.ValueOf(rsp)
-
-		if t := reflect.TypeOf(rsp); t.Kind() == reflect.Ptr {
-			v = reflect.Indirect(v)
-		}
-
-		v.Set(reflect.ValueOf(r.Response))
-
-		return nil
-	}
-
-	return fmt.Errorf("rpc: can't find service %s", req.Method())
-}
-
-func (m *MockClient) Stream(ctx context.Context, req client.Request, opts ...client.CallOption) (client.Streamer, error) {
+func (m *MockClient) Stream(ctx context.Context, req client.Request, opts ...client.CallOption) (client.Stream, error) {
 	m.Lock()
 	defer m.Unlock()
 
@@ -138,15 +97,7 @@ func (m *MockClient) Stream(ctx context.Context, req client.Request, opts ...cli
 	return nil, nil
 }
 
-func (m *MockClient) StreamRemote(ctx context.Context, addr string, req client.Request, opts ...client.CallOption) (client.Streamer, error) {
-	m.Lock()
-	defer m.Unlock()
-
-	// TODO: mock stream
-	return nil, nil
-}
-
-func (m *MockClient) Publish(ctx context.Context, p client.Publication, opts ...client.PublishOption) error {
+func (m *MockClient) Publish(ctx context.Context, p client.Message, opts ...client.PublishOption) error {
 	return nil
 }
 

client/mock/mock_test.go

@@ -1,11 +1,10 @@
 package mock
 
 import (
+	"context"
 	"testing"
 
 	"github.com/micro/go-micro/errors"
-
-	"golang.org/x/net/context"
 )
 
 func TestClient(t *testing.T) {
@@ -22,7 +21,7 @@ func TestClient(t *testing.T) {
 	c := NewClient(Response("go.mock", response))
 
 	for _, r := range response {
-		req := c.NewJsonRequest("go.mock", r.Method, map[string]interface{}{"foo": "bar"})
+		req := c.NewRequest("go.mock", r.Method, map[string]interface{}{"foo": "bar"})
 		var rsp interface{}
 
 		err := c.Call(context.TODO(), req, &rsp)

client/options.go

@@ -1,6 +1,7 @@
 package client
 
 import (
+	"context"
 	"time"
 
 	"github.com/micro/go-micro/broker"
@@ -8,8 +9,6 @@ import (
 	"github.com/micro/go-micro/registry"
 	"github.com/micro/go-micro/selector"
 	"github.com/micro/go-micro/transport"
-
-	"golang.org/x/net/context"
 )
 
 type Options struct {
@@ -41,6 +40,8 @@ type Options struct {
 type CallOptions struct {
 	SelectOptions []selector.SelectOption
 
+	// Address of remote host
+	Address string
 	// Backoff func
 	Backoff BackoffFunc
 	// Check if retriable func
@@ -67,7 +68,8 @@ type PublishOptions struct {
 }
 
 type RequestOptions struct {
-	Stream bool
+	ContentType string
+	Stream      bool
 
 	// Other options for implementations of the interface
 	// can be stored in a context
@@ -227,6 +229,13 @@ func DialTimeout(d time.Duration) Option {
 
 // Call Options
 
+// WithAddress sets the remote address to use rather than using service discovery
+func WithAddress(a string) CallOption {
+	return func(o *CallOptions) {
+		o.Address = a
+	}
+}
+
 func WithSelectOption(so ...selector.SelectOption) CallOption {
 	return func(o *CallOptions) {
 		o.SelectOptions = append(o.SelectOptions, so...)
@@ -282,6 +291,12 @@ func WithDialTimeout(d time.Duration) CallOption {
 
 // Request Options
 
+func WithContentType(ct string) RequestOption {
+	return func(o *RequestOptions) {
+		o.ContentType = ct
+	}
+}
+
 func StreamingRequest() RequestOption {
 	return func(o *RequestOptions) {
 		o.Stream = true

client/retry.go

@@ -1,6 +1,8 @@
 package client
 
-import "golang.org/x/net/context"
+import (
+	"context"
+)
 
 // note that returning either false or a non-nil error will result in the call not being retried
 type RetryFunc func(ctx context.Context, req Request, retryCount int, err error) (bool, error)

client/rpc_client.go

@@ -2,6 +2,7 @@ package client
 
 import (
 	"bytes"
+	"context"
 	"fmt"
 	"sync"
 	"time"
@@ -10,16 +11,17 @@ import (
 	"github.com/micro/go-micro/codec"
 	"github.com/micro/go-micro/errors"
 	"github.com/micro/go-micro/metadata"
+	"github.com/micro/go-micro/registry"
 	"github.com/micro/go-micro/selector"
 	"github.com/micro/go-micro/transport"
-
-	"golang.org/x/net/context"
+	"sync/atomic"
 )
 
 type rpcClient struct {
 	once sync.Once
 	opts Options
 	pool *pool
+	seq  uint64
 }
 
 func newRpcClient(opt ...Option) Client {
@@ -29,6 +31,7 @@ func newRpcClient(opt ...Option) Client {
 		once: sync.Once{},
 		opts: opts,
 		pool: newPool(opts.PoolSize, opts.PoolTTL),
+		seq:  0,
 	}
 
 	c := Client(rc)
@@ -85,11 +88,15 @@ func (r *rpcClient) call(ctx context.Context, address string, req Request, resp
 		r.pool.release(address, c, grr)
 	}()
 
+	seq := r.seq
+	atomic.AddUint64(&r.seq, 1)
+
 	stream := &rpcStream{
 		context: ctx,
 		request: req,
 		closed:  make(chan bool),
 		codec:   newRpcPlusCodec(msg, c, cf),
+		seq:     seq,
 	}
 	defer stream.Close()
 
@@ -128,7 +135,7 @@ func (r *rpcClient) call(ctx context.Context, address string, req Request, resp
 	}
 }
 
-func (r *rpcClient) stream(ctx context.Context, address string, req Request, opts CallOptions) (Streamer, error) {
+func (r *rpcClient) stream(ctx context.Context, address string, req Request, opts CallOptions) (Stream, error) {
 	msg := &transport.Message{
 		Header: make(map[string]string),
 	}
@@ -207,13 +214,25 @@ func (r *rpcClient) Options() Options {
 	return r.opts
 }
 
-func (r *rpcClient) CallRemote(ctx context.Context, address string, request Request, response interface{}, opts ...CallOption) error {
-	// make a copy of call opts
-	callOpts := r.opts.CallOptions
-	for _, opt := range opts {
-		opt(&callOpts)
+func (r *rpcClient) next(request Request, opts CallOptions) (selector.Next, error) {
+	// return remote address
+	if len(opts.Address) > 0 {
+		return func() (*registry.Node, error) {
+			return &registry.Node{
+				Address: opts.Address,
+			}, nil
+		}, nil
 	}
-	return r.call(ctx, address, request, response, callOpts)
+
+	// get next nodes from the selector
+	next, err := r.opts.Selector.Select(request.Service(), opts.SelectOptions...)
+	if err != nil && err == selector.ErrNotFound {
+		return nil, errors.NotFound("go.micro.client", err.Error())
+	} else if err != nil {
+		return nil, errors.InternalServerError("go.micro.client", err.Error())
+	}
+
+	return next, nil
 }
 
 func (r *rpcClient) Call(ctx context.Context, request Request, response interface{}, opts ...CallOption) error {
@@ -223,12 +242,9 @@ func (r *rpcClient) Call(ctx context.Context, request Request, response interfac
 		opt(&callOpts)
 	}
 
-	// get next nodes from the selector
-	next, err := r.opts.Selector.Select(request.Service(), callOpts.SelectOptions...)
-	if err != nil && err == selector.ErrNotFound {
-		return errors.NotFound("go.micro.client", err.Error())
-	} else if err != nil {
-		return errors.InternalServerError("go.micro.client", err.Error())
+	next, err := r.next(request, callOpts)
+	if err != nil {
+		return err
 	}
 
 	// check if we already have a deadline
@@ -324,28 +340,16 @@ func (r *rpcClient) Call(ctx context.Context, request Request, response interfac
 	return gerr
 }
 
-func (r *rpcClient) StreamRemote(ctx context.Context, address string, request Request, opts ...CallOption) (Streamer, error) {
-	// make a copy of call opts
-	callOpts := r.opts.CallOptions
-	for _, opt := range opts {
-		opt(&callOpts)
-	}
-	return r.stream(ctx, address, request, callOpts)
-}
-
-func (r *rpcClient) Stream(ctx context.Context, request Request, opts ...CallOption) (Streamer, error) {
+func (r *rpcClient) Stream(ctx context.Context, request Request, opts ...CallOption) (Stream, error) {
 	// make a copy of call opts
 	callOpts := r.opts.CallOptions
 	for _, opt := range opts {
 		opt(&callOpts)
 	}
 
-	// get next nodes from the selector
-	next, err := r.opts.Selector.Select(request.Service(), callOpts.SelectOptions...)
-	if err != nil && err == selector.ErrNotFound {
-		return nil, errors.NotFound("go.micro.client", err.Error())
-	} else if err != nil {
-		return nil, errors.InternalServerError("go.micro.client", err.Error())
+	next, err := r.next(request, callOpts)
+	if err != nil {
+		return nil, err
 	}
 
 	// check if we already have a deadline
@@ -367,7 +371,7 @@ func (r *rpcClient) Stream(ctx context.Context, request Request, opts ...CallOpt
 	default:
 	}
 
-	call := func(i int) (Streamer, error) {
+	call := func(i int) (Stream, error) {
 		// call backoff first. Someone may want an initial start delay
 		t, err := callOpts.Backoff(ctx, request, i)
 		if err != nil {
@@ -397,7 +401,7 @@ func (r *rpcClient) Stream(ctx context.Context, request Request, opts ...CallOpt
 	}
 
 	type response struct {
-		stream Streamer
+		stream Stream
 		err    error
 	}
 
@@ -435,49 +439,38 @@ func (r *rpcClient) Stream(ctx context.Context, request Request, opts ...CallOpt
 	return nil, grr
 }
 
-func (r *rpcClient) Publish(ctx context.Context, p Publication, opts ...PublishOption) error {
+func (r *rpcClient) Publish(ctx context.Context, msg Message, opts ...PublishOption) error {
 	md, ok := metadata.FromContext(ctx)
 	if !ok {
 		md = make(map[string]string)
 	}
-	md["Content-Type"] = p.ContentType()
+	md["Content-Type"] = msg.ContentType()
 
 	// encode message body
-	cf, err := r.newCodec(p.ContentType())
+	cf, err := r.newCodec(msg.ContentType())
 	if err != nil {
 		return errors.InternalServerError("go.micro.client", err.Error())
 	}
 	b := &buffer{bytes.NewBuffer(nil)}
-	if err := cf(b).Write(&codec.Message{Type: codec.Publication}, p.Message()); err != nil {
+	if err := cf(b).Write(&codec.Message{Type: codec.Publication}, msg.Payload()); err != nil {
 		return errors.InternalServerError("go.micro.client", err.Error())
 	}
 	r.once.Do(func() {
 		r.opts.Broker.Connect()
 	})
 
-	return r.opts.Broker.Publish(p.Topic(), &broker.Message{
+	return r.opts.Broker.Publish(msg.Topic(), &broker.Message{
 		Header: md,
 		Body:   b.Bytes(),
 	})
 }
 
-func (r *rpcClient) NewPublication(topic string, message interface{}) Publication {
-	return newRpcPublication(topic, message, r.opts.ContentType)
+func (r *rpcClient) NewMessage(topic string, message interface{}) Message {
+	return newMessage(topic, message, r.opts.ContentType)
 }
 
-func (r *rpcClient) NewProtoPublication(topic string, message interface{}) Publication {
-	return newRpcPublication(topic, message, "application/octet-stream")
-}
 func (r *rpcClient) NewRequest(service, method string, request interface{}, reqOpts ...RequestOption) Request {
-	return newRpcRequest(service, method, request, r.opts.ContentType, reqOpts...)
-}
-
-func (r *rpcClient) NewProtoRequest(service, method string, request interface{}, reqOpts ...RequestOption) Request {
-	return newRpcRequest(service, method, request, "application/octet-stream", reqOpts...)
-}
-
-func (r *rpcClient) NewJsonRequest(service, method string, request interface{}, reqOpts ...RequestOption) Request {
-	return newRpcRequest(service, method, request, "application/json", reqOpts...)
+	return newRequest(service, method, request, r.opts.ContentType, reqOpts...)
 }
 
 func (r *rpcClient) String() string {

client/rpc_client_test.go

@@ -1,16 +1,61 @@
 package client
 
 import (
+	"context"
 	"fmt"
 	"testing"
 
 	"github.com/micro/go-micro/registry"
 	"github.com/micro/go-micro/registry/mock"
 	"github.com/micro/go-micro/selector"
-
-	"golang.org/x/net/context"
 )
 
+func TestCallAddress(t *testing.T) {
+	var called bool
+	service := "test.service"
+	method := "Test.Method"
+	address := "10.1.10.1:8080"
+
+	wrap := func(cf CallFunc) CallFunc {
+		return func(ctx context.Context, addr string, req Request, rsp interface{}, opts CallOptions) error {
+			called = true
+
+			if req.Service() != service {
+				return fmt.Errorf("expected service: %s got %s", service, req.Service())
+			}
+
+			if req.Method() != method {
+				return fmt.Errorf("expected service: %s got %s", method, req.Method())
+			}
+
+			if addr != address {
+				return fmt.Errorf("expected address: %s got %s", address, addr)
+			}
+
+			// don't do the call
+			return nil
+		}
+	}
+
+	r := mock.NewRegistry()
+	c := NewClient(
+		Registry(r),
+		WrapCall(wrap),
+	)
+	c.Options().Selector.Init(selector.Registry(r))
+
+	req := c.NewRequest(service, method, nil)
+
+	// test calling remote address
+	if err := c.Call(context.Background(), req, nil, WithAddress(address)); err != nil {
+		t.Fatal("call with address error", err)
+	}
+
+	if !called {
+		t.Fatal("wrapper not called")
+	}
+
+}
 func TestCallWrapper(t *testing.T) {
 	var called bool
 	id := "test.1"

client/rpc_message.go

@@ -0,0 +1,27 @@
+package client
+
+type message struct {
+	topic       string
+	contentType string
+	payload     interface{}
+}
+
+func newMessage(topic string, payload interface{}, contentType string) Message {
+	return &message{
+		payload:     payload,
+		topic:       topic,
+		contentType: contentType,
+	}
+}
+
+func (m *message) ContentType() string {
+	return m.contentType
+}
+
+func (m *message) Topic() string {
+	return m.topic
+}
+
+func (m *message) Payload() interface{} {
+	return m.payload
+}

client/rpc_publication.go

@@ -1,27 +0,0 @@
-package client
-
-type rpcPublication struct {
-	topic       string
-	contentType string
-	message     interface{}
-}
-
-func newRpcPublication(topic string, message interface{}, contentType string) Publication {
-	return &rpcPublication{
-		message:     message,
-		topic:       topic,
-		contentType: contentType,
-	}
-}
-
-func (r *rpcPublication) ContentType() string {
-	return r.contentType
-}
-
-func (r *rpcPublication) Topic() string {
-	return r.topic
-}
-
-func (r *rpcPublication) Message() interface{} {
-	return r.message
-}

client/rpc_request.go

@@ -8,13 +8,18 @@ type rpcRequest struct {
 	opts        RequestOptions
 }
 
-func newRpcRequest(service, method string, request interface{}, contentType string, reqOpts ...RequestOption) Request {
+func newRequest(service, method string, request interface{}, contentType string, reqOpts ...RequestOption) Request {
 	var opts RequestOptions
 
 	for _, o := range reqOpts {
 		o(&opts)
 	}
 
+	// set the content-type specified
+	if len(opts.ContentType) > 0 {
+		contentType = opts.ContentType
+	}
+
 	return &rpcRequest{
 		service:     service,
 		method:      method,

client/rpc_request_test.go

@@ -0,0 +1,23 @@
+package client
+
+import (
+	"testing"
+)
+
+func TestRequestOptions(t *testing.T) {
+	r := newRequest("service", "method", nil, "application/json")
+	if r.Service() != "service" {
+		t.Fatalf("expected 'service' got %s", r.Service())
+	}
+	if r.Method() != "method" {
+		t.Fatalf("expected 'method' got %s", r.Method())
+	}
+	if r.ContentType() != "application/json" {
+		t.Fatalf("expected 'method' got %s", r.ContentType())
+	}
+
+	r2 := newRequest("service", "method", nil, "application/json", WithContentType("application/protobuf"))
+	if r2.ContentType() != "application/protobuf" {
+		t.Fatalf("expected 'method' got %s", r2.ContentType())
+	}
+}

client/rpc_stream.go

@@ -1,10 +1,9 @@
 package client
 
 import (
+	"context"
 	"io"
 	"sync"
-
-	"golang.org/x/net/context"
 )
 
 // Implements the streamer interface
@@ -45,7 +44,6 @@ func (r *rpcStream) Send(msg interface{}) error {
 	}
 
 	seq := r.seq
-	r.seq++
 
 	req := request{
 		Service:       r.request.Service(),

client/wrapper.go

@@ -0,0 +1,17 @@
+package client
+
+import (
+	"context"
+)
+
+// CallFunc represents the individual call func
+type CallFunc func(ctx context.Context, address string, req Request, rsp interface{}, opts CallOptions) error
+
+// CallWrapper is a low level wrapper for the CallFunc
+type CallWrapper func(CallFunc) CallFunc
+
+// Wrapper wraps a client and returns a client
+type Wrapper func(Client) Client
+
+// StreamWrapper wraps a Stream and returns the equivalent
+type StreamWrapper func(Stream) Stream

cmd/cmd.go

@@ -78,6 +78,16 @@ var (
 			EnvVar: "MICRO_CLIENT_POOL_TTL",
 			Usage:  "Sets the client connection pool ttl. e.g 500ms, 5s, 1m. Default: 1m",
 		},
+		cli.IntFlag{
+			Name:   "register_ttl",
+			EnvVar: "MICRO_REGISTER_TTL",
+			Usage:  "Register TTL in seconds",
+		},
+		cli.IntFlag{
+			Name:   "register_interval",
+			EnvVar: "MICRO_REGISTER_INTERVAL",
+			Usage:  "Register interval in seconds",
+		},
 		cli.StringFlag{
 			Name:   "server_name",
 			EnvVar: "MICRO_SERVER_NAME",
@@ -370,6 +380,10 @@ func (c *cmd) Before(ctx *cli.Context) error {
 		serverOpts = append(serverOpts, server.Advertise(ctx.String("server_advertise")))
 	}
 
+	if ttl := time.Duration(ctx.GlobalInt("register_ttl")); ttl > 0 {
+		serverOpts = append(serverOpts, server.RegisterTTL(ttl*time.Second))
+	}
+
 	// client opts
 	if r := ctx.Int("client_retries"); r > 0 {
 		clientOpts = append(clientOpts, client.Retries(r))

cmd/options.go

@@ -1,14 +1,14 @@
 package cmd
 
 import (
+	"context"
+
 	"github.com/micro/go-micro/broker"
 	"github.com/micro/go-micro/client"
 	"github.com/micro/go-micro/registry"
 	"github.com/micro/go-micro/selector"
 	"github.com/micro/go-micro/server"
 	"github.com/micro/go-micro/transport"
-
-	"golang.org/x/net/context"
 )
 
 type Options struct {

function.go

@@ -1,10 +1,10 @@
 package micro
 
 import (
+	"context"
 	"time"
 
 	"github.com/micro/go-micro/server"
-	"golang.org/x/net/context"
 )
 
 type function struct {
@@ -23,7 +23,7 @@ func fnHandlerWrapper(f Function) server.HandlerWrapper {
 
 func fnSubWrapper(f Function) server.SubscriberWrapper {
 	return func(s server.SubscriberFunc) server.SubscriberFunc {
-		return func(ctx context.Context, msg server.Publication) error {
+		return func(ctx context.Context, msg server.Message) error {
 			defer f.Done()
 			return s(ctx, msg)
 		}

function_test.go

@@ -1,13 +1,12 @@
 package micro
 
 import (
+	"context"
 	"sync"
 	"testing"
 
 	"github.com/micro/go-micro/registry/mock"
 	proto "github.com/micro/go-micro/server/debug/proto"
-
-	"golang.org/x/net/context"
 )
 
 func TestFunction(t *testing.T) {

go-micro.go

@@ -2,10 +2,10 @@
 package micro
 
 import (
+	"context"
+
 	"github.com/micro/go-micro/client"
 	"github.com/micro/go-micro/server"
-
-	"golang.org/x/net/context"
 )
 
 type serviceKey struct{}

metadata/metadata.go

@@ -2,7 +2,7 @@
 package metadata
 
 import (
-	"golang.org/x/net/context"
+	"context"
 )
 
 type metaKey struct{}

metadata/metadata_test.go

@@ -1,9 +1,8 @@
 package metadata
 
 import (
+	"context"
 	"testing"
-
-	"golang.org/x/net/context"
 )
 
 func TestMetadataContext(t *testing.T) {

options.go

@@ -1,6 +1,7 @@
 package micro
 
 import (
+	"context"
 	"time"
 
 	"github.com/micro/cli"
@@ -11,8 +12,6 @@ import (
 	"github.com/micro/go-micro/selector"
 	"github.com/micro/go-micro/server"
 	"github.com/micro/go-micro/transport"
-
-	"golang.org/x/net/context"
 )
 
 type Options struct {
@@ -101,6 +100,8 @@ func Registry(r registry.Registry) Option {
 		o.Server.Init(server.Registry(r))
 		// Update Selector
 		o.Client.Options().Selector.Init(selector.Registry(r))
+		// Update Broker
+		o.Broker.Init(broker.Registry(r))
 	}
 }
 

publisher.go

@@ -1,8 +1,9 @@
 package micro
 
 import (
+	"context"
+
 	"github.com/micro/go-micro/client"
-	"golang.org/x/net/context"
 )
 
 type publisher struct {
@@ -11,5 +12,5 @@ type publisher struct {
 }
 
 func (p *publisher) Publish(ctx context.Context, msg interface{}, opts ...client.PublishOption) error {
-	return p.c.Publish(ctx, p.c.NewPublication(p.topic, msg))
+	return p.c.Publish(ctx, p.c.NewMessage(p.topic, msg))
 }

registry/consul/options.go

@@ -1,9 +1,11 @@
 package consul
 
 import (
+	"context"
+	"time"
+
 	consul "github.com/hashicorp/consul/api"
 	"github.com/micro/go-micro/registry"
-	"golang.org/x/net/context"
 )
 
 func Config(c *consul.Config) registry.Option {
@@ -14,3 +16,22 @@ func Config(c *consul.Config) registry.Option {
 		o.Context = context.WithValue(o.Context, "consul_config", c)
 	}
 }
+
+//
+// TCPCheck will tell the service provider to check the service address
+// and port every `t` interval. It will enabled only if `t` is greater than 0.
+// See `TCP + Interval` for more information [1].
+//
+// [1] https://www.consul.io/docs/agent/checks.html
+//
+func TCPCheck(t time.Duration) registry.Option {
+	return func(o *registry.Options) {
+		if t <= time.Duration(0) {
+			return
+		}
+		if o.Context == nil {
+			o.Context = context.Background()
+		}
+		o.Context = context.WithValue(o.Context, "consul_tcp_check", t)
+	}
+}

registry/consul_registry.go

@@ -115,16 +115,39 @@ func (c *consulRegistry) Deregister(s *Service) error {
 	return c.Client.Agent().ServiceDeregister(node.Id)
 }
 
+func getDeregisterTTL(t time.Duration) time.Duration {
+	// splay slightly for the watcher?
+	splay := time.Second * 5
+	deregTTL := t + splay
+
+	// consul has a minimum timeout on deregistration of 1 minute.
+	if t < time.Minute {
+		deregTTL = time.Minute + splay
+	}
+
+	return deregTTL
+}
+
 func (c *consulRegistry) Register(s *Service, opts ...RegisterOption) error {
 	if len(s.Nodes) == 0 {
 		return errors.New("Require at least one node")
 	}
 
+	var regTCPCheck bool
+	var regInterval time.Duration
+
 	var options RegisterOptions
 	for _, o := range opts {
 		o(&options)
 	}
 
+	if c.opts.Context != nil {
+		if tcpCheckInterval, ok := c.opts.Context.Value("consul_tcp_check").(time.Duration); ok {
+			regTCPCheck = true
+			regInterval = tcpCheckInterval
+		}
+	}
+
 	// create hash of service; uint64
 	h, err := hash.Hash(s, nil)
 	if err != nil {
@@ -155,16 +178,19 @@ func (c *consulRegistry) Register(s *Service, opts ...RegisterOption) error {
 
 	var check *consul.AgentServiceCheck
 
-	// if the TTL is greater than 0 create an associated check
-	if options.TTL > time.Duration(0) {
-		// splay slightly for the watcher?
-		splay := time.Second * 5
-		deregTTL := options.TTL + splay
-		// consul has a minimum timeout on deregistration of 1 minute.
-		if options.TTL < time.Minute {
-			deregTTL = time.Minute + splay
+	if regTCPCheck {
+		deregTTL := getDeregisterTTL(regInterval)
+
+		check = &consul.AgentServiceCheck{
+			TCP:                            fmt.Sprintf("%s:%d", node.Address, node.Port),
+			Interval:                       fmt.Sprintf("%v", regInterval),
+			DeregisterCriticalServiceAfter: fmt.Sprintf("%v", deregTTL),
 		}
 
+		// if the TTL is greater than 0 create an associated check
+	} else if options.TTL > time.Duration(0) {
+		deregTTL := getDeregisterTTL(options.TTL)
+
 		check = &consul.AgentServiceCheck{
 			TTL: fmt.Sprintf("%v", options.TTL),
 			DeregisterCriticalServiceAfter: fmt.Sprintf("%v", deregTTL),
@@ -211,18 +237,18 @@ func (c *consulRegistry) GetService(name string) ([]*Service, error) {
 		}
 
 		// version is now a tag
-		version, found := decodeVersion(s.Service.Tags)
+		version, _ := decodeVersion(s.Service.Tags)
 		// service ID is now the node id
 		id := s.Service.ID
 		// key is always the version
 		key := version
+
 		// address is service address
 		address := s.Service.Address
 
-		// if we can't get the version we bail
-		// use old the old ways
-		if !found {
-			continue
+		// use node address
+		if len(address) == 0 {
+			address = s.Node.Address
 		}
 
 		svc, ok := serviceMap[key]
@@ -236,6 +262,7 @@ func (c *consulRegistry) GetService(name string) ([]*Service, error) {
 		}
 
 		var del bool
+
 		for _, check := range s.Checks {
 			// delete the node if the status is critical
 			if check.Status == "critical" {

registry/consul_watcher.go

@@ -60,7 +60,7 @@ func (cw *consulWatcher) serviceHandler(idx uint64, data interface{}) {
 	for _, e := range entries {
 		serviceName = e.Service.Service
 		// version is now a tag
-		version, found := decodeVersion(e.Service.Tags)
+		version, _ := decodeVersion(e.Service.Tags)
 		// service ID is now the node id
 		id := e.Service.ID
 		// key is always the version
@@ -68,9 +68,9 @@ func (cw *consulWatcher) serviceHandler(idx uint64, data interface{}) {
 		// address is service address
 		address := e.Service.Address
 
-		// if we can't get the version we bail
-		if !found {
-			continue
+		// use node address
+		if len(address) == 0 {
+			address = e.Node.Address
 		}
 
 		svc, ok := serviceMap[key]

registry/options.go

@@ -1,10 +1,9 @@
 package registry
 
 import (
+	"context"
 	"crypto/tls"
 	"time"
-
-	"golang.org/x/net/context"
 )
 
 type Options struct {

selector/cache/options.go

@@ -1,10 +1,10 @@
 package cache
 
 import (
+	"context"
 	"time"
 
 	"github.com/micro/go-micro/selector"
-	"golang.org/x/net/context"
 )
 
 type ttlKey struct{}

selector/options.go

@@ -1,9 +1,9 @@
 package selector
 
 import (
-	"github.com/micro/go-micro/registry"
+	"context"
 
-	"golang.org/x/net/context"
+	"github.com/micro/go-micro/registry"
 )
 
 type Options struct {

server/context.go

@@ -1,7 +1,7 @@
 package server
 
 import (
-	"golang.org/x/net/context"
+	"context"
 )
 
 type serverKey struct{}

server/debug/debug.go

@@ -1,12 +1,11 @@
 package debug
 
 import (
+	"context"
 	"runtime"
 	"time"
 
 	proto "github.com/micro/go-micro/server/debug/proto"
-
-	"golang.org/x/net/context"
 )
 
 // The debug handler represents an internal server handler

server/extractor_test.go

@@ -1,11 +1,11 @@
 package server
 
 import (
+	"context"
 	"reflect"
 	"testing"
 
 	"github.com/micro/go-micro/registry"
-	"golang.org/x/net/context"
 )
 
 type testHandler struct{}

server/options.go

@@ -1,6 +1,7 @@
 package server
 
 import (
+	"context"
 	"time"
 
 	"github.com/micro/go-micro/broker"
@@ -8,8 +9,6 @@ import (
 	"github.com/micro/go-micro/registry"
 	"github.com/micro/go-micro/server/debug"
 	"github.com/micro/go-micro/transport"
-
-	"golang.org/x/net/context"
 )
 
 type Options struct {

server/rpc_request.go

@@ -8,10 +8,10 @@ type rpcRequest struct {
 	stream      bool
 }
 
-type rpcPublication struct {
+type rpcMessage struct {
 	topic       string
 	contentType string
-	message     interface{}
+	payload     interface{}
 }
 
 func (r *rpcRequest) ContentType() string {
@@ -34,14 +34,14 @@ func (r *rpcRequest) Stream() bool {
 	return r.stream
 }
 
-func (r *rpcPublication) ContentType() string {
+func (r *rpcMessage) ContentType() string {
 	return r.contentType
 }
 
-func (r *rpcPublication) Topic() string {
+func (r *rpcMessage) Topic() string {
 	return r.topic
 }
 
-func (r *rpcPublication) Message() interface{} {
-	return r.message
+func (r *rpcMessage) Payload() interface{} {
+	return r.payload
 }

server/rpc_server.go

@@ -1,6 +1,7 @@
 package server
 
 import (
+	"context"
 	"fmt"
 	"runtime/debug"
 	"sort"
@@ -16,9 +17,7 @@ import (
 	"github.com/micro/go-micro/registry"
 	"github.com/micro/go-micro/transport"
 
-	"github.com/micro/misc/lib/addr"
-
-	"golang.org/x/net/context"
+	"github.com/micro/util/go/lib/addr"
 )
 
 type rpcServer struct {

server/rpc_service.go

@@ -7,6 +7,7 @@ package server
 // Meh, we need to get rid of this shit
 
 import (
+	"context"
 	"errors"
 	"io"
 	"reflect"
@@ -16,7 +17,6 @@ import (
 	"unicode/utf8"
 
 	"github.com/micro/go-log"
-	"golang.org/x/net/context"
 )
 
 var (
@@ -119,9 +119,9 @@ func prepareMethod(method reflect.Method) *methodType {
 
 	if stream {
 		// check stream type
-		streamType := reflect.TypeOf((*Streamer)(nil)).Elem()
+		streamType := reflect.TypeOf((*Stream)(nil)).Elem()
 		if !argType.Implements(streamType) {
-			log.Log(mname, "argument does not implement Streamer interface:", argType)
+			log.Log(mname, "argument does not implement Stream interface:", argType)
 			return nil
 		}
 	} else {

server/rpc_stream.go

@@ -1,9 +1,8 @@
 package server
 
 import (
+	"context"
 	"sync"
-
-	"golang.org/x/net/context"
 )
 
 // Implements the Streamer interface

server/server.go

@@ -2,13 +2,13 @@
 package server
 
 import (
+	"context"
 	"os"
 	"os/signal"
 	"syscall"
 
 	"github.com/micro/go-log"
 	"github.com/pborman/uuid"
-	"golang.org/x/net/context"
 )
 
 type Server interface {
@@ -25,9 +25,9 @@ type Server interface {
 	String() string
 }
 
-type Publication interface {
+type Message interface {
 	Topic() string
-	Message() interface{}
+	Payload() interface{}
 	ContentType() string
 }
 
@@ -40,11 +40,11 @@ type Request interface {
 	Stream() bool
 }
 
-// Streamer represents a stream established with a client.
+// Stream represents a stream established with a client.
 // A stream can be bidirectional which is indicated by the request.
 // The last error will be left in Error().
 // EOF indicated end of the stream.
-type Streamer interface {
+type Stream interface {
 	Context() context.Context
 	Request() Request
 	Send(interface{}) error

server/subscriber.go

@@ -2,6 +2,7 @@ package server
 
 import (
 	"bytes"
+	"context"
 	"fmt"
 	"reflect"
 
@@ -9,7 +10,6 @@ import (
 	"github.com/micro/go-micro/codec"
 	"github.com/micro/go-micro/metadata"
 	"github.com/micro/go-micro/registry"
-	"golang.org/x/net/context"
 )
 
 const (
@@ -204,7 +204,7 @@ func (s *rpcServer) createSubHandler(sb *subscriber, opts Options) broker.Handle
 				return err
 			}
 
-			fn := func(ctx context.Context, msg Publication) error {
+			fn := func(ctx context.Context, msg Message) error {
 				var vals []reflect.Value
 				if sb.typ.Kind() != reflect.Func {
 					vals = append(vals, sb.rcvr)
@@ -213,7 +213,7 @@ func (s *rpcServer) createSubHandler(sb *subscriber, opts Options) broker.Handle
 					vals = append(vals, reflect.ValueOf(ctx))
 				}
 
-				vals = append(vals, reflect.ValueOf(msg.Message()))
+				vals = append(vals, reflect.ValueOf(msg.Payload()))
 
 				returnValues := handler.method.Call(vals)
 				if err := returnValues[0].Interface(); err != nil {
@@ -229,10 +229,10 @@ func (s *rpcServer) createSubHandler(sb *subscriber, opts Options) broker.Handle
 			s.wg.Add(1)
 			go func() {
 				defer s.wg.Done()
-				fn(ctx, &rpcPublication{
+				fn(ctx, &rpcMessage{
 					topic:       sb.topic,
 					contentType: ct,
-					message:     req.Interface(),
+					payload:     req.Interface(),
 				})
 			}()
 		}

server/wrapper.go

@@ -1,7 +1,7 @@
 package server
 
 import (
-	"golang.org/x/net/context"
+	"context"
 )
 
 // HandlerFunc represents a single method of a handler. It's used primarily
@@ -12,7 +12,7 @@ type HandlerFunc func(ctx context.Context, req Request, rsp interface{}) error
 // SubscriberFunc represents a single method of a subscriber. It's used primarily
 // for the wrappers. What's handed to the actual method is the concrete
 // publication message.
-type SubscriberFunc func(ctx context.Context, msg Publication) error
+type SubscriberFunc func(ctx context.Context, msg Message) error
 
 // HandlerWrapper wraps the HandlerFunc and returns the equivalent
 type HandlerWrapper func(HandlerFunc) HandlerFunc
@@ -20,8 +20,8 @@ type HandlerWrapper func(HandlerFunc) HandlerFunc
 // SubscriberWrapper wraps the SubscriberFunc and returns the equivalent
 type SubscriberWrapper func(SubscriberFunc) SubscriberFunc
 
-// StreamerWrapper wraps a Streamer interface and returns the equivalent.
+// StreamWrapper wraps a Stream interface and returns the equivalent.
 // Because streams exist for the lifetime of a method invocation this
 // is a convenient way to wrap a Stream as its in use for trace, monitoring,
 // metrics, etc.
-type StreamerWrapper func(Streamer) Streamer
+type StreamWrapper func(Stream) Stream

service.go

@@ -7,7 +7,8 @@ import (
 	"syscall"
 	"time"
 
-	log "github.com/micro/go-log"
+	"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"
@@ -66,8 +67,22 @@ 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(
+		_ = s.opts.Cmd.Init(
 			cmd.Broker(&s.opts.Broker),
 			cmd.Registry(&s.opts.Registry),
 			cmd.Transport(&s.opts.Transport),
@@ -153,7 +168,7 @@ func (s *service) Run() error {
 	go s.run(ex)
 
 	ch := make(chan os.Signal, 1)
-	signal.Notify(ch, syscall.SIGTERM, syscall.SIGINT, syscall.SIGKILL)
+	signal.Notify(ch, syscall.SIGTERM, syscall.SIGINT, syscall.SIGQUIT)
 
 	select {
 	// wait on kill signal
@@ -165,9 +180,5 @@ func (s *service) Run() error {
 	// exit reg loop
 	close(ex)
 
-	if err := s.Stop(); err != nil {
-		return err
-	}
-
-	return nil
+	return s.Stop()
 }

service_test.go

@@ -1,13 +1,12 @@
 package micro
 
 import (
+	"context"
 	"sync"
 	"testing"
 
 	"github.com/micro/go-micro/registry/mock"
 	proto "github.com/micro/go-micro/server/debug/proto"
-
-	"golang.org/x/net/context"
 )
 
 func TestService(t *testing.T) {
@@ -31,35 +30,30 @@ func TestService(t *testing.T) {
 	// we can't test service.Init as it parses the command line
 	// service.Init()
 
-	// register handler
-	// do that later
-
-	go func() {
-		// wait for start
-		wg.Wait()
-
-		// test call debug
-		req := service.Client().NewRequest(
-			"test.service",
-			"Debug.Health",
-			new(proto.HealthRequest),
-		)
-
-		rsp := new(proto.HealthResponse)
-
-		err := service.Client().Call(context.TODO(), req, rsp)
-		if err != nil {
-			t.Fatal(err)
-		}
-
-		if rsp.Status != "ok" {
-			t.Fatalf("service response: %s", rsp.Status)
-		}
-
-		// shutdown the service
-		cancel()
-	}()
-
 	// run service
-	service.Run()
+	go service.Run()
+
+	// wait for start
+	wg.Wait()
+
+	// test call debug
+	req := service.Client().NewRequest(
+		"test.service",
+		"Debug.Health",
+		new(proto.HealthRequest),
+	)
+
+	rsp := new(proto.HealthResponse)
+
+	err := service.Client().Call(context.TODO(), req, rsp)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if rsp.Status != "ok" {
+		t.Fatalf("service response: %s", rsp.Status)
+	}
+
+	// shutdown the service
+	cancel()
 }

transport/http_transport.go

@@ -14,9 +14,9 @@ import (
 	"time"
 
 	"github.com/micro/go-log"
-	maddr "github.com/micro/misc/lib/addr"
-	mnet "github.com/micro/misc/lib/net"
-	mls "github.com/micro/misc/lib/tls"
+	maddr "github.com/micro/util/go/lib/addr"
+	mnet "github.com/micro/util/go/lib/net"
+	mls "github.com/micro/util/go/lib/tls"
 )
 
 type buffer struct {

transport/options.go

@@ -1,11 +1,11 @@
 package transport
 
 import (
+	"context"
 	"crypto/tls"
 	"time"
 
 	"github.com/micro/go-micro/transport/codec"
-	"golang.org/x/net/context"
 )
 
 type Options struct {

transport/transport.go

@@ -50,15 +50,3 @@ var (
 func NewTransport(opts ...Option) Transport {
 	return newHTTPTransport(opts...)
 }
-
-func Dial(addr string, opts ...DialOption) (Client, error) {
-	return DefaultTransport.Dial(addr, opts...)
-}
-
-func Listen(addr string, opts ...ListenOption) (Listener, error) {
-	return DefaultTransport.Listen(addr, opts...)
-}
-
-func String() string {
-	return DefaultTransport.String()
-}

wrapper.go

@@ -1,10 +1,10 @@
 package micro
 
 import (
+	"context"
+
 	"github.com/micro/go-micro/client"
 	"github.com/micro/go-micro/metadata"
-
-	"golang.org/x/net/context"
 )
 
 type clientWrapper struct {
@@ -36,12 +36,12 @@ func (c *clientWrapper) Call(ctx context.Context, req client.Request, rsp interf
 	return c.Client.Call(ctx, req, rsp, opts...)
 }
 
-func (c *clientWrapper) Stream(ctx context.Context, req client.Request, opts ...client.CallOption) (client.Streamer, error) {
+func (c *clientWrapper) Stream(ctx context.Context, req client.Request, opts ...client.CallOption) (client.Stream, error) {
 	ctx = c.setHeaders(ctx)
 	return c.Client.Stream(ctx, req, opts...)
 }
 
-func (c *clientWrapper) Publish(ctx context.Context, p client.Publication, opts ...client.PublishOption) error {
+func (c *clientWrapper) Publish(ctx context.Context, p client.Message, opts ...client.PublishOption) error {
 	ctx = c.setHeaders(ctx)
 	return c.Client.Publish(ctx, p, opts...)
 }

wrapper_test.go

@@ -1,11 +1,10 @@
 package micro
 
 import (
+	"context"
 	"testing"
 
 	"github.com/micro/go-micro/metadata"
-
-	"golang.org/x/net/context"
 )
 
 func TestWrapper(t *testing.T) {