Update runtime.go

* subs not filtered correctly on publish. simplify retry logic

* check fo invalid ackwait

.github/FUNDING.yml

@@ -1,3 +1,3 @@
 # These are supported funding model platforms
 
-github: asim
+github: micro

.github/PULL_REQUEST_TEMPLATE.md

@@ -1,10 +1,9 @@
 ## Pull Request template
 Please, go through these steps before clicking submit on this PR.
 
-1. Make sure this PR targets the `develop` branch. We follow the git-flow branching model.
-2. Give a descriptive title to your PR.
-3. Provide a description of your changes. 
-4. Make sure you have some relevant tests.
-5. Put `closes #XXXX` in your comment to auto-close the issue that your PR fixes (if applicable).
+1. Give a descriptive title to your PR.
+2. Provide a description of your changes.
+3. Make sure you have some relevant tests.
+4. Put `closes #XXXX` in your comment to auto-close the issue that your PR fixes (if applicable).
 
 **PLEASE REMOVE THIS TEMPLATE BEFORE SUBMITTING**

.github/workflows/docker.yml

@@ -0,0 +1,22 @@
+name: Docker
+
+on:
+  push:
+    branches:
+    - master
+    tags:
+    - v2.*
+    - v3.*
+jobs:
+  build:
+     runs-on: ubuntu-latest
+     steps:
+       - uses: actions/checkout@v2
+         name: Check out repository
+       - uses: elgohr/Publish-Docker-Github-Action@2.12
+         name: Build and Push Docker Image
+         with:
+           name: micro/go-micro
+           username: ${{ secrets.DOCKER_USERNAME }}
+           password: ${{ secrets.DOCKER_PASSWORD }}
+           tag_names: true

.github/workflows/pr.yml

@@ -8,10 +8,10 @@ jobs:
     runs-on: ubuntu-latest
     steps:
 
-    - name: Set up Go 1.15
+    - name: Set up Go 1.13
       uses: actions/setup-go@v1
       with:
-        go-version: 1.15
+        go-version: 1.13
       id: go
 
     - name: Check out code into the Go module directory
@@ -25,4 +25,7 @@ jobs:
       id: tests
       env:
         IN_TRAVIS_CI: yes
-      run: go test -v ./...
+      run: |
+        wget -qO- https://binaries.cockroachdb.com/cockroach-v20.1.4.linux-amd64.tgz | tar  xvz
+        cockroach-v20.1.4.linux-amd64/cockroach start-single-node --insecure &
+        go test -v ./...

.github/workflows/scripts/build-all-examples.sh

@@ -0,0 +1,41 @@
+#!/bin/bash
+# set -x
+
+function build_binary {
+    echo building $1
+    pushd $1
+    go build -o _main
+    local ret=$?
+    if [ $ret -gt 0 ]; then 
+        failed=1
+        failed_arr+=($1)
+    fi
+    popd
+}
+
+function is_main {
+    grep "package main" -l -dskip $1/*.go > /dev/null 2>&1
+}
+
+
+function check_dir {
+    is_main $1
+    local ret=$?
+    if [ $ret == 0 ]; then
+        build_binary $1 $2
+    fi
+    for filename in $1/*; do
+        if [ -d $filename ]; then
+            check_dir $filename $2
+        fi
+    done
+}
+failed_arr=()
+failed=0
+go mod edit -replace github.com/micro/go-micro/v2=github.com/$2/v2@$1 
+check_dir . $1
+if [ $failed -gt 0 ]; then
+    echo Some builds failed
+    printf '%s\n' "${failed_arr[@]}"
+fi
+exit $failed

.github/workflows/scripts/build-micro.sh

@@ -0,0 +1,19 @@
+#!/bin/bash
+# set -x
+
+failed=0
+go mod edit -replace github.com/micro/go-micro/v2=github.com/$2/v2@$1 
+# basic test, build the binary
+go install
+failed=$?
+if [ $failed -gt 0 ]; then
+    exit $failed
+fi
+# unit tests
+IN_TRAVIS_CI=yes go test -v ./...
+
+./scripts/test-docker.sh
+# Generate keys for JWT tests
+ssh-keygen -f /tmp/sshkey -m pkcs8 -q -N ""
+ssh-keygen -f /tmp/sshkey -e  -m pkcs8 > /tmp/sshkey.pub
+go clean -testcache && IN_TRAVIS_CI=yes go test --tags=integration -v ./test

.github/workflows/tests.yml

@@ -7,13 +7,17 @@ jobs:
     name: Test repo
     runs-on: ubuntu-latest
     steps:
-
-    - name: Set up Go 1.15
+    - name: Set up Go 1.13
       uses: actions/setup-go@v1
       with:
-        go-version: 1.15
+        go-version: 1.13
       id: go
 
+    - name: Setup Kind
+      uses: engineerd/setup-kind@v0.4.0
+      with:
+        version: v0.8.1
+
     - name: Check out code into the Go module directory
       uses: actions/checkout@v2
 
@@ -25,5 +29,37 @@ jobs:
       id: tests
       env:
         IN_TRAVIS_CI: yes
-      run: go test -v ./...
+      run: |
+        kubectl apply -f runtime/kubernetes/test/test.yaml
+        sudo mkdir -p /var/run/secrets/kubernetes.io/serviceaccount
+        sudo chmod 777 /var/run/secrets/kubernetes.io/serviceaccount
+        wget -qO- https://binaries.cockroachdb.com/cockroach-v20.1.4.linux-amd64.tgz | tar -xvz
+        cockroach-v20.1.4.linux-amd64/cockroach start-single-node --insecure &
+        wget -q https://github.com/nats-io/nats-streaming-server/releases/download/v0.18.0/nats-streaming-server-v0.18.0-linux-amd64.zip
+        unzip ./nats-streaming-server-v0.18.0-linux-amd64.zip
+        export PATH=$PATH:./nats-streaming-server-v0.18.0-linux-amd64
+        nats-streaming-server &
+        go test -tags kubernetes,nats -v ./...
+  
+    - name: Notify of test failure
+      if: failure()
+      uses: rtCamp/action-slack-notify@v2.0.0
+      env:
+        SLACK_CHANNEL: build
+        SLACK_COLOR: '#BF280A'
+        SLACK_ICON: https://github.githubassets.com/images/modules/logos_page/GitHub-Mark.png
+        SLACK_TITLE: Tests Failed
+        SLACK_USERNAME: GitHub Actions
+        SLACK_WEBHOOK: ${{ secrets.SLACK_WEBHOOK_URL }}
+        
+    - name: Notify of test success
+      if: success()
+      uses: rtCamp/action-slack-notify@v2.0.0
+      env:
+        SLACK_CHANNEL: build
+        SLACK_COLOR: '#1FAD2B'
+        SLACK_ICON: https://github.githubassets.com/images/modules/logos_page/GitHub-Mark.png
+        SLACK_TITLE: Tests Passed
+        SLACK_USERNAME: GitHub Actions
+        SLACK_WEBHOOK: ${{ secrets.SLACK_WEBHOOK_URL }}
 

CNAME

@@ -0,0 +1 @@
+go-micro.dev

Dockerfile

@@ -0,0 +1,13 @@
+FROM golang:1.13-alpine
+
+RUN mkdir /user && \
+    echo 'nobody:x:65534:65534:nobody:/:' > /user/passwd && \
+    echo 'nobody:x:65534:' > /user/group
+
+ENV GO111MODULE=on
+RUN apk --no-cache add make git gcc libtool musl-dev ca-certificates dumb-init && \
+    rm -rf /var/cache/apk/* /tmp/*
+
+WORKDIR /
+COPY ./go.mod ./go.sum ./
+RUN go mod download && rm go.mod go.sum

README.md

@@ -1,6 +1,6 @@
-# Go Micro [![License](https://img.shields.io/:license-apache-blue.svg)](https://opensource.org/licenses/Apache-2.0) [![Go.Dev reference](https://img.shields.io/badge/go.dev-reference-007d9c?logo=go&logoColor=white&style=flat-square)](https://pkg.go.dev/github.com/asim/go-micro/v3?tab=doc)
+# Go Micro [![License](https://img.shields.io/:license-apache-blue.svg)](https://opensource.org/licenses/Apache-2.0) [![Go.Dev reference](https://img.shields.io/badge/go.dev-reference-007d9c?logo=go&logoColor=white&style=flat-square)](https://pkg.go.dev/github.com/micro/go-micro/v3?tab=overview) [![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 framework for distributed systems development.
+Go Micro is a standard library for microservices.
 
 ## Overview
 
@@ -34,8 +34,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.
 
-- **RPC Client/Server** - 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.
+- **gRPC Transport** - gRPC 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.
 
 - **Async Messaging** - PubSub is built in as a first class citizen for asynchronous communication and event driven architectures. 
 Event notifications are a core pattern in micro service development. The default messaging system is a HTTP event message broker.
@@ -44,49 +43,14 @@ Event notifications are a core pattern in micro service development. The default
 leadership are built in as a Sync interface. When using an eventually consistent database or scheduling use the Sync interface.
 
 - **Pluggable Interfaces** - Go Micro makes use of Go interfaces for each distributed system abstraction. Because of this these interfaces 
-are pluggable and allows Go Micro to be runtime agnostic. You can plugin any underlying technology.
+are pluggable and allows Go Micro to be runtime agnostic. You can plugin any underlying technology. Find plugins in 
+[github.com/micro/go-plugins](https://github.com/micro/go-plugins).
 
 ## Getting Started
 
-To make use of Go Micro
-
-```golang
-import "github.com/asim/go-micro/v3"
-
-// create a new service
-service := micro.NewService(
-    micro.Name("helloworld"),
-)
-
-// initialise flags
-service.Init()
-
-// start the service
-service.Run()
-```
-
-See the [examples](https://github.com/micro/go-micro/tree/master/examples) for detailed information on usage.
-
-## Command Line Interface
-
-See [cmd/gomu](https://github.com/asim/go-micro/tree/master/cmd/gomu) for the command line interface.
-
-## Code Generation
-
-See [cmd/protoc-gen-micro](https://github.com/micro/go-micro/tree/master/cmd/protoc-gen-micro) for protobuf code generation.
-
-## Example Usage
-
-See [examples](https://github.com/micro/go-micro/tree/master/examples) directory for usage examples.
-
-## Plugins
-
-See [plugins](https://github.com/micro/go-micro/tree/master/plugins) directory for all the plugins.
-
-## Services
-
-See [services](https://github.com/micro/go-micro/tree/master/services) directory for third party services.
+See [pkg.go.dev](https://pkg.go.dev/github.com/micro/go-micro/v3?tab=overview) for usage.
 
 ## License
 
 Go Micro is Apache 2.0 licensed.
+

api/api.go

@@ -5,8 +5,8 @@ import (
 	"regexp"
 	"strings"
 
-	"github.com/asim/go-micro/v3/registry"
-	"github.com/asim/go-micro/v3/server"
+	"github.com/micro/go-micro/v3/registry"
+	"github.com/micro/go-micro/v3/server"
 )
 
 type Api interface {
@@ -18,7 +18,7 @@ type Api interface {
 	Register(*Endpoint) error
 	// Register a route
 	Deregister(*Endpoint) error
-	// Implemenation of api
+	// Implementation of api
 	String() string
 }
 

api/handler/api/api.go

@@ -4,13 +4,13 @@ package api
 import (
 	"net/http"
 
-	goapi "github.com/asim/go-micro/v3/api"
-	"github.com/asim/go-micro/v3/api/handler"
-	api "github.com/asim/go-micro/v3/api/proto"
-	"github.com/asim/go-micro/v3/client"
-	"github.com/asim/go-micro/v3/errors"
-	"github.com/asim/go-micro/v3/selector"
-	"github.com/asim/go-micro/v3/util/ctx"
+	goapi "github.com/micro/go-micro/v3/api"
+	"github.com/micro/go-micro/v3/api/handler"
+	api "github.com/micro/go-micro/v3/api/proto"
+	"github.com/micro/go-micro/v3/client"
+	"github.com/micro/go-micro/v3/errors"
+	"github.com/micro/go-micro/v3/util/ctx"
+	"github.com/micro/go-micro/v3/util/router"
 )
 
 type apiHandler struct {
@@ -71,10 +71,8 @@ func (a *apiHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 
 	// create the context from headers
 	cx := ctx.FromRequest(r)
-	// create strategy
-	so := selector.WithStrategy(strategy(service.Services))
 
-	if err := c.Call(cx, req, rsp, client.WithSelectOption(so)); err != nil {
+	if err := c.Call(cx, req, rsp, client.WithRouter(router.New(service.Services))); err != nil {
 		w.Header().Set("Content-Type", "application/json")
 		ce := errors.Parse(err.Error())
 		switch ce.Code {

api/handler/api/util.go

@@ -7,9 +7,7 @@ import (
 	"net/http"
 	"strings"
 
-	api "github.com/asim/go-micro/v3/api/proto"
-	"github.com/asim/go-micro/v3/registry"
-	"github.com/asim/go-micro/v3/selector"
+	api "github.com/micro/go-micro/v3/api/proto"
 	"github.com/oxtoacart/bpool"
 )
 
@@ -109,11 +107,3 @@ func requestToProto(r *http.Request) (*api.Request, error) {
 
 	return req, nil
 }
-
-// strategy is a hack for selection
-func strategy(services []*registry.Service) selector.Strategy {
-	return func(_ []*registry.Service) selector.Next {
-		// ignore input to this function, use services above
-		return selector.Random(services)
-	}
-}

api/handler/event/event.go

@@ -10,10 +10,10 @@ import (
 	"strings"
 	"time"
 
-	"github.com/asim/go-micro/v3/api/handler"
-	proto "github.com/asim/go-micro/v3/api/proto"
-	"github.com/asim/go-micro/v3/util/ctx"
 	"github.com/google/uuid"
+	"github.com/micro/go-micro/v3/api/handler"
+	proto "github.com/micro/go-micro/v3/api/proto"
+	"github.com/micro/go-micro/v3/util/ctx"
 	"github.com/oxtoacart/bpool"
 )
 

api/handler/http/http.go

@@ -4,13 +4,14 @@ package http
 import (
 	"errors"
 	"fmt"
+	"math/rand"
 	"net/http"
 	"net/http/httputil"
 	"net/url"
 
-	"github.com/asim/go-micro/v3/api"
-	"github.com/asim/go-micro/v3/api/handler"
-	"github.com/asim/go-micro/v3/selector"
+	"github.com/micro/go-micro/v3/api"
+	"github.com/micro/go-micro/v3/api/handler"
+	"github.com/micro/go-micro/v3/registry"
 )
 
 const (
@@ -20,7 +21,7 @@ const (
 type httpHandler struct {
 	options handler.Options
 
-	// set with different initialiser
+	// set with different initializer
 	s *api.Service
 }
 
@@ -64,16 +65,19 @@ func (h *httpHandler) getService(r *http.Request) (string, error) {
 		return "", errors.New("no route found")
 	}
 
-	// create a random selector
-	next := selector.Random(service.Services)
-
-	// get the next node
-	s, err := next()
-	if err != nil {
-		return "", nil
+	// get the nodes for this service
+	var nodes []*registry.Node
+	for _, srv := range service.Services {
+		nodes = append(nodes, srv.Nodes...)
 	}
 
-	return fmt.Sprintf("http://%s", s.Address), nil
+	// select a random node
+	if len(nodes) == 0 {
+		return "", errors.New("no route found")
+	}
+	node := nodes[rand.Int()%len(nodes)]
+
+	return fmt.Sprintf("http://%s", node.Address), nil
 }
 
 func (h *httpHandler) String() string {

api/handler/http/http_test.go

@@ -6,16 +6,17 @@ import (
 	"net/http/httptest"
 	"testing"
 
-	"github.com/asim/go-micro/v3/api/handler"
-	"github.com/asim/go-micro/v3/api/resolver"
-	"github.com/asim/go-micro/v3/api/resolver/vpath"
-	"github.com/asim/go-micro/v3/api/router"
-	regRouter "github.com/asim/go-micro/v3/api/router/registry"
-	"github.com/asim/go-micro/v3/registry"
+	"github.com/micro/go-micro/v3/api/handler"
+	"github.com/micro/go-micro/v3/api/resolver"
+	"github.com/micro/go-micro/v3/api/resolver/vpath"
+	"github.com/micro/go-micro/v3/api/router"
+	regRouter "github.com/micro/go-micro/v3/api/router/registry"
+	"github.com/micro/go-micro/v3/registry"
+	"github.com/micro/go-micro/v3/registry/memory"
 )
 
 func testHttp(t *testing.T, path, service, ns string) {
-	r := registry.NewMemoryRegistry()
+	r := memory.NewRegistry()
 
 	l, err := net.Listen("tcp", "127.0.0.1:0")
 	if err != nil {
@@ -57,7 +58,7 @@ func testHttp(t *testing.T, path, service, ns string) {
 		router.WithHandler("http"),
 		router.WithRegistry(r),
 		router.WithResolver(vpath.NewResolver(
-			resolver.WithNamespace(resolver.StaticNamespace(ns)),
+			resolver.WithServicePrefix(ns),
 		)),
 	)
 

api/handler/options.go

@@ -1,8 +1,9 @@
 package handler
 
 import (
-	"github.com/asim/go-micro/v3/api/router"
-	"github.com/asim/go-micro/v3/client"
+	"github.com/micro/go-micro/v3/api/router"
+	"github.com/micro/go-micro/v3/client"
+	"github.com/micro/go-micro/v3/client/grpc"
 )
 
 var (
@@ -26,7 +27,7 @@ func NewOptions(opts ...Option) Options {
 	}
 
 	if options.Client == nil {
-		WithClient(client.NewClient())(&options)
+		WithClient(grpc.NewClient())(&options)
 	}
 
 	// set namespace if blank
@@ -61,7 +62,7 @@ func WithClient(c client.Client) Option {
 	}
 }
 
-// WithmaxRecvSize specifies max body size
+// WithMaxRecvSize specifies max body size
 func WithMaxRecvSize(size int64) Option {
 	return func(o *Options) {
 		o.MaxRecvSize = size

api/handler/rpc/rpc.go

@@ -5,25 +5,23 @@ import (
 	"encoding/json"
 	"io"
 	"net/http"
-	"net/textproto"
 	"strconv"
 	"strings"
 
-	"github.com/asim/go-micro/v3/api"
-	"github.com/asim/go-micro/v3/api/handler"
-	"github.com/asim/go-micro/v3/api/internal/proto"
-	"github.com/asim/go-micro/v3/client"
-	"github.com/asim/go-micro/v3/codec"
-	"github.com/asim/go-micro/v3/codec/jsonrpc"
-	"github.com/asim/go-micro/v3/codec/protorpc"
-	"github.com/asim/go-micro/v3/errors"
-	"github.com/asim/go-micro/v3/logger"
-	"github.com/asim/go-micro/v3/metadata"
-	"github.com/asim/go-micro/v3/registry"
-	"github.com/asim/go-micro/v3/selector"
-	"github.com/asim/go-micro/v3/util/ctx"
-	"github.com/asim/go-micro/v3/util/qson"
 	jsonpatch "github.com/evanphx/json-patch/v5"
+	"github.com/micro/go-micro/v3/api"
+	"github.com/micro/go-micro/v3/api/handler"
+	"github.com/micro/go-micro/v3/api/internal/proto"
+	"github.com/micro/go-micro/v3/client"
+	"github.com/micro/go-micro/v3/codec"
+	"github.com/micro/go-micro/v3/codec/jsonrpc"
+	"github.com/micro/go-micro/v3/codec/protorpc"
+	"github.com/micro/go-micro/v3/errors"
+	"github.com/micro/go-micro/v3/logger"
+	"github.com/micro/go-micro/v3/metadata"
+	"github.com/micro/go-micro/v3/util/ctx"
+	"github.com/micro/go-micro/v3/util/qson"
+	"github.com/micro/go-micro/v3/util/router"
 	"github.com/oxtoacart/bpool"
 )
 
@@ -65,14 +63,6 @@ func (b *buffer) Write(_ []byte) (int, error) {
 	return 0, nil
 }
 
-// strategy is a hack for selection
-func strategy(services []*registry.Service) selector.Strategy {
-	return func(_ []*registry.Service) selector.Next {
-		// ignore input to this function, use services above
-		return selector.Random(services)
-	}
-}
-
 func (h *rpcHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	bsize := handler.DefaultMaxRecvSize
 	if h.opts.MaxRecvSize > 0 {
@@ -113,36 +103,17 @@ func (h *rpcHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 
 	// create context
 	cx := ctx.FromRequest(r)
-	// get context from http handler wrappers
-	md, ok := metadata.FromContext(r.Context())
-	if !ok {
-		md = make(metadata.Metadata)
-	}
-	// fill contex with http headers
-	md["Host"] = r.Host
-	md["Method"] = r.Method
-	// get canonical headers
-	for k := range r.Header {
-		// may be need to get all values for key like r.Header.Values() provide in go 1.14
-		md[textproto.CanonicalMIMEHeaderKey(k)] = r.Header.Get(k)
-	}
-
-	// merge context with overwrite
-	cx = metadata.MergeContext(cx, md, true)
 
 	// set merged context to request
 	*r = *r.Clone(cx)
 	// if stream we currently only support json
 	if isStream(r, service) {
-		// drop older context as it can have timeouts and create new
-		//		md, _ := metadata.FromContext(cx)
-		//serveWebsocket(context.TODO(), w, r, service, c)
 		serveWebsocket(cx, w, r, service, c)
 		return
 	}
 
-	// create strategy
-	so := selector.WithStrategy(strategy(service.Services))
+	// create custom router
+	callOpt := client.WithRouter(router.New(service.Services))
 
 	// walk the standard call path
 	// get payload
@@ -174,7 +145,7 @@ func (h *rpcHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		)
 
 		// make the call
-		if err := c.Call(cx, req, response, client.WithSelectOption(so)); err != nil {
+		if err := c.Call(cx, req, response, callOpt); err != nil {
 			writeError(w, r, err)
 			return
 		}
@@ -209,7 +180,7 @@ func (h *rpcHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 			client.WithContentType(ct),
 		)
 		// make the call
-		if err := c.Call(cx, req, &response, client.WithSelectOption(so)); err != nil {
+		if err := c.Call(cx, req, &response, callOpt); err != nil {
 			writeError(w, r, err)
 			return
 		}
@@ -278,7 +249,7 @@ func requestPayload(r *http.Request) ([]byte, error) {
 			return nil, err
 		}
 		return raw.Marshal()
-	case strings.Contains(ct, "application/www-x-form-urlencoded"):
+	case strings.Contains(ct, "application/x-www-form-urlencoded"):
 		r.ParseForm()
 
 		// generate a new set of values from the form
@@ -294,7 +265,7 @@ func requestPayload(r *http.Request) ([]byte, error) {
 
 	// otherwise as per usual
 	ctx := r.Context()
-	// dont user meadata.FromContext as it mangles names
+	// dont user metadata.FromContext as it mangles names
 	md, ok := metadata.FromContext(ctx)
 	if !ok {
 		md = make(map[string]string)

api/handler/rpc/rpc_test.go

@@ -4,11 +4,10 @@ import (
 	"bytes"
 	"encoding/json"
 	"net/http"
-	"reflect"
 	"testing"
 
-	go_api "github.com/asim/go-micro/v3/api/proto"
 	"github.com/golang/protobuf/proto"
+	go_api "github.com/micro/go-micro/v3/api/proto"
 )
 
 func TestRequestPayloadFromRequest(t *testing.T) {
@@ -28,12 +27,7 @@ func TestRequestPayloadFromRequest(t *testing.T) {
 		t.Fatal("Failed to marshal proto to JSON ", err)
 	}
 
-	type jsonUrl struct {
-		Key1 string `json:"key1"`
-		Key2 string `json:"key2"`
-		Name string `json:"name"`
-	}
-	jUrl := &jsonUrl{Key1: "val1", Key2: "val2", Name: "Test"}
+	jsonUrlBytes := []byte(`{"key1":"val1","key2":"val2","name":"Test"}`)
 
 	t.Run("extracting a json from a POST request with url params", func(t *testing.T) {
 		r, err := http.NewRequest("POST", "http://localhost/my/path?key1=val1&key2=val2", bytes.NewReader(jsonBytes))
@@ -45,12 +39,8 @@ func TestRequestPayloadFromRequest(t *testing.T) {
 		if err != nil {
 			t.Fatalf("Failed to extract payload from request: %v", err)
 		}
-		extJUrl := &jsonUrl{}
-		if err := json.Unmarshal(extByte, extJUrl); err != nil {
-			t.Fatalf("Failed to unmarshal payload from request: %v", err)
-		}
-		if !reflect.DeepEqual(extJUrl, jUrl) {
-			t.Fatalf("Expected %v and %v to match", extJUrl, jUrl)
+		if string(extByte) != string(jsonUrlBytes) {
+			t.Fatalf("Expected %v and %v to match", string(extByte), jsonUrlBytes)
 		}
 	})
 

api/handler/rpc/stream.go

@@ -9,14 +9,14 @@ import (
 	"strings"
 	"time"
 
-	"github.com/asim/go-micro/v3/api"
-	"github.com/asim/go-micro/v3/client"
-	raw "github.com/asim/go-micro/v3/codec/bytes"
-	"github.com/asim/go-micro/v3/logger"
-	"github.com/asim/go-micro/v3/selector"
 	"github.com/gobwas/httphead"
 	"github.com/gobwas/ws"
 	"github.com/gobwas/ws/wsutil"
+	"github.com/micro/go-micro/v3/api"
+	"github.com/micro/go-micro/v3/client"
+	raw "github.com/micro/go-micro/v3/codec/bytes"
+	"github.com/micro/go-micro/v3/logger"
+	"github.com/micro/go-micro/v3/util/router"
 )
 
 // serveWebsocket will stream rpc back over websockets assuming json
@@ -38,7 +38,7 @@ func serveWebsocket(ctx context.Context, w http.ResponseWriter, r *http.Request,
 	}
 
 	hdr := make(http.Header)
-	if proto, ok := r.Header["Sec-Websocket-Protocol"]; ok {
+	if proto, ok := r.Header["Sec-WebSocket-Protocol"]; ok {
 		for _, p := range proto {
 			switch p {
 			case "binary":
@@ -110,9 +110,11 @@ func serveWebsocket(ctx context.Context, w http.ResponseWriter, r *http.Request,
 		client.StreamingRequest(),
 	)
 
-	so := selector.WithStrategy(strategy(service.Services))
+	// create custom router
+	callOpt := client.WithRouter(router.New(service.Services))
+
 	// create a new stream
-	stream, err := c.Stream(ctx, req, client.WithSelectOption(so))
+	stream, err := c.Stream(ctx, req, callOpt)
 	if err != nil {
 		if logger.V(logger.ErrorLevel, logger.DefaultLogger) {
 			logger.Error(err)

api/handler/web/web.go

@@ -5,15 +5,16 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"math/rand"
 	"net"
 	"net/http"
 	"net/http/httputil"
 	"net/url"
 	"strings"
 
-	"github.com/asim/go-micro/v3/api"
-	"github.com/asim/go-micro/v3/api/handler"
-	"github.com/asim/go-micro/v3/selector"
+	"github.com/micro/go-micro/v3/api"
+	"github.com/micro/go-micro/v3/api/handler"
+	"github.com/micro/go-micro/v3/registry"
 )
 
 const (
@@ -70,16 +71,19 @@ func (wh *webHandler) getService(r *http.Request) (string, error) {
 		return "", errors.New("no route found")
 	}
 
-	// create a random selector
-	next := selector.Random(service.Services)
-
-	// get the next node
-	s, err := next()
-	if err != nil {
-		return "", nil
+	// get the nodes
+	var nodes []*registry.Node
+	for _, srv := range service.Services {
+		nodes = append(nodes, srv.Nodes...)
+	}
+	if len(nodes) == 0 {
+		return "", errors.New("no route found")
 	}
 
-	return fmt.Sprintf("http://%s", s.Address), nil
+	// select a random node
+	node := nodes[rand.Int()%len(nodes)]
+
+	return fmt.Sprintf("http://%s", node.Address), nil
 }
 
 // serveWebSocket used to serve a web socket proxied connection

api/resolver/grpc/grpc.go

@@ -6,12 +6,17 @@ import (
 	"net/http"
 	"strings"
 
-	"github.com/asim/go-micro/v3/api/resolver"
+	"github.com/micro/go-micro/v3/api/resolver"
 )
 
-type Resolver struct{}
+type Resolver struct {
+	opts resolver.Options
+}
+
+func (r *Resolver) Resolve(req *http.Request, opts ...resolver.ResolveOption) (*resolver.Endpoint, error) {
+	// parse options
+	options := resolver.NewResolveOptions(opts...)
 
-func (r *Resolver) Resolve(req *http.Request) (*resolver.Endpoint, error) {
 	// /foo.Bar/Service
 	if req.URL.Path == "/" {
 		return nil, errors.New("unknown name")
@@ -26,6 +31,7 @@ func (r *Resolver) Resolve(req *http.Request) (*resolver.Endpoint, error) {
 		Host:   req.Host,
 		Method: req.Method,
 		Path:   req.URL.Path,
+		Domain: options.Domain,
 	}, nil
 }
 
@@ -34,5 +40,5 @@ func (r *Resolver) String() string {
 }
 
 func NewResolver(opts ...resolver.Option) resolver.Resolver {
-	return &Resolver{}
+	return &Resolver{opts: resolver.NewOptions(opts...)}
 }

api/resolver/host/host.go

@@ -4,19 +4,23 @@ package host
 import (
 	"net/http"
 
-	"github.com/asim/go-micro/v3/api/resolver"
+	"github.com/micro/go-micro/v3/api/resolver"
 )
 
 type Resolver struct {
 	opts resolver.Options
 }
 
-func (r *Resolver) Resolve(req *http.Request) (*resolver.Endpoint, error) {
+func (r *Resolver) Resolve(req *http.Request, opts ...resolver.ResolveOption) (*resolver.Endpoint, error) {
+	// parse options
+	options := resolver.NewResolveOptions(opts...)
+
 	return &resolver.Endpoint{
 		Name:   req.Host,
 		Host:   req.Host,
 		Method: req.Method,
 		Path:   req.URL.Path,
+		Domain: options.Domain,
 	}, nil
 }
 

api/resolver/options.go

@@ -1,23 +1,16 @@
 package resolver
 
 import (
-	"net/http"
+	"github.com/micro/go-micro/v3/registry"
 )
 
-// NewOptions returns new initialised options
-func NewOptions(opts ...Option) Options {
-	var options Options
-	for _, o := range opts {
-		o(&options)
-	}
-
-	if options.Namespace == nil {
-		options.Namespace = StaticNamespace("go.micro")
-	}
-
-	return options
+type Options struct {
+	Handler       string
+	ServicePrefix string
 }
 
+type Option func(o *Options)
+
 // WithHandler sets the handler being used
 func WithHandler(h string) Option {
 	return func(o *Options) {
@@ -25,9 +18,46 @@ func WithHandler(h string) Option {
 	}
 }
 
-// WithNamespace sets the function which determines the namespace for a request
-func WithNamespace(n func(*http.Request) string) Option {
+// WithServicePrefix sets the ServicePrefix option
+func WithServicePrefix(p string) Option {
 	return func(o *Options) {
-		o.Namespace = n
+		o.ServicePrefix = p
 	}
 }
+
+// NewOptions returns new initialised options
+func NewOptions(opts ...Option) Options {
+	var options Options
+	for _, o := range opts {
+		o(&options)
+	}
+	return options
+}
+
+// ResolveOptions are used when resolving a request
+type ResolveOptions struct {
+	Domain string
+}
+
+// ResolveOption sets an option
+type ResolveOption func(*ResolveOptions)
+
+// Domain sets the resolve Domain option
+func Domain(n string) ResolveOption {
+	return func(o *ResolveOptions) {
+		o.Domain = n
+	}
+}
+
+// NewResolveOptions returns new initialised resolve options
+func NewResolveOptions(opts ...ResolveOption) ResolveOptions {
+	var options ResolveOptions
+	for _, o := range opts {
+		o(&options)
+	}
+	if len(options.Domain) == 0 {
+		options.Domain = registry.DefaultDomain
+	}
+
+	return options
+}

api/resolver/path/path.go

@@ -5,26 +5,29 @@ import (
 	"net/http"
 	"strings"
 
-	"github.com/asim/go-micro/v3/api/resolver"
+	"github.com/micro/go-micro/v3/api/resolver"
 )
 
 type Resolver struct {
 	opts resolver.Options
 }
 
-func (r *Resolver) Resolve(req *http.Request) (*resolver.Endpoint, error) {
+func (r *Resolver) Resolve(req *http.Request, opts ...resolver.ResolveOption) (*resolver.Endpoint, error) {
+	// parse options
+	options := resolver.NewResolveOptions(opts...)
+
 	if req.URL.Path == "/" {
 		return nil, resolver.ErrNotFound
 	}
 
 	parts := strings.Split(req.URL.Path[1:], "/")
-	ns := r.opts.Namespace(req)
 
 	return &resolver.Endpoint{
-		Name:   ns + "." + parts[0],
+		Name:   r.opts.ServicePrefix + "." + parts[0],
 		Host:   req.Host,
 		Method: req.Method,
 		Path:   req.URL.Path,
+		Domain: options.Domain,
 	}, nil
 }
 

api/resolver/resolver.go

@@ -13,7 +13,7 @@ var (
 
 // Resolver resolves requests to endpoints
 type Resolver interface {
-	Resolve(r *http.Request) (*Endpoint, error)
+	Resolve(r *http.Request, opts ...ResolveOption) (*Endpoint, error)
 	String() string
 }
 
@@ -27,18 +27,6 @@ type Endpoint struct {
 	Method string
 	// HTTP Path e.g /greeter.
 	Path string
-}
-
-type Options struct {
-	Handler   string
-	Namespace func(*http.Request) string
-}
-
-type Option func(o *Options)
-
-// StaticNamespace returns the same namespace for each request
-func StaticNamespace(ns string) func(*http.Request) string {
-	return func(*http.Request) string {
-		return ns
-	}
+	// Domain endpoint exists within
+	Domain string
 }

api/resolver/subdomain/subdomain.go

@@ -0,0 +1,85 @@
+// Package subdomain is a resolver which uses the subdomain to determine the domain to route to. It
+// offloads the endpoint resolution to a child resolver which is provided in New.
+package subdomain
+
+import (
+	"net"
+	"net/http"
+	"strings"
+
+	"github.com/micro/go-micro/v3/api/resolver"
+	"github.com/micro/go-micro/v3/logger"
+	"golang.org/x/net/publicsuffix"
+)
+
+func NewResolver(parent resolver.Resolver, opts ...resolver.Option) resolver.Resolver {
+	options := resolver.NewOptions(opts...)
+	return &Resolver{options, parent}
+}
+
+type Resolver struct {
+	opts resolver.Options
+	resolver.Resolver
+}
+
+func (r *Resolver) Resolve(req *http.Request, opts ...resolver.ResolveOption) (*resolver.Endpoint, error) {
+	if dom := r.Domain(req); len(dom) > 0 {
+		opts = append(opts, resolver.Domain(dom))
+	}
+
+	return r.Resolver.Resolve(req, opts...)
+}
+
+func (r *Resolver) Domain(req *http.Request) string {
+	// determine the host, e.g. foobar.m3o.app
+	host := req.URL.Hostname()
+	if len(host) == 0 {
+		if h, _, err := net.SplitHostPort(req.Host); err == nil {
+			host = h // host does contain a port
+		} else if strings.Contains(err.Error(), "missing port in address") {
+			host = req.Host // host does not contain a port
+		}
+	}
+
+	// check for an ip address
+	if net.ParseIP(host) != nil {
+		return ""
+	}
+
+	// check for dev enviroment
+	if host == "localhost" || host == "127.0.0.1" {
+		return ""
+	}
+
+	// extract the top level domain plus one (e.g. 'myapp.com')
+	domain, err := publicsuffix.EffectiveTLDPlusOne(host)
+	if err != nil {
+		logger.Debugf("Unable to extract domain from %v", host)
+		return ""
+	}
+
+	// there was no subdomain
+	if host == domain {
+		return ""
+	}
+
+	// remove the domain from the host, leaving the subdomain, e.g. "staging.foo.myapp.com" => "staging.foo"
+	subdomain := strings.TrimSuffix(host, "."+domain)
+
+	// ignore the API subdomain
+	if subdomain == "api" {
+		return ""
+	}
+
+	// return the reversed subdomain as the namespace, e.g. "staging.foo" => "foo-staging"
+	comps := strings.Split(subdomain, ".")
+	for i := len(comps)/2 - 1; i >= 0; i-- {
+		opp := len(comps) - 1 - i
+		comps[i], comps[opp] = comps[opp], comps[i]
+	}
+	return strings.Join(comps, "-")
+}
+
+func (r *Resolver) String() string {
+	return "subdomain"
+}

api/resolver/subdomain/subdomain_test.go

@@ -0,0 +1,71 @@
+package subdomain
+
+import (
+	"net/http"
+	"net/url"
+	"testing"
+
+	"github.com/micro/go-micro/v3/api/resolver/vpath"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestResolve(t *testing.T) {
+	tt := []struct {
+		Name   string
+		Host   string
+		Result string
+	}{
+		{
+			Name:   "Top level domain",
+			Host:   "micro.mu",
+			Result: "micro",
+		},
+		{
+			Name:   "Effective top level domain",
+			Host:   "micro.com.au",
+			Result: "micro",
+		},
+		{
+			Name:   "Subdomain dev",
+			Host:   "dev.micro.mu",
+			Result: "dev",
+		},
+		{
+			Name:   "Subdomain foo",
+			Host:   "foo.micro.mu",
+			Result: "foo",
+		},
+		{
+			Name:   "Multi-level subdomain",
+			Host:   "staging.myapp.m3o.app",
+			Result: "myapp-staging",
+		},
+		{
+			Name:   "Dev host",
+			Host:   "127.0.0.1",
+			Result: "micro",
+		},
+		{
+			Name:   "Localhost",
+			Host:   "localhost",
+			Result: "micro",
+		},
+		{
+			Name:   "IP host",
+			Host:   "81.151.101.146",
+			Result: "micro",
+		},
+	}
+
+	for _, tc := range tt {
+		t.Run(tc.Name, func(t *testing.T) {
+			r := NewResolver(vpath.NewResolver())
+			result, err := r.Resolve(&http.Request{URL: &url.URL{Host: tc.Host, Path: "foo/bar"}})
+			assert.Nil(t, err, "Expecter err to be nil")
+			if result != nil {
+				assert.Equal(t, tc.Result, result.Domain, "Expected %v but got %v", tc.Result, result.Domain)
+			}
+		})
+	}
+}

api/resolver/vpath/vpath.go

@@ -7,7 +7,7 @@ import (
 	"regexp"
 	"strings"
 
-	"github.com/asim/go-micro/v3/api/resolver"
+	"github.com/micro/go-micro/v3/api/resolver"
 )
 
 func NewResolver(opts ...resolver.Option) resolver.Resolver {
@@ -22,36 +22,41 @@ var (
 	re = regexp.MustCompile("^v[0-9]+$")
 )
 
-func (r *Resolver) Resolve(req *http.Request) (*resolver.Endpoint, error) {
+func (r *Resolver) Resolve(req *http.Request, opts ...resolver.ResolveOption) (*resolver.Endpoint, error) {
 	if req.URL.Path == "/" {
 		return nil, errors.New("unknown name")
 	}
 
+	options := resolver.NewResolveOptions(opts...)
+
 	parts := strings.Split(req.URL.Path[1:], "/")
 	if len(parts) == 1 {
 		return &resolver.Endpoint{
-			Name:   r.withNamespace(req, parts...),
+			Name:   r.withPrefix(parts...),
 			Host:   req.Host,
 			Method: req.Method,
 			Path:   req.URL.Path,
+			Domain: options.Domain,
 		}, nil
 	}
 
 	// /v1/foo
 	if re.MatchString(parts[0]) {
 		return &resolver.Endpoint{
-			Name:   r.withNamespace(req, parts[0:2]...),
+			Name:   r.withPrefix(parts[0:2]...),
 			Host:   req.Host,
 			Method: req.Method,
 			Path:   req.URL.Path,
+			Domain: options.Domain,
 		}, nil
 	}
 
 	return &resolver.Endpoint{
-		Name:   r.withNamespace(req, parts[0]),
+		Name:   r.withPrefix(parts[0]),
 		Host:   req.Host,
 		Method: req.Method,
 		Path:   req.URL.Path,
+		Domain: options.Domain,
 	}, nil
 }
 
@@ -59,11 +64,12 @@ func (r *Resolver) String() string {
 	return "path"
 }
 
-func (r *Resolver) withNamespace(req *http.Request, parts ...string) string {
-	ns := r.opts.Namespace(req)
-	if len(ns) == 0 {
-		return strings.Join(parts, ".")
+// withPrefix transforms "foo" into "go.micro.api.foo"
+func (r *Resolver) withPrefix(parts ...string) string {
+	p := r.opts.ServicePrefix
+	if len(p) > 0 {
+		parts = append([]string{p}, parts...)
 	}
 
-	return strings.Join(append([]string{ns}, parts...), ".")
+	return strings.Join(parts, ".")
 }

api/router/options.go

@@ -1,9 +1,10 @@
 package router
 
 import (
-	"github.com/asim/go-micro/v3/api/resolver"
-	"github.com/asim/go-micro/v3/api/resolver/vpath"
-	"github.com/asim/go-micro/v3/registry"
+	"github.com/micro/go-micro/v3/api/resolver"
+	"github.com/micro/go-micro/v3/api/resolver/vpath"
+	"github.com/micro/go-micro/v3/registry"
+	"github.com/micro/go-micro/v3/registry/mdns"
 )
 
 type Options struct {
@@ -17,7 +18,7 @@ type Option func(o *Options)
 func NewOptions(opts ...Option) Options {
 	options := Options{
 		Handler:  "meta",
-		Registry: registry.DefaultRegistry,
+		Registry: mdns.NewRegistry(),
 	}
 
 	for _, o := range opts {

api/router/registry/registry.go

@@ -10,13 +10,13 @@ import (
 	"sync"
 	"time"
 
-	"github.com/asim/go-micro/v3/api"
-	"github.com/asim/go-micro/v3/api/router"
-	"github.com/asim/go-micro/v3/api/router/util"
-	"github.com/asim/go-micro/v3/logger"
-	"github.com/asim/go-micro/v3/metadata"
-	"github.com/asim/go-micro/v3/registry"
-	"github.com/asim/go-micro/v3/registry/cache"
+	"github.com/micro/go-micro/v3/api"
+	"github.com/micro/go-micro/v3/api/router"
+	"github.com/micro/go-micro/v3/logger"
+	"github.com/micro/go-micro/v3/metadata"
+	"github.com/micro/go-micro/v3/registry"
+	"github.com/micro/go-micro/v3/registry/cache"
+	util "github.com/micro/go-micro/v3/util/router"
 )
 
 // endpoint struct, that holds compiled pcre
@@ -99,7 +99,7 @@ func (r *registryRouter) process(res *registry.Result) {
 	service, err := r.rc.GetService(res.Service.Name)
 	if err != nil {
 		if logger.V(logger.ErrorLevel, logger.DefaultLogger) {
-			logger.Errorf("unable to get service: %v", err)
+			logger.Errorf("unable to get %v service: %v", res.Service.Name, err)
 		}
 		return
 	}
@@ -127,7 +127,10 @@ func (r *registryRouter) store(services []*registry.Service) {
 			key := fmt.Sprintf("%s.%s", service.Name, sep.Name)
 			// decode endpoint
 			end := api.Decode(sep.Metadata)
-
+			// no endpoint or no name
+			if end == nil || len(end.Name) == 0 {
+				continue
+			}
 			// if we got nothing skip
 			if err := api.Validate(end); err != nil {
 				if logger.V(logger.TraceLevel, logger.DefaultLogger) {
@@ -260,7 +263,7 @@ func (r *registryRouter) watch() {
 			res, err := w.Next()
 			if err != nil {
 				if logger.V(logger.ErrorLevel, logger.DefaultLogger) {
-					logger.Errorf("error getting next endoint: %v", err)
+					logger.Errorf("error getting next endpoint: %v", err)
 				}
 				close(ch)
 				break
@@ -434,7 +437,7 @@ func (r *registryRouter) Route(req *http.Request) (*api.Service, error) {
 	name := rp.Name
 
 	// get service
-	services, err := r.rc.GetService(name)
+	services, err := r.rc.GetService(name, registry.GetDomain(rp.Domain))
 	if err != nil {
 		return nil, err
 	}

api/router/registry/registry_test.go

@@ -3,7 +3,7 @@ package registry
 import (
 	"testing"
 
-	"github.com/asim/go-micro/v3/registry"
+	"github.com/micro/go-micro/v3/registry"
 	"github.com/stretchr/testify/assert"
 )
 

api/router/router.go

@@ -4,7 +4,7 @@ package router
 import (
 	"net/http"
 
-	"github.com/asim/go-micro/v3/api"
+	"github.com/micro/go-micro/v3/api"
 )
 
 // Router is used to determine an endpoint for a request

api/router/router_test.go

@@ -0,0 +1,251 @@
+package router_test
+
+import (
+	"context"
+	"fmt"
+	"io/ioutil"
+	"log"
+	"net/http"
+	"testing"
+	"time"
+
+	"github.com/micro/go-micro/v3/api"
+	"github.com/micro/go-micro/v3/api/handler"
+	"github.com/micro/go-micro/v3/api/handler/rpc"
+	"github.com/micro/go-micro/v3/api/router"
+	rregistry "github.com/micro/go-micro/v3/api/router/registry"
+	rstatic "github.com/micro/go-micro/v3/api/router/static"
+	"github.com/micro/go-micro/v3/client"
+	gcli "github.com/micro/go-micro/v3/client/grpc"
+	rmemory "github.com/micro/go-micro/v3/registry/memory"
+	rt "github.com/micro/go-micro/v3/router"
+	regRouter "github.com/micro/go-micro/v3/router/registry"
+	"github.com/micro/go-micro/v3/server"
+	gsrv "github.com/micro/go-micro/v3/server/grpc"
+	pb "github.com/micro/go-micro/v3/server/grpc/proto"
+)
+
+// server is used to implement helloworld.GreeterServer.
+type testServer struct {
+	msgCount int
+}
+
+// TestHello implements helloworld.GreeterServer
+func (s *testServer) Call(ctx context.Context, req *pb.Request, rsp *pb.Response) error {
+	rsp.Msg = "Hello " + req.Uuid
+	return nil
+}
+
+// TestHello implements helloworld.GreeterServer
+func (s *testServer) CallPcre(ctx context.Context, req *pb.Request, rsp *pb.Response) error {
+	rsp.Msg = "Hello " + req.Uuid
+	return nil
+}
+
+// TestHello implements helloworld.GreeterServer
+func (s *testServer) CallPcreInvalid(ctx context.Context, req *pb.Request, rsp *pb.Response) error {
+	rsp.Msg = "Hello " + req.Uuid
+	return nil
+}
+
+func initial(t *testing.T) (server.Server, client.Client) {
+	r := rmemory.NewRegistry()
+
+	// create a new client
+	s := gsrv.NewServer(
+		server.Name("foo"),
+		server.Registry(r),
+	)
+
+	rtr := regRouter.NewRouter(
+		rt.Registry(r),
+	)
+
+	// create a new server
+	c := gcli.NewClient(
+		client.Router(rtr),
+	)
+
+	h := &testServer{}
+	pb.RegisterTestHandler(s, h)
+
+	if err := s.Start(); err != nil {
+		t.Fatalf("failed to start: %v", err)
+	}
+
+	return s, c
+}
+
+func check(t *testing.T, addr string, path string, expected string) {
+	req, err := http.NewRequest("POST", fmt.Sprintf(path, addr), nil)
+	if err != nil {
+		t.Fatalf("Failed to created http.Request: %v", err)
+	}
+	req.Header.Set("Content-Type", "application/json")
+	rsp, err := (&http.Client{}).Do(req)
+	if err != nil {
+		t.Fatalf("Failed to created http.Request: %v", err)
+	}
+	defer rsp.Body.Close()
+
+	buf, err := ioutil.ReadAll(rsp.Body)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	jsonMsg := expected
+	if string(buf) != jsonMsg {
+		t.Fatalf("invalid message received, parsing error %s != %s", buf, jsonMsg)
+	}
+}
+
+func TestRouterRegistryPcre(t *testing.T) {
+	s, c := initial(t)
+	defer s.Stop()
+
+	router := rregistry.NewRouter(
+		router.WithHandler(rpc.Handler),
+		router.WithRegistry(s.Options().Registry),
+	)
+	hrpc := rpc.NewHandler(
+		handler.WithClient(c),
+		handler.WithRouter(router),
+	)
+	hsrv := &http.Server{
+		Handler:        hrpc,
+		Addr:           "127.0.0.1:6543",
+		WriteTimeout:   15 * time.Second,
+		ReadTimeout:    15 * time.Second,
+		IdleTimeout:    20 * time.Second,
+		MaxHeaderBytes: 1024 * 1024 * 1, // 1Mb
+	}
+
+	go func() {
+		log.Println(hsrv.ListenAndServe())
+	}()
+
+	defer hsrv.Close()
+	time.Sleep(1 * time.Second)
+	check(t, hsrv.Addr, "http://%s/api/v0/test/call/TEST", `{"msg":"Hello TEST"}`)
+}
+
+func TestRouterStaticPcre(t *testing.T) {
+	s, c := initial(t)
+	defer s.Stop()
+
+	router := rstatic.NewRouter(
+		router.WithHandler(rpc.Handler),
+		router.WithRegistry(s.Options().Registry),
+	)
+
+	err := router.Register(&api.Endpoint{
+		Name:    "foo.Test.Call",
+		Method:  []string{"POST"},
+		Path:    []string{"^/api/v0/test/call/?$"},
+		Handler: "rpc",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	hrpc := rpc.NewHandler(
+		handler.WithClient(c),
+		handler.WithRouter(router),
+	)
+	hsrv := &http.Server{
+		Handler:        hrpc,
+		Addr:           "127.0.0.1:6543",
+		WriteTimeout:   15 * time.Second,
+		ReadTimeout:    15 * time.Second,
+		IdleTimeout:    20 * time.Second,
+		MaxHeaderBytes: 1024 * 1024 * 1, // 1Mb
+	}
+
+	go func() {
+		log.Println(hsrv.ListenAndServe())
+	}()
+	defer hsrv.Close()
+
+	time.Sleep(1 * time.Second)
+	check(t, hsrv.Addr, "http://%s/api/v0/test/call", `{"msg":"Hello "}`)
+}
+
+func TestRouterStaticGpath(t *testing.T) {
+	s, c := initial(t)
+	defer s.Stop()
+
+	router := rstatic.NewRouter(
+		router.WithHandler(rpc.Handler),
+		router.WithRegistry(s.Options().Registry),
+	)
+
+	err := router.Register(&api.Endpoint{
+		Name:    "foo.Test.Call",
+		Method:  []string{"POST"},
+		Path:    []string{"/api/v0/test/call/{uuid}"},
+		Handler: "rpc",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	hrpc := rpc.NewHandler(
+		handler.WithClient(c),
+		handler.WithRouter(router),
+	)
+	hsrv := &http.Server{
+		Handler:        hrpc,
+		Addr:           "127.0.0.1:6543",
+		WriteTimeout:   15 * time.Second,
+		ReadTimeout:    15 * time.Second,
+		IdleTimeout:    20 * time.Second,
+		MaxHeaderBytes: 1024 * 1024 * 1, // 1Mb
+	}
+
+	go func() {
+		log.Println(hsrv.ListenAndServe())
+	}()
+	defer hsrv.Close()
+
+	time.Sleep(1 * time.Second)
+	check(t, hsrv.Addr, "http://%s/api/v0/test/call/TEST", `{"msg":"Hello TEST"}`)
+}
+
+func TestRouterStaticPcreInvalid(t *testing.T) {
+	var ep *api.Endpoint
+	var err error
+
+	s, c := initial(t)
+	defer s.Stop()
+
+	router := rstatic.NewRouter(
+		router.WithHandler(rpc.Handler),
+		router.WithRegistry(s.Options().Registry),
+	)
+
+	ep = &api.Endpoint{
+		Name:    "foo.Test.Call",
+		Method:  []string{"POST"},
+		Path:    []string{"^/api/v0/test/call/?"},
+		Handler: "rpc",
+	}
+
+	err = router.Register(ep)
+	if err == nil {
+		t.Fatalf("invalid endpoint %v", ep)
+	}
+
+	ep = &api.Endpoint{
+		Name:    "foo.Test.Call",
+		Method:  []string{"POST"},
+		Path:    []string{"/api/v0/test/call/?$"},
+		Handler: "rpc",
+	}
+
+	err = router.Register(ep)
+	if err == nil {
+		t.Fatalf("invalid endpoint %v", ep)
+	}
+
+	_ = c
+}

api/router/static/static.go

@@ -8,13 +8,13 @@ import (
 	"strings"
 	"sync"
 
-	"github.com/asim/go-micro/v3/api"
-	"github.com/asim/go-micro/v3/api/router"
-	"github.com/asim/go-micro/v3/api/router/util"
-	"github.com/asim/go-micro/v3/logger"
-	"github.com/asim/go-micro/v3/metadata"
-	"github.com/asim/go-micro/v3/registry"
-	rutil "github.com/asim/go-micro/v3/util/registry"
+	"github.com/micro/go-micro/v3/api"
+	"github.com/micro/go-micro/v3/api/router"
+	"github.com/micro/go-micro/v3/logger"
+	"github.com/micro/go-micro/v3/metadata"
+	"github.com/micro/go-micro/v3/registry"
+	rutil "github.com/micro/go-micro/v3/util/registry"
+	util "github.com/micro/go-micro/v3/util/router"
 )
 
 type endpoint struct {

api/server/acme/autocert/autocert.go

@@ -7,8 +7,8 @@ import (
 	"net"
 	"os"
 
-	"github.com/asim/go-micro/v3/api/server/acme"
-	"github.com/asim/go-micro/v3/logger"
+	"github.com/micro/go-micro/v3/api/server/acme"
+	"github.com/micro/go-micro/v3/logger"
 	"golang.org/x/crypto/acme/autocert"
 )
 

api/server/acme/certmagic/certmagic.go

@@ -7,9 +7,9 @@ import (
 	"net"
 	"time"
 
-	"github.com/asim/go-micro/v3/api/server/acme"
-	"github.com/asim/go-micro/v3/logger"
 	"github.com/caddyserver/certmagic"
+	"github.com/micro/go-micro/v3/api/server/acme"
+	"github.com/micro/go-micro/v3/logger"
 )
 
 type certmagicProvider struct {

api/server/acme/certmagic/storage.go

@@ -2,7 +2,6 @@ package certmagic
 
 import (
 	"bytes"
-	"context"
 	"encoding/gob"
 	"errors"
 	"fmt"
@@ -10,9 +9,9 @@ import (
 	"strings"
 	"time"
 
-	"github.com/asim/go-micro/v3/store"
-	"github.com/asim/go-micro/v3/sync"
 	"github.com/caddyserver/certmagic"
+	"github.com/micro/go-micro/v3/store"
+	"github.com/micro/go-micro/v3/sync"
 )
 
 // File represents a "File" that will be stored in store.Store - the contents and last modified time
@@ -31,7 +30,7 @@ type storage struct {
 	store store.Store
 }
 
-func (s *storage) Lock(ctx context.Context, key string) error {
+func (s *storage) Lock(key string) error {
 	return s.lock.Lock(key, sync.LockTTL(10*time.Minute))
 }
 

api/server/acme/options.go

@@ -1,6 +1,6 @@
 package acme
 
-import "github.com/go-acme/lego/v4/challenge"
+import "github.com/go-acme/lego/v3/challenge"
 
 // Option (or Options) are passed to New() to configure providers
 type Option func(o *Options)

api/server/cors/cors.go

@@ -4,54 +4,41 @@ import (
 	"net/http"
 )
 
-type Config struct {
-	AllowOrigin      string
-	AllowCredentials bool
-	AllowMethods     string
-	AllowHeaders     string
+// CombinedCORSHandler wraps a server and provides CORS headers
+func CombinedCORSHandler(h http.Handler) http.Handler {
+	return corsHandler{h}
 }
 
-// CombinedCORSHandler wraps a server and provides CORS headers
-func CombinedCORSHandler(h http.Handler, config *Config) http.Handler {
-	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		if config != nil {
-			SetHeaders(w, r, config)
-		}
-		if r.Method == "OPTIONS" {
-			return
-		}
+type corsHandler struct {
+	handler http.Handler
+}
 
-		h.ServeHTTP(w, r)
-	})
+func (c corsHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	SetHeaders(w, r)
+
+	if r.Method == "OPTIONS" {
+		return
+	}
+
+	c.handler.ServeHTTP(w, r)
 }
 
 // SetHeaders sets the CORS headers
-func SetHeaders(w http.ResponseWriter, _ *http.Request, config *Config) {
+func SetHeaders(w http.ResponseWriter, r *http.Request) {
 	set := func(w http.ResponseWriter, k, v string) {
 		if v := w.Header().Get(k); len(v) > 0 {
 			return
 		}
 		w.Header().Set(k, v)
 	}
-	//For forward-compatible code, default values may not be provided in the future
-	if config.AllowCredentials {
-		set(w, "Access-Control-Allow-Credentials", "true")
+
+	if origin := r.Header.Get("Origin"); len(origin) > 0 {
+		set(w, "Access-Control-Allow-Origin", origin)
 	} else {
-		set(w, "Access-Control-Allow-Credentials", "false")
-	}
-	if config.AllowOrigin == "" {
 		set(w, "Access-Control-Allow-Origin", "*")
-	} else {
-		set(w, "Access-Control-Allow-Origin", config.AllowOrigin)
-	}
-	if config.AllowMethods == "" {
-		set(w, "Access-Control-Allow-Methods", "POST, PATCH, GET, OPTIONS, PUT, DELETE")
-	} else {
-		set(w, "Access-Control-Allow-Methods", config.AllowMethods)
-	}
-	if config.AllowHeaders == "" {
-		set(w, "Access-Control-Allow-Headers", "Accept, Content-Type, Content-Length, Accept-Encoding, X-CSRF-Token, Authorization")
-	} else {
-		set(w, "Access-Control-Allow-Headers", config.AllowHeaders)
 	}
+
+	set(w, "Access-Control-Allow-Credentials", "true")
+	set(w, "Access-Control-Allow-Methods", "POST, PATCH, GET, OPTIONS, PUT, DELETE")
+	set(w, "Access-Control-Allow-Headers", "Accept, Content-Type, Content-Length, Accept-Encoding, X-CSRF-Token, Authorization")
 }

api/server/http/http.go

@@ -8,10 +8,10 @@ import (
 	"os"
 	"sync"
 
-	"github.com/asim/go-micro/v3/api/server"
-	"github.com/asim/go-micro/v3/api/server/cors"
-	"github.com/asim/go-micro/v3/logger"
 	"github.com/gorilla/handlers"
+	"github.com/micro/go-micro/v3/api/server"
+	"github.com/micro/go-micro/v3/api/server/cors"
+	"github.com/micro/go-micro/v3/logger"
 )
 
 type httpServer struct {
@@ -60,7 +60,7 @@ func (s *httpServer) Handle(path string, handler http.Handler) {
 
 	// wrap with cors
 	if s.opts.EnableCORS {
-		handler = cors.CombinedCORSHandler(handler, s.opts.CORSConfig)
+		handler = cors.CombinedCORSHandler(handler)
 	}
 
 	// wrap with logger

api/server/http/http_test.go

@@ -2,8 +2,6 @@ package http
 
 import (
 	"fmt"
-	"github.com/asim/go-micro/v3/api/server"
-	"github.com/asim/go-micro/v3/api/server/cors"
 	"io/ioutil"
 	"net/http"
 	"testing"
@@ -41,74 +39,3 @@ func TestHTTPServer(t *testing.T) {
 		t.Fatal(err)
 	}
 }
-
-func TestCORSHTTPServer(t *testing.T) {
-	testResponse := "hello world"
-	testAllowOrigin := "*"
-	testAllowCredentials := true
-	testAllowMethods := "GET"
-	testAllowHeaders := "Accept, Content-Type, Content-Length"
-
-	s := NewServer("localhost:0",
-		server.EnableCORS(true),
-		server.CORSConfig(&cors.Config{
-			AllowCredentials: testAllowCredentials,
-			AllowOrigin:      testAllowOrigin,
-			AllowMethods:     testAllowMethods,
-			AllowHeaders:     testAllowHeaders,
-		}),
-	)
-
-	s.Handle("/", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		fmt.Fprint(w, testResponse)
-	}))
-
-	if err := s.Start(); err != nil {
-		t.Fatal(err)
-	}
-
-	rsp, err := http.Get(fmt.Sprintf("http://%s/", s.Address()))
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer rsp.Body.Close()
-
-	b, err := ioutil.ReadAll(rsp.Body)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	if string(b) != testResponse {
-		t.Fatalf("Unexpected response, got %s, expected %s", string(b), testResponse)
-	}
-
-	allowCredentials := rsp.Header.Get("Access-Control-Allow-Credentials")
-	getTestCredentialsStr := func() string {
-		if testAllowCredentials == true {
-			return "true"
-		} else {
-			return "false"
-		}
-	}
-	if getTestCredentialsStr() != allowCredentials {
-		t.Fatalf("Unexpected Access-Control-Allow-Credentials, got %s, expected %s", allowCredentials, getTestCredentialsStr())
-	}
-
-	allowOrigin := rsp.Header.Get("Access-Control-Allow-Origin")
-	if testAllowOrigin != allowOrigin {
-		t.Fatalf("Unexpected Access-Control-Allow-Origins, got %s, expected %s", allowOrigin, testAllowOrigin)
-	}
-
-	allowMethods := rsp.Header.Get("Access-Control-Allow-Methods")
-	if testAllowMethods != allowMethods {
-		t.Fatalf("Unexpected Access-Control-Allow-Methods, got %s, expected %s", allowMethods, testAllowMethods)
-	}
-	allowHeaders := rsp.Header.Get("Access-Control-Allow-Headers")
-	if testAllowHeaders != allowHeaders {
-		t.Fatalf("Unexpected Access-Control-Allow-Headers, got %s, expected %s", allowHeaders, testAllowHeaders)
-	}
-
-	if err := s.Stop(); err != nil {
-		t.Fatal(err)
-	}
-}

api/server/options.go

@@ -2,11 +2,10 @@ package server
 
 import (
 	"crypto/tls"
-	"github.com/asim/go-micro/v3/api/server/cors"
 	"net/http"
 
-	"github.com/asim/go-micro/v3/api/resolver"
-	"github.com/asim/go-micro/v3/api/server/acme"
+	"github.com/micro/go-micro/v3/api/resolver"
+	"github.com/micro/go-micro/v3/api/server/acme"
 )
 
 type Option func(o *Options)
@@ -14,7 +13,6 @@ type Option func(o *Options)
 type Options struct {
 	EnableACME   bool
 	EnableCORS   bool
-	CORSConfig   *cors.Config
 	ACMEProvider acme.Provider
 	EnableTLS    bool
 	ACMEHosts    []string
@@ -25,9 +23,9 @@ type Options struct {
 
 type Wrapper func(h http.Handler) http.Handler
 
-func WrapHandler(w Wrapper) Option {
+func WrapHandler(w ...Wrapper) Option {
 	return func(o *Options) {
-		o.Wrappers = append(o.Wrappers, w)
+		o.Wrappers = append(o.Wrappers, w...)
 	}
 }
 
@@ -37,12 +35,6 @@ func EnableCORS(b bool) Option {
 	}
 }
 
-func CORSConfig(c *cors.Config) Option {
-	return func(o *Options) {
-		o.CORSConfig = c
-	}
-}
-
 func EnableACME(b bool) Option {
 	return func(o *Options) {
 		o.EnableACME = b

auth/auth.go

@@ -31,24 +31,20 @@ type Auth interface {
 	Options() Options
 	// Generate a new account
 	Generate(id string, opts ...GenerateOption) (*Account, error)
+	// Verify an account has access to a resource using the rules
+	Verify(acc *Account, res *Resource, opts ...VerifyOption) error
 	// Inspect a token
 	Inspect(token string) (*Account, error)
 	// Token generated using refresh token or credentials
 	Token(opts ...TokenOption) (*Token, error)
-	// String returns the name of the implementation
-	String() string
-}
-
-// Rules manages access to resources
-type Rules interface {
-	// Verify an account has access to a resource using the rules
-	Verify(acc *Account, res *Resource, opts ...VerifyOption) error
 	// Grant access to a resource
 	Grant(rule *Rule) error
 	// Revoke access to a resource
 	Revoke(rule *Rule) error
-	// List returns all the rules used to verify requests
-	List(...ListOption) ([]*Rule, error)
+	// Rules returns all the rules used to verify requests
+	Rules(...RulesOption) ([]*Rule, error)
+	// String returns the name of the implementation
+	String() string
 }
 
 // Account provided by an auth provider

auth/jwt/jwt.go

@@ -1,45 +1,35 @@
+// Package jwt is a jwt implementation of the auth interface
 package jwt
 
 import (
 	"sync"
 	"time"
 
-	"github.com/asim/go-micro/v3/auth"
-	"github.com/asim/go-micro/v3/cmd"
-	jwtToken "github.com/asim/go-micro/plugins/auth/jwt/v3/token"
+	"github.com/micro/go-micro/v3/auth"
+	"github.com/micro/go-micro/v3/util/token"
+	"github.com/micro/go-micro/v3/util/token/jwt"
 )
 
-func init() {
-	cmd.DefaultAuths["jwt"] = NewAuth
-}
-
 // NewAuth returns a new instance of the Auth service
 func NewAuth(opts ...auth.Option) auth.Auth {
-	j := new(jwt)
+	j := new(jwtAuth)
 	j.Init(opts...)
 	return j
 }
 
-func NewRules() auth.Rules {
-	return new(jwtRules)
-}
-
-type jwt struct {
-	sync.Mutex
+type jwtAuth struct {
 	options auth.Options
-	jwt     jwtToken.Provider
-}
+	token   token.Provider
+	rules   []*auth.Rule
 
-type jwtRules struct {
 	sync.Mutex
-	rules []*auth.Rule
 }
 
-func (j *jwt) String() string {
+func (j *jwtAuth) String() string {
 	return "jwt"
 }
 
-func (j *jwt) Init(opts ...auth.Option) {
+func (j *jwtAuth) Init(opts ...auth.Option) {
 	j.Lock()
 	defer j.Unlock()
 
@@ -47,31 +37,35 @@ func (j *jwt) Init(opts ...auth.Option) {
 		o(&j.options)
 	}
 
-	j.jwt = jwtToken.New(
-		jwtToken.WithPrivateKey(j.options.PrivateKey),
-		jwtToken.WithPublicKey(j.options.PublicKey),
+	j.token = jwt.NewTokenProvider(
+		token.WithPrivateKey(j.options.PrivateKey),
+		token.WithPublicKey(j.options.PublicKey),
 	)
 }
 
-func (j *jwt) Options() auth.Options {
+func (j *jwtAuth) Options() auth.Options {
 	j.Lock()
 	defer j.Unlock()
 	return j.options
 }
 
-func (j *jwt) Generate(id string, opts ...auth.GenerateOption) (*auth.Account, error) {
+func (j *jwtAuth) Generate(id string, opts ...auth.GenerateOption) (*auth.Account, error) {
 	options := auth.NewGenerateOptions(opts...)
+	if len(options.Issuer) == 0 {
+		options.Issuer = j.Options().Issuer
+	}
+
 	account := &auth.Account{
 		ID:       id,
 		Type:     options.Type,
 		Scopes:   options.Scopes,
 		Metadata: options.Metadata,
-		Issuer:   j.Options().Namespace,
+		Issuer:   options.Issuer,
 	}
 
 	// generate a JWT secret which can be provided to the Token() method
 	// and exchanged for an access token
-	secret, err := j.jwt.Generate(account)
+	secret, err := j.token.Generate(account, token.WithExpiry(time.Hour*24*365))
 	if err != nil {
 		return nil, err
 	}
@@ -81,14 +75,14 @@ func (j *jwt) Generate(id string, opts ...auth.GenerateOption) (*auth.Account, e
 	return account, nil
 }
 
-func (j *jwtRules) Grant(rule *auth.Rule) error {
+func (j *jwtAuth) Grant(rule *auth.Rule) error {
 	j.Lock()
 	defer j.Unlock()
 	j.rules = append(j.rules, rule)
 	return nil
 }
 
-func (j *jwtRules) Revoke(rule *auth.Rule) error {
+func (j *jwtAuth) Revoke(rule *auth.Rule) error {
 	j.Lock()
 	defer j.Unlock()
 
@@ -103,7 +97,7 @@ func (j *jwtRules) Revoke(rule *auth.Rule) error {
 	return nil
 }
 
-func (j *jwtRules) Verify(acc *auth.Account, res *auth.Resource, opts ...auth.VerifyOption) error {
+func (j *jwtAuth) Verify(acc *auth.Account, res *auth.Resource, opts ...auth.VerifyOption) error {
 	j.Lock()
 	defer j.Unlock()
 
@@ -112,20 +106,20 @@ func (j *jwtRules) Verify(acc *auth.Account, res *auth.Resource, opts ...auth.Ve
 		o(&options)
 	}
 
-	return auth.Verify(j.rules, acc, res)
+	return auth.VerifyAccess(j.rules, acc, res)
 }
 
-func (j *jwtRules) List(opts ...auth.ListOption) ([]*auth.Rule, error) {
+func (j *jwtAuth) Rules(opts ...auth.RulesOption) ([]*auth.Rule, error) {
 	j.Lock()
 	defer j.Unlock()
 	return j.rules, nil
 }
 
-func (j *jwt) Inspect(token string) (*auth.Account, error) {
-	return j.jwt.Inspect(token)
+func (j *jwtAuth) Inspect(token string) (*auth.Account, error) {
+	return j.token.Inspect(token)
 }
 
-func (j *jwt) Token(opts ...auth.TokenOption) (*auth.Token, error) {
+func (j *jwtAuth) Token(opts ...auth.TokenOption) (*auth.Token, error) {
 	options := auth.NewTokenOptions(opts...)
 
 	secret := options.RefreshToken
@@ -133,17 +127,17 @@ func (j *jwt) Token(opts ...auth.TokenOption) (*auth.Token, error) {
 		secret = options.Secret
 	}
 
-	account, err := j.jwt.Inspect(secret)
+	account, err := j.token.Inspect(secret)
 	if err != nil {
 		return nil, err
 	}
 
-	access, err := j.jwt.Generate(account, jwtToken.WithExpiry(options.Expiry))
+	access, err := j.token.Generate(account, token.WithExpiry(options.Expiry))
 	if err != nil {
 		return nil, err
 	}
 
-	refresh, err := j.jwt.Generate(account, jwtToken.WithExpiry(options.Expiry+time.Hour))
+	refresh, err := j.token.Generate(account, token.WithExpiry(options.Expiry+time.Hour))
 	if err != nil {
 		return nil, err
 	}

auth/noop.go

@@ -1,91 +0,0 @@
-package auth
-
-import (
-	"github.com/google/uuid"
-)
-
-var (
-	DefaultAuth = NewAuth()
-)
-
-func NewAuth(opts ...Option) Auth {
-	options := Options{}
-
-	for _, o := range opts {
-		o(&options)
-	}
-
-	return &noop{
-		opts: options,
-	}
-}
-
-func NewRules() Rules {
-	return new(noopRules)
-}
-
-type noop struct {
-	opts Options
-}
-
-type noopRules struct{}
-
-// String returns the name of the implementation
-func (n *noop) String() string {
-	return "noop"
-}
-
-// Init the auth
-func (n *noop) Init(opts ...Option) {
-	for _, o := range opts {
-		o(&n.opts)
-	}
-}
-
-// Options set for auth
-func (n *noop) Options() Options {
-	return n.opts
-}
-
-// Generate a new account
-func (n *noop) Generate(id string, opts ...GenerateOption) (*Account, error) {
-	options := NewGenerateOptions(opts...)
-
-	return &Account{
-		ID:       id,
-		Secret:   options.Secret,
-		Metadata: options.Metadata,
-		Scopes:   options.Scopes,
-		Issuer:   n.Options().Namespace,
-	}, nil
-}
-
-// Grant access to a resource
-func (n *noopRules) Grant(rule *Rule) error {
-	return nil
-}
-
-// Revoke access to a resource
-func (n *noopRules) Revoke(rule *Rule) error {
-	return nil
-}
-
-// Rules used to verify requests
-// Verify an account has access to a resource
-func (n *noopRules) Verify(acc *Account, res *Resource, opts ...VerifyOption) error {
-	return nil
-}
-
-func (n *noopRules) List(opts ...ListOption) ([]*Rule, error) {
-	return []*Rule{}, nil
-}
-
-// Inspect a token
-func (n *noop) Inspect(token string) (*Account, error) {
-	return &Account{ID: uuid.New().String(), Issuer: n.Options().Namespace}, nil
-}
-
-// Token generation using an account id and secret
-func (n *noop) Token(opts ...TokenOption) (*Token, error) {
-	return &Token{}, nil
-}

auth/noop/noop.go

@@ -0,0 +1,81 @@
+package noop
+
+import (
+	"github.com/google/uuid"
+	"github.com/micro/go-micro/v3/auth"
+)
+
+func NewAuth(opts ...auth.Option) auth.Auth {
+	var options auth.Options
+	for _, o := range opts {
+		o(&options)
+	}
+
+	return &noop{
+		opts: options,
+	}
+}
+
+type noop struct {
+	opts auth.Options
+}
+
+// String returns the name of the implementation
+func (n *noop) String() string {
+	return "noop"
+}
+
+// Init the auth
+func (n *noop) Init(opts ...auth.Option) {
+	for _, o := range opts {
+		o(&n.opts)
+	}
+}
+
+// Options set for auth
+func (n *noop) Options() auth.Options {
+	return n.opts
+}
+
+// Generate a new account
+func (n *noop) Generate(id string, opts ...auth.GenerateOption) (*auth.Account, error) {
+	options := auth.NewGenerateOptions(opts...)
+
+	return &auth.Account{
+		ID:       id,
+		Secret:   options.Secret,
+		Metadata: options.Metadata,
+		Scopes:   options.Scopes,
+		Issuer:   n.Options().Issuer,
+	}, nil
+}
+
+// Grant access to a resource
+func (n *noop) Grant(rule *auth.Rule) error {
+	return nil
+}
+
+// Revoke access to a resource
+func (n *noop) Revoke(rule *auth.Rule) error {
+	return nil
+}
+
+// Rules used to verify requests
+func (n *noop) Rules(opts ...auth.RulesOption) ([]*auth.Rule, error) {
+	return []*auth.Rule{}, nil
+}
+
+// Verify an account has access to a resource
+func (n *noop) Verify(acc *auth.Account, res *auth.Resource, opts ...auth.VerifyOption) error {
+	return nil
+}
+
+// Inspect a token
+func (n *noop) Inspect(token string) (*auth.Account, error) {
+	return &auth.Account{ID: uuid.New().String(), Issuer: n.Options().Issuer}, nil
+}
+
+// Token generation using an account id and secret
+func (n *noop) Token(opts ...auth.TokenOption) (*auth.Token, error) {
+	return &auth.Token{}, nil
+}

auth/options.go

@@ -3,6 +3,8 @@ package auth
 import (
 	"context"
 	"time"
+
+	"github.com/micro/go-micro/v3/store"
 )
 
 func NewOptions(opts ...Option) Options {
@@ -14,8 +16,8 @@ func NewOptions(opts ...Option) Options {
 }
 
 type Options struct {
-	// Namespace the service belongs to
-	Namespace string
+	// Issuer of the service's account
+	Issuer string
 	// ID is the services auth ID
 	ID string
 	// Secret is used to authenticate the service
@@ -26,8 +28,14 @@ type Options struct {
 	PublicKey string
 	// PrivateKey for encoding JWTs
 	PrivateKey string
+	// LoginURL is the relative url path where a user can login
+	LoginURL string
+	// Store to back auth
+	Store store.Store
 	// Addrs sets the addresses of auth
 	Addrs []string
+	// Context to store other options
+	Context context.Context
 }
 
 type Option func(o *Options)
@@ -39,10 +47,17 @@ func Addrs(addrs ...string) Option {
 	}
 }
 
-// Namespace the service belongs to
-func Namespace(n string) Option {
+// Issuer of the services account
+func Issuer(i string) Option {
 	return func(o *Options) {
-		o.Namespace = n
+		o.Issuer = i
+	}
+}
+
+// Store to back auth
+func Store(s store.Store) Option {
+	return func(o *Options) {
+		o.Store = s
 	}
 }
 
@@ -75,6 +90,13 @@ func ClientToken(token *Token) Option {
 	}
 }
 
+// LoginURL sets the auth LoginURL
+func LoginURL(url string) Option {
+	return func(o *Options) {
+		o.LoginURL = url
+	}
+}
+
 type GenerateOptions struct {
 	// Metadata associated with the account
 	Metadata map[string]string
@@ -86,6 +108,8 @@ type GenerateOptions struct {
 	Type string
 	// Secret used to authenticate the account
 	Secret string
+	// Issuer of the account, e.g. micro
+	Issuer string
 }
 
 type GenerateOption func(o *GenerateOptions)
@@ -125,6 +149,13 @@ func WithScopes(s ...string) GenerateOption {
 	}
 }
 
+// WithIssuer for the generated account
+func WithIssuer(i string) GenerateOption {
+	return func(o *GenerateOptions) {
+		o.Issuer = i
+	}
+}
+
 // NewGenerateOptions from a slice of options
 func NewGenerateOptions(opts ...GenerateOption) GenerateOptions {
 	var options GenerateOptions
@@ -143,6 +174,8 @@ type TokenOptions struct {
 	RefreshToken string
 	// Expiry is the time the token should live for
 	Expiry time.Duration
+	// Issuer of the account
+	Issuer string
 }
 
 type TokenOption func(o *TokenOptions)
@@ -167,6 +200,12 @@ func WithToken(rt string) TokenOption {
 	}
 }
 
+func WithTokenIssuer(iss string) TokenOption {
+	return func(o *TokenOptions) {
+		o.Issuer = iss
+	}
+}
+
 // NewTokenOptions from a slice of options
 func NewTokenOptions(opts ...TokenOption) TokenOptions {
 	var options TokenOptions
@@ -183,7 +222,8 @@ func NewTokenOptions(opts ...TokenOption) TokenOptions {
 }
 
 type VerifyOptions struct {
-	Context context.Context
+	Context   context.Context
+	Namespace string
 }
 
 type VerifyOption func(o *VerifyOptions)
@@ -193,15 +233,27 @@ func VerifyContext(ctx context.Context) VerifyOption {
 		o.Context = ctx
 	}
 }
-
-type ListOptions struct {
-	Context context.Context
+func VerifyNamespace(ns string) VerifyOption {
+	return func(o *VerifyOptions) {
+		o.Namespace = ns
+	}
 }
 
-type ListOption func(o *ListOptions)
+type RulesOptions struct {
+	Context   context.Context
+	Namespace string
+}
 
-func RulesContext(ctx context.Context) ListOption {
-	return func(o *ListOptions) {
+type RulesOption func(o *RulesOptions)
+
+func RulesContext(ctx context.Context) RulesOption {
+	return func(o *RulesOptions) {
 		o.Context = ctx
 	}
 }
+
+func RulesNamespace(ns string) RulesOption {
+	return func(o *RulesOptions) {
+		o.Namespace = ns
+	}
+}

auth/rules.go

@@ -6,10 +6,10 @@ import (
 	"strings"
 )
 
-// Verify an account has access to a resource using the rules provided. If the account does not have
+// VerifyAccess an account has access to a resource using the rules provided. If the account does not have
 // access an error will be returned. If there are no rules provided which match the resource, an error
 // will be returned
-func Verify(rules []*Rule, acc *Account, res *Resource) error {
+func VerifyAccess(rules []*Rule, acc *Account, res *Resource) error {
 	// the rule is only to be applied if the type matches the resource or is catch-all (*)
 	validTypes := []string{"*", res.Type}
 
@@ -83,7 +83,7 @@ func Verify(rules []*Rule, acc *Account, res *Resource) error {
 // not case sensitive.
 func include(slice []string, val string) bool {
 	for _, s := range slice {
-		if strings.EqualFold(s, val) {
+		if strings.ToLower(s) == strings.ToLower(val) {
 			return true
 		}
 	}

auth/rules_test.go

@@ -280,7 +280,7 @@ func TestVerify(t *testing.T) {
 
 	for _, tc := range tt {
 		t.Run(tc.Name, func(t *testing.T) {
-			if err := Verify(tc.Rules, tc.Account, tc.Resource); err != tc.Error {
+			if err := VerifyAccess(tc.Rules, tc.Account, tc.Resource); err != tc.Error {
 				t.Errorf("Expected %v but got %v", tc.Error, err)
 			}
 		})

broker/broker.go

@@ -14,54 +14,18 @@ type Broker interface {
 }
 
 // Handler is used to process messages via a subscription of a topic.
-// The handler is passed a publication interface which contains the
-// message and optional Ack method to acknowledge receipt of the message.
-type Handler func(Event) error
+type Handler func(*Message) error
+
+type ErrorHandler func(*Message, error)
 
 type Message struct {
 	Header map[string]string
 	Body   []byte
 }
 
-// Event is given to a subscription handler for processing
-type Event interface {
-	Topic() string
-	Message() *Message
-	Ack() error
-	Error() error
-}
-
 // Subscriber is a convenience return type for the Subscribe method
 type Subscriber interface {
 	Options() SubscribeOptions
 	Topic() string
 	Unsubscribe() error
 }
-
-var (
-	DefaultBroker Broker = NewBroker()
-)
-
-func Init(opts ...Option) error {
-	return DefaultBroker.Init(opts...)
-}
-
-func Connect() error {
-	return DefaultBroker.Connect()
-}
-
-func Disconnect() error {
-	return DefaultBroker.Disconnect()
-}
-
-func Publish(topic string, msg *Message, opts ...PublishOption) error {
-	return DefaultBroker.Publish(topic, msg, opts...)
-}
-
-func Subscribe(topic string, handler Handler, opts ...SubscribeOption) (Subscriber, error) {
-	return DefaultBroker.Subscribe(topic, handler, opts...)
-}
-
-func String() string {
-	return DefaultBroker.String()
-}

broker/http.go

@@ -1,711 +0,0 @@
-// Package http provides a http based message broker
-package broker
-
-import (
-	"bytes"
-	"context"
-	"crypto/tls"
-	"errors"
-	"fmt"
-	"io"
-	"io/ioutil"
-	"math/rand"
-	"net"
-	"net/http"
-	"net/url"
-	"runtime"
-	"sync"
-	"time"
-
-	"github.com/asim/go-micro/v3/codec/json"
-	merr "github.com/asim/go-micro/v3/errors"
-	"github.com/asim/go-micro/v3/registry"
-	"github.com/asim/go-micro/v3/registry/cache"
-	maddr "github.com/asim/go-micro/v3/util/addr"
-	mnet "github.com/asim/go-micro/v3/util/net"
-	mls "github.com/asim/go-micro/v3/util/tls"
-	"github.com/google/uuid"
-	"golang.org/x/net/http2"
-)
-
-// HTTP Broker is a point to point async broker
-type httpBroker struct {
-	id      string
-	address string
-	opts    Options
-
-	mux *http.ServeMux
-
-	c *http.Client
-	r registry.Registry
-
-	sync.RWMutex
-	subscribers map[string][]*httpSubscriber
-	running     bool
-	exit        chan chan error
-
-	// offline message inbox
-	mtx   sync.RWMutex
-	inbox map[string][][]byte
-}
-
-type httpSubscriber struct {
-	opts  SubscribeOptions
-	id    string
-	topic string
-	fn    Handler
-	svc   *registry.Service
-	hb    *httpBroker
-}
-
-type httpEvent struct {
-	m   *Message
-	t   string
-	err error
-}
-
-var (
-	DefaultPath      = "/"
-	DefaultAddress   = "127.0.0.1:0"
-	serviceName      = "micro.http.broker"
-	broadcastVersion = "ff.http.broadcast"
-	registerTTL      = time.Minute
-	registerInterval = time.Second * 30
-)
-
-func init() {
-	rand.Seed(time.Now().Unix())
-}
-
-func newTransport(config *tls.Config) *http.Transport {
-	if config == nil {
-		config = &tls.Config{
-			InsecureSkipVerify: true,
-		}
-	}
-
-	dialTLS := func(network string, addr string) (net.Conn, error) {
-		return tls.Dial(network, addr, config)
-	}
-
-	t := &http.Transport{
-		Proxy: http.ProxyFromEnvironment,
-		Dial: (&net.Dialer{
-			Timeout:   30 * time.Second,
-			KeepAlive: 30 * time.Second,
-		}).Dial,
-		TLSHandshakeTimeout: 10 * time.Second,
-		DialTLS:             dialTLS,
-	}
-	runtime.SetFinalizer(&t, func(tr **http.Transport) {
-		(*tr).CloseIdleConnections()
-	})
-
-	// setup http2
-	http2.ConfigureTransport(t)
-
-	return t
-}
-
-func newHttpBroker(opts ...Option) Broker {
-	options := Options{
-		Codec:    json.Marshaler{},
-		Context:  context.TODO(),
-		Registry: registry.DefaultRegistry,
-	}
-
-	for _, o := range opts {
-		o(&options)
-	}
-
-	// set address
-	addr := DefaultAddress
-
-	if len(options.Addrs) > 0 && len(options.Addrs[0]) > 0 {
-		addr = options.Addrs[0]
-	}
-
-	h := &httpBroker{
-		id:          uuid.New().String(),
-		address:     addr,
-		opts:        options,
-		r:           options.Registry,
-		c:           &http.Client{Transport: newTransport(options.TLSConfig)},
-		subscribers: make(map[string][]*httpSubscriber),
-		exit:        make(chan chan error),
-		mux:         http.NewServeMux(),
-		inbox:       make(map[string][][]byte),
-	}
-
-	// specify the message handler
-	h.mux.Handle(DefaultPath, h)
-
-	// get optional handlers
-	if h.opts.Context != nil {
-		handlers, ok := h.opts.Context.Value("http_handlers").(map[string]http.Handler)
-		if ok {
-			for pattern, handler := range handlers {
-				h.mux.Handle(pattern, handler)
-			}
-		}
-	}
-
-	return h
-}
-
-func (h *httpEvent) Ack() error {
-	return nil
-}
-
-func (h *httpEvent) Error() error {
-	return h.err
-}
-
-func (h *httpEvent) Message() *Message {
-	return h.m
-}
-
-func (h *httpEvent) Topic() string {
-	return h.t
-}
-
-func (h *httpSubscriber) Options() SubscribeOptions {
-	return h.opts
-}
-
-func (h *httpSubscriber) Topic() string {
-	return h.topic
-}
-
-func (h *httpSubscriber) Unsubscribe() error {
-	return h.hb.unsubscribe(h)
-}
-
-func (h *httpBroker) saveMessage(topic string, msg []byte) {
-	h.mtx.Lock()
-	defer h.mtx.Unlock()
-
-	// get messages
-	c := h.inbox[topic]
-
-	// save message
-	c = append(c, msg)
-
-	// max length 64
-	if len(c) > 64 {
-		c = c[:64]
-	}
-
-	// save inbox
-	h.inbox[topic] = c
-}
-
-func (h *httpBroker) getMessage(topic string, num int) [][]byte {
-	h.mtx.Lock()
-	defer h.mtx.Unlock()
-
-	// get messages
-	c, ok := h.inbox[topic]
-	if !ok {
-		return nil
-	}
-
-	// more message than requests
-	if len(c) >= num {
-		msg := c[:num]
-		h.inbox[topic] = c[num:]
-		return msg
-	}
-
-	// reset inbox
-	h.inbox[topic] = nil
-
-	// return all messages
-	return c
-}
-
-func (h *httpBroker) subscribe(s *httpSubscriber) error {
-	h.Lock()
-	defer h.Unlock()
-
-	if err := h.r.Register(s.svc, registry.RegisterTTL(registerTTL)); err != nil {
-		return err
-	}
-
-	h.subscribers[s.topic] = append(h.subscribers[s.topic], s)
-	return nil
-}
-
-func (h *httpBroker) unsubscribe(s *httpSubscriber) error {
-	h.Lock()
-	defer h.Unlock()
-
-	//nolint:prealloc
-	var subscribers []*httpSubscriber
-
-	// look for subscriber
-	for _, sub := range h.subscribers[s.topic] {
-		// deregister and skip forward
-		if sub == s {
-			_ = h.r.Deregister(sub.svc)
-			continue
-		}
-		// keep subscriber
-		subscribers = append(subscribers, sub)
-	}
-
-	// set subscribers
-	h.subscribers[s.topic] = subscribers
-
-	return nil
-}
-
-func (h *httpBroker) run(l net.Listener) {
-	t := time.NewTicker(registerInterval)
-	defer t.Stop()
-
-	for {
-		select {
-		// heartbeat for each subscriber
-		case <-t.C:
-			h.RLock()
-			for _, subs := range h.subscribers {
-				for _, sub := range subs {
-					_ = h.r.Register(sub.svc, registry.RegisterTTL(registerTTL))
-				}
-			}
-			h.RUnlock()
-		// received exit signal
-		case ch := <-h.exit:
-			ch <- l.Close()
-			h.RLock()
-			for _, subs := range h.subscribers {
-				for _, sub := range subs {
-					_ = h.r.Deregister(sub.svc)
-				}
-			}
-			h.RUnlock()
-			return
-		}
-	}
-}
-
-func (h *httpBroker) ServeHTTP(w http.ResponseWriter, req *http.Request) {
-	if req.Method != "POST" {
-		err := merr.BadRequest("go.micro.broker", "Method not allowed")
-		http.Error(w, err.Error(), http.StatusMethodNotAllowed)
-		return
-	}
-	defer req.Body.Close()
-
-	req.ParseForm()
-
-	b, err := ioutil.ReadAll(req.Body)
-	if err != nil {
-		errr := merr.InternalServerError("go.micro.broker", "Error reading request body: %v", err)
-		w.WriteHeader(500)
-		w.Write([]byte(errr.Error()))
-		return
-	}
-
-	var m *Message
-	if err = h.opts.Codec.Unmarshal(b, &m); err != nil {
-		errr := merr.InternalServerError("go.micro.broker", "Error parsing request body: %v", err)
-		w.WriteHeader(500)
-		w.Write([]byte(errr.Error()))
-		return
-	}
-
-	topic := m.Header["Micro-Topic"]
-	//delete(m.Header, ":topic")
-
-	if len(topic) == 0 {
-		errr := merr.InternalServerError("go.micro.broker", "Topic not found")
-		w.WriteHeader(500)
-		w.Write([]byte(errr.Error()))
-		return
-	}
-
-	p := &httpEvent{m: m, t: topic}
-	id := req.Form.Get("id")
-
-	//nolint:prealloc
-	var subs []Handler
-
-	h.RLock()
-	for _, subscriber := range h.subscribers[topic] {
-		if id != subscriber.id {
-			continue
-		}
-		subs = append(subs, subscriber.fn)
-	}
-	h.RUnlock()
-
-	// execute the handler
-	for _, fn := range subs {
-		p.err = fn(p)
-	}
-}
-
-func (h *httpBroker) Address() string {
-	h.RLock()
-	defer h.RUnlock()
-	return h.address
-}
-
-func (h *httpBroker) Connect() error {
-	h.RLock()
-	if h.running {
-		h.RUnlock()
-		return nil
-	}
-	h.RUnlock()
-
-	h.Lock()
-	defer h.Unlock()
-
-	var l net.Listener
-	var err error
-
-	if h.opts.Secure || h.opts.TLSConfig != nil {
-		config := h.opts.TLSConfig
-
-		fn := func(addr string) (net.Listener, error) {
-			if config == nil {
-				hosts := []string{addr}
-
-				// check if its a valid host:port
-				if host, _, err := net.SplitHostPort(addr); err == nil {
-					if len(host) == 0 {
-						hosts = maddr.IPs()
-					} else {
-						hosts = []string{host}
-					}
-				}
-
-				// generate a certificate
-				cert, err := mls.Certificate(hosts...)
-				if err != nil {
-					return nil, err
-				}
-				config = &tls.Config{Certificates: []tls.Certificate{cert}}
-			}
-			return tls.Listen("tcp", addr, config)
-		}
-
-		l, err = mnet.Listen(h.address, fn)
-	} else {
-		fn := func(addr string) (net.Listener, error) {
-			return net.Listen("tcp", addr)
-		}
-
-		l, err = mnet.Listen(h.address, fn)
-	}
-
-	if err != nil {
-		return err
-	}
-
-	addr := h.address
-	h.address = l.Addr().String()
-
-	go http.Serve(l, h.mux)
-	go func() {
-		h.run(l)
-		h.Lock()
-		h.opts.Addrs = []string{addr}
-		h.address = addr
-		h.Unlock()
-	}()
-
-	// get registry
-	reg := h.opts.Registry
-	if reg == nil {
-		reg = registry.DefaultRegistry
-	}
-	// set cache
-	h.r = cache.New(reg)
-
-	// set running
-	h.running = true
-	return nil
-}
-
-func (h *httpBroker) Disconnect() error {
-	h.RLock()
-	if !h.running {
-		h.RUnlock()
-		return nil
-	}
-	h.RUnlock()
-
-	h.Lock()
-	defer h.Unlock()
-
-	// stop cache
-	rc, ok := h.r.(cache.Cache)
-	if ok {
-		rc.Stop()
-	}
-
-	// exit and return err
-	ch := make(chan error)
-	h.exit <- ch
-	err := <-ch
-
-	// set not running
-	h.running = false
-	return err
-}
-
-func (h *httpBroker) Init(opts ...Option) error {
-	h.RLock()
-	if h.running {
-		h.RUnlock()
-		return errors.New("cannot init while connected")
-	}
-	h.RUnlock()
-
-	h.Lock()
-	defer h.Unlock()
-
-	for _, o := range opts {
-		o(&h.opts)
-	}
-
-	if len(h.opts.Addrs) > 0 && len(h.opts.Addrs[0]) > 0 {
-		h.address = h.opts.Addrs[0]
-	}
-
-	if len(h.id) == 0 {
-		h.id = "go.micro.http.broker-" + uuid.New().String()
-	}
-
-	// get registry
-	reg := h.opts.Registry
-	if reg == nil {
-		reg = registry.DefaultRegistry
-	}
-
-	// get cache
-	if rc, ok := h.r.(cache.Cache); ok {
-		rc.Stop()
-	}
-
-	// set registry
-	h.r = cache.New(reg)
-
-	// reconfigure tls config
-	if c := h.opts.TLSConfig; c != nil {
-		h.c = &http.Client{
-			Transport: newTransport(c),
-		}
-	}
-
-	return nil
-}
-
-func (h *httpBroker) Options() Options {
-	return h.opts
-}
-
-func (h *httpBroker) Publish(topic string, msg *Message, opts ...PublishOption) error {
-	// create the message first
-	m := &Message{
-		Header: make(map[string]string),
-		Body:   msg.Body,
-	}
-
-	for k, v := range msg.Header {
-		m.Header[k] = v
-	}
-
-	m.Header["Micro-Topic"] = topic
-
-	// encode the message
-	b, err := h.opts.Codec.Marshal(m)
-	if err != nil {
-		return err
-	}
-
-	// save the message
-	h.saveMessage(topic, b)
-
-	// now attempt to get the service
-	h.RLock()
-	s, err := h.r.GetService(serviceName)
-	if err != nil {
-		h.RUnlock()
-		return err
-	}
-	h.RUnlock()
-
-	pub := func(node *registry.Node, t string, b []byte) error {
-		scheme := "http"
-
-		// check if secure is added in metadata
-		if node.Metadata["secure"] == "true" {
-			scheme = "https"
-		}
-
-		vals := url.Values{}
-		vals.Add("id", node.Id)
-
-		uri := fmt.Sprintf("%s://%s%s?%s", scheme, node.Address, DefaultPath, vals.Encode())
-		r, err := h.c.Post(uri, "application/json", bytes.NewReader(b))
-		if err != nil {
-			return err
-		}
-
-		// discard response body
-		io.Copy(ioutil.Discard, r.Body)
-		r.Body.Close()
-		return nil
-	}
-
-	srv := func(s []*registry.Service, b []byte) {
-		for _, service := range s {
-			var nodes []*registry.Node
-
-			for _, node := range service.Nodes {
-				// only use nodes tagged with broker http
-				if node.Metadata["broker"] != "http" {
-					continue
-				}
-
-				// look for nodes for the topic
-				if node.Metadata["topic"] != topic {
-					continue
-				}
-
-				nodes = append(nodes, node)
-			}
-
-			// only process if we have nodes
-			if len(nodes) == 0 {
-				continue
-			}
-
-			switch service.Version {
-			// broadcast version means broadcast to all nodes
-			case broadcastVersion:
-				var success bool
-
-				// publish to all nodes
-				for _, node := range nodes {
-					// publish async
-					if err := pub(node, topic, b); err == nil {
-						success = true
-					}
-				}
-
-				// save if it failed to publish at least once
-				if !success {
-					h.saveMessage(topic, b)
-				}
-			default:
-				// select node to publish to
-				node := nodes[rand.Int()%len(nodes)]
-
-				// publish async to one node
-				if err := pub(node, topic, b); err != nil {
-					// if failed save it
-					h.saveMessage(topic, b)
-				}
-			}
-		}
-	}
-
-	// do the rest async
-	go func() {
-		// get a third of the backlog
-		messages := h.getMessage(topic, 8)
-		delay := (len(messages) > 1)
-
-		// publish all the messages
-		for _, msg := range messages {
-			// serialize here
-			srv(s, msg)
-
-			// sending a backlog of messages
-			if delay {
-				time.Sleep(time.Millisecond * 100)
-			}
-		}
-	}()
-
-	return nil
-}
-
-func (h *httpBroker) Subscribe(topic string, handler Handler, opts ...SubscribeOption) (Subscriber, error) {
-	var err error
-	var host, port string
-	options := NewSubscribeOptions(opts...)
-
-	// parse address for host, port
-	host, port, err = net.SplitHostPort(h.Address())
-	if err != nil {
-		return nil, err
-	}
-
-	addr, err := maddr.Extract(host)
-	if err != nil {
-		return nil, err
-	}
-
-	var secure bool
-
-	if h.opts.Secure || h.opts.TLSConfig != nil {
-		secure = true
-	}
-
-	// register service
-	node := &registry.Node{
-		Id:      topic + "-" + h.id,
-		Address: mnet.HostPort(addr, port),
-		Metadata: map[string]string{
-			"secure": fmt.Sprintf("%t", secure),
-			"broker": "http",
-			"topic":  topic,
-		},
-	}
-
-	// check for queue group or broadcast queue
-	version := options.Queue
-	if len(version) == 0 {
-		version = broadcastVersion
-	}
-
-	service := &registry.Service{
-		Name:    serviceName,
-		Version: version,
-		Nodes:   []*registry.Node{node},
-	}
-
-	// generate subscriber
-	subscriber := &httpSubscriber{
-		opts:  options,
-		hb:    h,
-		id:    node.Id,
-		topic: topic,
-		fn:    handler,
-		svc:   service,
-	}
-
-	// subscribe now
-	if err := h.subscribe(subscriber); err != nil {
-		return nil, err
-	}
-
-	// return the subscriber
-	return subscriber, nil
-}
-
-func (h *httpBroker) String() string {
-	return "http"
-}
-
-// NewBroker returns a new http broker
-func NewBroker(opts ...Option) Broker {
-	return newHttpBroker(opts...)
-}

broker/http/http.go

@@ -18,22 +18,18 @@ import (
 	"time"
 
 	"github.com/google/uuid"
-	"github.com/asim/go-micro/v3/broker"
-	"github.com/asim/go-micro/v3/codec/json"
-	"github.com/asim/go-micro/v3/cmd"
-	merr "github.com/asim/go-micro/v3/errors"
-	"github.com/asim/go-micro/v3/registry"
-	"github.com/asim/go-micro/v3/registry/cache"
-	maddr "github.com/asim/go-micro/v3/util/addr"
-	mnet "github.com/asim/go-micro/v3/util/net"
-	mls "github.com/asim/go-micro/v3/util/tls"
+	"github.com/micro/go-micro/v3/broker"
+	"github.com/micro/go-micro/v3/codec/json"
+	merr "github.com/micro/go-micro/v3/errors"
+	"github.com/micro/go-micro/v3/registry"
+	"github.com/micro/go-micro/v3/registry/cache"
+	"github.com/micro/go-micro/v3/registry/mdns"
+	maddr "github.com/micro/go-micro/v3/util/addr"
+	mnet "github.com/micro/go-micro/v3/util/net"
+	mls "github.com/micro/go-micro/v3/util/tls"
 	"golang.org/x/net/http2"
 )
 
-func init() {
-	cmd.DefaultBrokers["http"] = NewBroker
-}
-
 // HTTP Broker is a point to point async broker
 type httpBroker struct {
 	id      string
@@ -64,15 +60,10 @@ type httpSubscriber struct {
 	hb    *httpBroker
 }
 
-type httpEvent struct {
-	m   *broker.Message
-	t   string
-	err error
-}
-
 var (
-	DefaultSubPath   = "/_sub"
-	serviceName      = "go.micro.http.broker"
+	DefaultPath      = "/"
+	DefaultAddress   = "127.0.0.1:0"
+	serviceName      = "micro.http.broker"
 	broadcastVersion = "ff.http.broadcast"
 	registerTTL      = time.Minute
 	registerInterval = time.Second * 30
@@ -116,7 +107,7 @@ func newHttpBroker(opts ...broker.Option) broker.Broker {
 	options := broker.Options{
 		Codec:    json.Marshaler{},
 		Context:  context.TODO(),
-		Registry: registry.DefaultRegistry,
+		Registry: mdns.NewRegistry(),
 	}
 
 	for _, o := range opts {
@@ -124,7 +115,8 @@ func newHttpBroker(opts ...broker.Option) broker.Broker {
 	}
 
 	// set address
-	addr := ":0"
+	addr := DefaultAddress
+
 	if len(options.Addrs) > 0 && len(options.Addrs[0]) > 0 {
 		addr = options.Addrs[0]
 	}
@@ -142,7 +134,7 @@ func newHttpBroker(opts ...broker.Option) broker.Broker {
 	}
 
 	// specify the message handler
-	h.mux.Handle(DefaultSubPath, h)
+	h.mux.Handle(DefaultPath, h)
 
 	// get optional handlers
 	if h.opts.Context != nil {
@@ -157,22 +149,6 @@ func newHttpBroker(opts ...broker.Option) broker.Broker {
 	return h
 }
 
-func (h *httpEvent) Ack() error {
-	return nil
-}
-
-func (h *httpEvent) Error() error {
-	return h.err
-}
-
-func (h *httpEvent) Message() *broker.Message {
-	return h.m
-}
-
-func (h *httpEvent) Topic() string {
-	return h.t
-}
-
 func (h *httpSubscriber) Options() broker.SubscribeOptions {
 	return h.opts
 }
@@ -312,16 +288,15 @@ func (h *httpBroker) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 		return
 	}
 
-	var m *broker.Message
-	if err = h.opts.Codec.Unmarshal(b, &m); err != nil {
+	var msg *broker.Message
+	if err = h.opts.Codec.Unmarshal(b, &msg); err != nil {
 		errr := merr.InternalServerError("go.micro.broker", "Error parsing request body: %v", err)
 		w.WriteHeader(500)
 		w.Write([]byte(errr.Error()))
 		return
 	}
 
-	topic := m.Header[":topic"]
-	delete(m.Header, ":topic")
+	topic := msg.Header["Micro-Topic"]
 
 	if len(topic) == 0 {
 		errr := merr.InternalServerError("go.micro.broker", "Topic not found")
@@ -330,7 +305,6 @@ func (h *httpBroker) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 		return
 	}
 
-	p := &httpEvent{m: m, t: topic}
 	id := req.Form.Get("id")
 
 	//nolint:prealloc
@@ -347,7 +321,7 @@ func (h *httpBroker) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 
 	// execute the handler
 	for _, fn := range subs {
-		p.err = fn(p)
+		fn(msg)
 	}
 }
 
@@ -425,7 +399,7 @@ func (h *httpBroker) Connect() error {
 	// get registry
 	reg := h.opts.Registry
 	if reg == nil {
-		reg = registry.DefaultRegistry
+		reg = mdns.NewRegistry()
 	}
 	// set cache
 	h.r = cache.New(reg)
@@ -488,7 +462,7 @@ func (h *httpBroker) Init(opts ...broker.Option) error {
 	// get registry
 	reg := h.opts.Registry
 	if reg == nil {
-		reg = registry.DefaultRegistry
+		reg = mdns.NewRegistry()
 	}
 
 	// get cache
@@ -524,7 +498,7 @@ func (h *httpBroker) Publish(topic string, msg *broker.Message, opts ...broker.P
 		m.Header[k] = v
 	}
 
-	m.Header[":topic"] = topic
+	m.Header["Micro-Topic"] = topic
 
 	// encode the message
 	b, err := h.opts.Codec.Marshal(m)
@@ -555,7 +529,7 @@ func (h *httpBroker) Publish(topic string, msg *broker.Message, opts ...broker.P
 		vals := url.Values{}
 		vals.Add("id", node.Id)
 
-		uri := fmt.Sprintf("%s://%s%s?%s", scheme, node.Address, DefaultSubPath, vals.Encode())
+		uri := fmt.Sprintf("%s://%s%s?%s", scheme, node.Address, DefaultPath, vals.Encode())
 		r, err := h.c.Post(uri, "application/json", bytes.NewReader(b))
 		if err != nil {
 			return err

broker/http/http_test.go

@@ -6,9 +6,9 @@ import (
 	"time"
 
 	"github.com/google/uuid"
-	"github.com/asim/go-micro/v3/broker"
-	"github.com/asim/go-micro/v3/registry"
-	"github.com/asim/go-micro/plugins/registry/memory/v3"
+	"github.com/micro/go-micro/v3/broker"
+	"github.com/micro/go-micro/v3/registry"
+	"github.com/micro/go-micro/v3/registry/memory"
 )
 
 var (
@@ -83,9 +83,8 @@ func sub(be *testing.B, c int) {
 	done := make(chan bool, c)
 
 	for i := 0; i < c; i++ {
-		sub, err := b.Subscribe(topic, func(p broker.Event) error {
+		sub, err := b.Subscribe(topic, func(m *broker.Message) error {
 			done <- true
-			m := p.Message()
 
 			if string(m.Body) != string(msg.Body) {
 				be.Fatalf("Unexpected msg %s, expected %s", string(m.Body), string(msg.Body))
@@ -140,9 +139,8 @@ func pub(be *testing.B, c int) {
 
 	done := make(chan bool, c*4)
 
-	sub, err := b.Subscribe(topic, func(p broker.Event) error {
+	sub, err := b.Subscribe(topic, func(m *broker.Message) error {
 		done <- true
-		m := p.Message()
 		if string(m.Body) != string(msg.Body) {
 			be.Fatalf("Unexpected msg %s, expected %s", string(m.Body), string(msg.Body))
 		}
@@ -208,8 +206,7 @@ func TestBroker(t *testing.T) {
 
 	done := make(chan bool)
 
-	sub, err := b.Subscribe("test", func(p broker.Event) error {
-		m := p.Message()
+	sub, err := b.Subscribe("test", func(m *broker.Message) error {
 
 		if string(m.Body) != string(msg.Body) {
 			t.Fatalf("Unexpected msg %s, expected %s", string(m.Body), string(msg.Body))
@@ -257,11 +254,9 @@ func TestConcurrentSubBroker(t *testing.T) {
 	var wg sync.WaitGroup
 
 	for i := 0; i < 10; i++ {
-		sub, err := b.Subscribe("test", func(p broker.Event) error {
+		sub, err := b.Subscribe("test", func(m *broker.Message) error {
 			defer wg.Done()
 
-			m := p.Message()
-
 			if string(m.Body) != string(msg.Body) {
 				t.Fatalf("Unexpected msg %s, expected %s", string(m.Body), string(msg.Body))
 			}
@@ -312,11 +307,9 @@ func TestConcurrentPubBroker(t *testing.T) {
 
 	var wg sync.WaitGroup
 
-	sub, err := b.Subscribe("test", func(p broker.Event) error {
+	sub, err := b.Subscribe("test", func(m *broker.Message) error {
 		defer wg.Done()
 
-		m := p.Message()
-
 		if string(m.Body) != string(msg.Body) {
 			t.Fatalf("Unexpected msg %s, expected %s", string(m.Body), string(msg.Body))
 		}

broker/http/options.go

@@ -0,0 +1,23 @@
+package http
+
+import (
+	"context"
+	"net/http"
+
+	"github.com/micro/go-micro/v3/broker"
+)
+
+// Handle registers the handler for the given pattern.
+func Handle(pattern string, handler http.Handler) broker.Option {
+	return func(o *broker.Options) {
+		if o.Context == nil {
+			o.Context = context.Background()
+		}
+		handlers, ok := o.Context.Value("http_handlers").(map[string]http.Handler)
+		if !ok {
+			handlers = make(map[string]http.Handler)
+		}
+		handlers[pattern] = handler
+		o.Context = context.WithValue(o.Context, "http_handlers", handlers)
+	}
+}

broker/http_test.go

@@ -1,383 +0,0 @@
-package broker_test
-
-import (
-	"sync"
-	"testing"
-	"time"
-
-	"github.com/asim/go-micro/v3/broker"
-	"github.com/asim/go-micro/v3/registry"
-	"github.com/google/uuid"
-)
-
-var (
-	// mock data
-	testData = map[string][]*registry.Service{
-		"foo": {
-			{
-				Name:    "foo",
-				Version: "1.0.0",
-				Nodes: []*registry.Node{
-					{
-						Id:      "foo-1.0.0-123",
-						Address: "localhost:9999",
-					},
-					{
-						Id:      "foo-1.0.0-321",
-						Address: "localhost:9999",
-					},
-				},
-			},
-			{
-				Name:    "foo",
-				Version: "1.0.1",
-				Nodes: []*registry.Node{
-					{
-						Id:      "foo-1.0.1-321",
-						Address: "localhost:6666",
-					},
-				},
-			},
-			{
-				Name:    "foo",
-				Version: "1.0.3",
-				Nodes: []*registry.Node{
-					{
-						Id:      "foo-1.0.3-345",
-						Address: "localhost:8888",
-					},
-				},
-			},
-		},
-	}
-)
-
-func newTestRegistry() registry.Registry {
-	return registry.NewMemoryRegistry(registry.Services(testData))
-}
-
-func sub(be *testing.B, c int) {
-	be.StopTimer()
-	m := newTestRegistry()
-
-	b := broker.NewBroker(broker.Registry(m))
-	topic := uuid.New().String()
-
-	if err := b.Init(); err != nil {
-		be.Fatalf("Unexpected init error: %v", err)
-	}
-
-	if err := b.Connect(); err != nil {
-		be.Fatalf("Unexpected connect error: %v", err)
-	}
-
-	msg := &broker.Message{
-		Header: map[string]string{
-			"Content-Type": "application/json",
-		},
-		Body: []byte(`{"message": "Hello World"}`),
-	}
-
-	var subs []broker.Subscriber
-	done := make(chan bool, c)
-
-	for i := 0; i < c; i++ {
-		sub, err := b.Subscribe(topic, func(p broker.Event) error {
-			done <- true
-			m := p.Message()
-
-			if string(m.Body) != string(msg.Body) {
-				be.Fatalf("Unexpected msg %s, expected %s", string(m.Body), string(msg.Body))
-			}
-
-			return nil
-		}, broker.Queue("shared"))
-		if err != nil {
-			be.Fatalf("Unexpected subscribe error: %v", err)
-		}
-		subs = append(subs, sub)
-	}
-
-	for i := 0; i < be.N; i++ {
-		be.StartTimer()
-		if err := b.Publish(topic, msg); err != nil {
-			be.Fatalf("Unexpected publish error: %v", err)
-		}
-		<-done
-		be.StopTimer()
-	}
-
-	for _, sub := range subs {
-		sub.Unsubscribe()
-	}
-
-	if err := b.Disconnect(); err != nil {
-		be.Fatalf("Unexpected disconnect error: %v", err)
-	}
-}
-
-func pub(be *testing.B, c int) {
-	be.StopTimer()
-	m := newTestRegistry()
-	b := broker.NewBroker(broker.Registry(m))
-	topic := uuid.New().String()
-
-	if err := b.Init(); err != nil {
-		be.Fatalf("Unexpected init error: %v", err)
-	}
-
-	if err := b.Connect(); err != nil {
-		be.Fatalf("Unexpected connect error: %v", err)
-	}
-
-	msg := &broker.Message{
-		Header: map[string]string{
-			"Content-Type": "application/json",
-		},
-		Body: []byte(`{"message": "Hello World"}`),
-	}
-
-	done := make(chan bool, c*4)
-
-	sub, err := b.Subscribe(topic, func(p broker.Event) error {
-		done <- true
-		m := p.Message()
-		if string(m.Body) != string(msg.Body) {
-			be.Fatalf("Unexpected msg %s, expected %s", string(m.Body), string(msg.Body))
-		}
-		return nil
-	}, broker.Queue("shared"))
-	if err != nil {
-		be.Fatalf("Unexpected subscribe error: %v", err)
-	}
-
-	var wg sync.WaitGroup
-	ch := make(chan int, c*4)
-	be.StartTimer()
-
-	for i := 0; i < c; i++ {
-		go func() {
-			for range ch {
-				if err := b.Publish(topic, msg); err != nil {
-					be.Fatalf("Unexpected publish error: %v", err)
-				}
-				select {
-				case <-done:
-				case <-time.After(time.Second):
-				}
-				wg.Done()
-			}
-		}()
-	}
-
-	for i := 0; i < be.N; i++ {
-		wg.Add(1)
-		ch <- i
-	}
-
-	wg.Wait()
-	be.StopTimer()
-	sub.Unsubscribe()
-	close(ch)
-	close(done)
-
-	if err := b.Disconnect(); err != nil {
-		be.Fatalf("Unexpected disconnect error: %v", err)
-	}
-}
-
-func TestBroker(t *testing.T) {
-	m := newTestRegistry()
-	b := broker.NewBroker(broker.Registry(m))
-
-	if err := b.Init(); err != nil {
-		t.Fatalf("Unexpected init error: %v", err)
-	}
-
-	if err := b.Connect(); err != nil {
-		t.Fatalf("Unexpected connect error: %v", err)
-	}
-
-	msg := &broker.Message{
-		Header: map[string]string{
-			"Content-Type": "application/json",
-		},
-		Body: []byte(`{"message": "Hello World"}`),
-	}
-
-	done := make(chan bool)
-
-	sub, err := b.Subscribe("test", func(p broker.Event) error {
-		m := p.Message()
-
-		if string(m.Body) != string(msg.Body) {
-			t.Fatalf("Unexpected msg %s, expected %s", string(m.Body), string(msg.Body))
-		}
-
-		close(done)
-		return nil
-	})
-	if err != nil {
-		t.Fatalf("Unexpected subscribe error: %v", err)
-	}
-
-	if err := b.Publish("test", msg); err != nil {
-		t.Fatalf("Unexpected publish error: %v", err)
-	}
-
-	<-done
-	sub.Unsubscribe()
-
-	if err := b.Disconnect(); err != nil {
-		t.Fatalf("Unexpected disconnect error: %v", err)
-	}
-}
-
-func TestConcurrentSubBroker(t *testing.T) {
-	m := newTestRegistry()
-	b := broker.NewBroker(broker.Registry(m))
-
-	if err := b.Init(); err != nil {
-		t.Fatalf("Unexpected init error: %v", err)
-	}
-
-	if err := b.Connect(); err != nil {
-		t.Fatalf("Unexpected connect error: %v", err)
-	}
-
-	msg := &broker.Message{
-		Header: map[string]string{
-			"Content-Type": "application/json",
-		},
-		Body: []byte(`{"message": "Hello World"}`),
-	}
-
-	var subs []broker.Subscriber
-	var wg sync.WaitGroup
-
-	for i := 0; i < 10; i++ {
-		sub, err := b.Subscribe("test", func(p broker.Event) error {
-			defer wg.Done()
-
-			m := p.Message()
-
-			if string(m.Body) != string(msg.Body) {
-				t.Fatalf("Unexpected msg %s, expected %s", string(m.Body), string(msg.Body))
-			}
-
-			return nil
-		})
-		if err != nil {
-			t.Fatalf("Unexpected subscribe error: %v", err)
-		}
-
-		wg.Add(1)
-		subs = append(subs, sub)
-	}
-
-	if err := b.Publish("test", msg); err != nil {
-		t.Fatalf("Unexpected publish error: %v", err)
-	}
-
-	wg.Wait()
-
-	for _, sub := range subs {
-		sub.Unsubscribe()
-	}
-
-	if err := b.Disconnect(); err != nil {
-		t.Fatalf("Unexpected disconnect error: %v", err)
-	}
-}
-
-func TestConcurrentPubBroker(t *testing.T) {
-	m := newTestRegistry()
-	b := broker.NewBroker(broker.Registry(m))
-
-	if err := b.Init(); err != nil {
-		t.Fatalf("Unexpected init error: %v", err)
-	}
-
-	if err := b.Connect(); err != nil {
-		t.Fatalf("Unexpected connect error: %v", err)
-	}
-
-	msg := &broker.Message{
-		Header: map[string]string{
-			"Content-Type": "application/json",
-		},
-		Body: []byte(`{"message": "Hello World"}`),
-	}
-
-	var wg sync.WaitGroup
-
-	sub, err := b.Subscribe("test", func(p broker.Event) error {
-		defer wg.Done()
-
-		m := p.Message()
-
-		if string(m.Body) != string(msg.Body) {
-			t.Fatalf("Unexpected msg %s, expected %s", string(m.Body), string(msg.Body))
-		}
-
-		return nil
-	})
-	if err != nil {
-		t.Fatalf("Unexpected subscribe error: %v", err)
-	}
-
-	for i := 0; i < 10; i++ {
-		wg.Add(1)
-
-		if err := b.Publish("test", msg); err != nil {
-			t.Fatalf("Unexpected publish error: %v", err)
-		}
-	}
-
-	wg.Wait()
-
-	sub.Unsubscribe()
-
-	if err := b.Disconnect(); err != nil {
-		t.Fatalf("Unexpected disconnect error: %v", err)
-	}
-}
-
-func BenchmarkSub1(b *testing.B) {
-	sub(b, 1)
-}
-func BenchmarkSub8(b *testing.B) {
-	sub(b, 8)
-}
-
-func BenchmarkSub32(b *testing.B) {
-	sub(b, 32)
-}
-
-func BenchmarkSub64(b *testing.B) {
-	sub(b, 64)
-}
-
-func BenchmarkSub128(b *testing.B) {
-	sub(b, 128)
-}
-
-func BenchmarkPub1(b *testing.B) {
-	pub(b, 1)
-}
-
-func BenchmarkPub8(b *testing.B) {
-	pub(b, 8)
-}
-
-func BenchmarkPub32(b *testing.B) {
-	pub(b, 32)
-}
-
-func BenchmarkPub64(b *testing.B) {
-	pub(b, 64)
-}
-
-func BenchmarkPub128(b *testing.B) {
-	pub(b, 128)
-}

broker/memory/memory.go

@@ -9,17 +9,11 @@ import (
 	"time"
 
 	"github.com/google/uuid"
-	"github.com/asim/go-micro/v3/cmd"
-	"github.com/asim/go-micro/v3/broker"
-	"github.com/asim/go-micro/v3/logger"
-	maddr "github.com/asim/go-micro/v3/util/addr"
-	mnet "github.com/asim/go-micro/v3/util/net"
+	"github.com/micro/go-micro/v3/broker"
+	maddr "github.com/micro/go-micro/v3/util/addr"
+	mnet "github.com/micro/go-micro/v3/util/net"
 )
 
-func init() {
-        cmd.DefaultBrokers["memory"] = NewBroker
-}
-
 type memoryBroker struct {
 	opts broker.Options
 
@@ -29,13 +23,6 @@ type memoryBroker struct {
 	Subscribers map[string][]*memorySubscriber
 }
 
-type memoryEvent struct {
-	opts    broker.Options
-	topic   string
-	err     error
-	message interface{}
-}
-
 type memorySubscriber struct {
 	id      string
 	topic   string
@@ -108,31 +95,12 @@ func (m *memoryBroker) Publish(topic string, msg *broker.Message, opts ...broker
 		return nil
 	}
 
-	var v interface{}
-	if m.opts.Codec != nil {
-		buf, err := m.opts.Codec.Marshal(msg)
-		if err != nil {
-			return err
-		}
-		v = buf
-	} else {
-		v = msg
-	}
-
-	p := &memoryEvent{
-		topic:   topic,
-		message: v,
-		opts:    m.opts,
-	}
-
 	for _, sub := range subs {
-		if err := sub.handler(p); err != nil {
-			p.err = err
-			if eh := m.opts.ErrorHandler; eh != nil {
-				eh(p)
-				continue
+		if err := sub.handler(msg); err != nil {
+			if eh := sub.opts.ErrorHandler; eh != nil {
+				eh(msg, err)
 			}
-			return err
+			continue
 		}
 	}
 
@@ -185,36 +153,6 @@ func (m *memoryBroker) String() string {
 	return "memory"
 }
 
-func (m *memoryEvent) Topic() string {
-	return m.topic
-}
-
-func (m *memoryEvent) Message() *broker.Message {
-	switch v := m.message.(type) {
-	case *broker.Message:
-		return v
-	case []byte:
-		msg := &broker.Message{}
-		if err := m.opts.Codec.Unmarshal(v, msg); err != nil {
-			if logger.V(logger.ErrorLevel, logger.DefaultLogger) {
-				logger.Errorf("[memory]: failed to unmarshal: %v\n", err)
-			}
-			return nil
-		}
-		return msg
-	}
-
-	return nil
-}
-
-func (m *memoryEvent) Ack() error {
-	return nil
-}
-
-func (m *memoryEvent) Error() error {
-	return m.err
-}
-
 func (m *memorySubscriber) Options() broker.SubscribeOptions {
 	return m.opts
 }

broker/memory/memory_test.go

@@ -4,7 +4,7 @@ import (
 	"fmt"
 	"testing"
 
-	"github.com/asim/go-micro/v3/broker"
+	"github.com/micro/go-micro/v3/broker"
 )
 
 func TestMemoryBroker(t *testing.T) {
@@ -17,7 +17,7 @@ func TestMemoryBroker(t *testing.T) {
 	topic := "test"
 	count := 10
 
-	fn := func(p broker.Event) error {
+	fn := func(m *broker.Message) error {
 		return nil
 	}
 

broker/nats/context.go

@@ -3,7 +3,7 @@ package nats
 import (
 	"context"
 
-	"github.com/asim/go-micro/v3/broker"
+	"github.com/micro/go-micro/v3/broker"
 )
 
 // setBrokerOption returns a function to setup a context with given value

broker/nats/nats.go

@@ -7,18 +7,13 @@ import (
 	"strings"
 	"sync"
 
-	"github.com/asim/go-micro/v3/broker"
-	"github.com/asim/go-micro/v3/cmd"
-	"github.com/asim/go-micro/v3/codec/json"
-	"github.com/asim/go-micro/v3/logger"
-	"github.com/asim/go-micro/v3/registry"
+	"github.com/micro/go-micro/v3/broker"
+	"github.com/micro/go-micro/v3/codec/json"
+	"github.com/micro/go-micro/v3/logger"
+	"github.com/micro/go-micro/v3/registry/mdns"
 	nats "github.com/nats-io/nats.go"
 )
 
-func init() {
-	cmd.DefaultBrokers["nats"] = NewBroker
-}
-
 type natsBroker struct {
 	sync.Once
 	sync.RWMutex
@@ -41,29 +36,6 @@ type subscriber struct {
 	opts broker.SubscribeOptions
 }
 
-type publication struct {
-	t   string
-	err error
-	m   *broker.Message
-}
-
-func (p *publication) Topic() string {
-	return p.t
-}
-
-func (p *publication) Message() *broker.Message {
-	return p.m
-}
-
-func (p *publication) Ack() error {
-	// nats does not support acking
-	return nil
-}
-
-func (p *publication) Error() error {
-	return p.err
-}
-
 func (s *subscriber) Options() broker.SubscribeOptions {
 	return s.opts
 }
@@ -136,6 +108,10 @@ func (n *natsBroker) Connect() error {
 
 		c, err := opts.Connect()
 		if err != nil {
+			if logger.V(logger.WarnLevel, logger.DefaultLogger) {
+				logger.Warnf("Error connecting to broker: %v", err)
+			}
+
 			return err
 		}
 		n.conn = c
@@ -196,7 +172,6 @@ func (n *natsBroker) Subscribe(topic string, handler broker.Handler, opts ...bro
 	n.RUnlock()
 
 	opt := broker.SubscribeOptions{
-		AutoAck: true,
 		Context: context.Background(),
 	}
 
@@ -205,29 +180,25 @@ func (n *natsBroker) Subscribe(topic string, handler broker.Handler, opts ...bro
 	}
 
 	fn := func(msg *nats.Msg) {
-		var m broker.Message
-		pub := &publication{t: msg.Subject}
-		eh := n.opts.ErrorHandler
+		var m *broker.Message
+		eh := opt.ErrorHandler
 		err := n.opts.Codec.Unmarshal(msg.Data, &m)
-		pub.err = err
-		pub.m = &m
 		if err != nil {
 			m.Body = msg.Data
 			if logger.V(logger.ErrorLevel, logger.DefaultLogger) {
 				logger.Error(err)
 			}
 			if eh != nil {
-				eh(pub)
+				eh(m, err)
 			}
 			return
 		}
-		if err := handler(pub); err != nil {
-			pub.err = err
+		if err := handler(m); err != nil {
 			if logger.V(logger.ErrorLevel, logger.DefaultLogger) {
 				logger.Error(err)
 			}
 			if eh != nil {
-				eh(pub)
+				eh(m, err)
 			}
 		}
 	}
@@ -311,7 +282,7 @@ func NewBroker(opts ...broker.Option) broker.Broker {
 		// Default codec
 		Codec:    json.Marshaler{},
 		Context:  context.Background(),
-		Registry: registry.DefaultRegistry,
+		Registry: mdns.NewRegistry(),
 	}
 
 	n := &natsBroker{

broker/nats/nats_test.go

@@ -4,7 +4,7 @@ import (
 	"fmt"
 	"testing"
 
-	"github.com/asim/go-micro/v3/broker"
+	"github.com/micro/go-micro/v3/broker"
 	nats "github.com/nats-io/nats.go"
 )
 

broker/nats/options.go

@@ -1,7 +1,7 @@
 package nats
 
 import (
-	"github.com/asim/go-micro/v3/broker"
+	"github.com/micro/go-micro/v3/broker"
 	nats "github.com/nats-io/nats.go"
 )
 

broker/options.go

@@ -4,8 +4,8 @@ import (
 	"context"
 	"crypto/tls"
 
-	"github.com/asim/go-micro/v3/codec"
-	"github.com/asim/go-micro/v3/registry"
+	"github.com/micro/go-micro/v3/codec"
+	"github.com/micro/go-micro/v3/registry"
 )
 
 type Options struct {
@@ -13,10 +13,6 @@ type Options struct {
 	Secure bool
 	Codec  codec.Marshaler
 
-	// Handler executed when error happens in broker mesage
-	// processing
-	ErrorHandler Handler
-
 	TLSConfig *tls.Config
 	// Registry used for clustering
 	Registry registry.Registry
@@ -32,9 +28,9 @@ type PublishOptions struct {
 }
 
 type SubscribeOptions struct {
-	// AutoAck defaults to true. When a handler returns
-	// with a nil error the message is acked.
-	AutoAck bool
+	// Handler executed when errors occur processing messages
+	ErrorHandler ErrorHandler
+
 	// Subscribers with the same queue name
 	// will create a shared subscription where each
 	// receives a subset of messages.
@@ -59,9 +55,7 @@ func PublishContext(ctx context.Context) PublishOption {
 type SubscribeOption func(*SubscribeOptions)
 
 func NewSubscribeOptions(opts ...SubscribeOption) SubscribeOptions {
-	opt := SubscribeOptions{
-		AutoAck: true,
-	}
+	opt := SubscribeOptions{}
 
 	for _, o := range opts {
 		o(&opt)
@@ -85,18 +79,10 @@ func Codec(c codec.Marshaler) Option {
 	}
 }
 
-// DisableAutoAck will disable auto acking of messages
-// after they have been handled.
-func DisableAutoAck() SubscribeOption {
-	return func(o *SubscribeOptions) {
-		o.AutoAck = false
-	}
-}
-
 // ErrorHandler will catch all broker errors that cant be handled
 // in normal way, for example Codec errors
-func ErrorHandler(h Handler) Option {
-	return func(o *Options) {
+func HandleError(h ErrorHandler) SubscribeOption {
+	return func(o *SubscribeOptions) {
 		o.ErrorHandler = h
 	}
 }

build/build.go

@@ -0,0 +1,32 @@
+// Package build is for building source into a package
+package build
+
+// Build is an interface for building packages
+type Build interface {
+	// Package builds a package
+	Package(name string, src *Source) (*Package, error)
+	// Remove removes the package
+	Remove(*Package) error
+}
+
+// Source is the source of a build
+type Source struct {
+	// Path to the source if local
+	Path string
+	// Language is the language of code
+	Language string
+	// Location of the source
+	Repository string
+}
+
+// Package is packaged format for source
+type Package struct {
+	// Name of the package
+	Name string
+	// Location of the package
+	Path string
+	// Type of package e.g tarball, binary, docker
+	Type string
+	// Source of the package
+	Source *Source
+}

build/docker/docker.go

@@ -8,18 +8,18 @@ import (
 	"os"
 	"path/filepath"
 
-	"github.com/asim/go-micro/v3/logger"
-	"github.com/asim/go-micro/v3/runtime/local/build"
 	docker "github.com/fsouza/go-dockerclient"
+	"github.com/micro/go-micro/v3/build"
+	"github.com/micro/go-micro/v3/logger"
 )
 
-type Builder struct {
+type dockerBuild struct {
 	Options build.Options
 	Client  *docker.Client
 }
 
-func (d *Builder) Build(s *build.Source) (*build.Package, error) {
-	image := filepath.Join(s.Repository.Path, s.Repository.Name)
+func (d *dockerBuild) Package(name string, s *build.Source) (*build.Package, error) {
+	image := name
 
 	buf := new(bytes.Buffer)
 	tw := tar.NewWriter(buf)
@@ -28,7 +28,7 @@ func (d *Builder) Build(s *build.Source) (*build.Package, error) {
 	dockerFile := "Dockerfile"
 
 	// open docker file
-	f, err := os.Open(filepath.Join(s.Repository.Path, s.Repository.Name, dockerFile))
+	f, err := os.Open(filepath.Join(s.Path, dockerFile))
 	if err != nil {
 		return nil, err
 	}
@@ -71,12 +71,15 @@ func (d *Builder) Build(s *build.Source) (*build.Package, error) {
 	}, nil
 }
 
-func (d *Builder) Clean(b *build.Package) error {
-	image := filepath.Join(b.Path, b.Name)
-	return d.Client.RemoveImage(image)
+func (d *dockerBuild) Remove(b *build.Package) error {
+	return d.Client.RemoveImage(b.Name)
 }
 
-func NewBuilder(opts ...build.Option) build.Builder {
+func (d *dockerBuild) String() string {
+	return "docker"
+}
+
+func NewBuild(opts ...build.Option) build.Build {
 	options := build.Options{}
 	for _, o := range opts {
 		o(&options)
@@ -86,7 +89,7 @@ func NewBuilder(opts ...build.Option) build.Builder {
 	if err != nil {
 		logger.Fatal(err)
 	}
-	return &Builder{
+	return &dockerBuild{
 		Options: options,
 		Client:  client,
 	}

build/golang/golang.go

@@ -6,10 +6,10 @@ import (
 	"os/exec"
 	"path/filepath"
 
-	"github.com/asim/go-micro/v3/runtime/local/build"
+	"github.com/micro/go-micro/v3/build"
 )
 
-type Builder struct {
+type goBuild struct {
 	Options build.Options
 	Cmd     string
 	Path    string
@@ -34,35 +34,40 @@ func whichGo() string {
 	return "go"
 }
 
-func (g *Builder) Build(s *build.Source) (*build.Package, error) {
-	binary := filepath.Join(g.Path, s.Repository.Name)
-	source := filepath.Join(s.Repository.Path, s.Repository.Name)
+func (g *goBuild) Package(name string, src *build.Source) (*build.Package, error) {
+	binary := filepath.Join(g.Path, name)
+	source := src.Path
 
 	cmd := exec.Command(g.Cmd, "build", "-o", binary, source)
 	if err := cmd.Run(); err != nil {
 		return nil, err
 	}
+
 	return &build.Package{
-		Name:   s.Repository.Name,
+		Name:   name,
 		Path:   binary,
-		Type:   "go",
-		Source: s,
+		Type:   g.String(),
+		Source: src,
 	}, nil
 }
 
-func (g *Builder) Clean(b *build.Package) error {
+func (g *goBuild) Remove(b *build.Package) error {
 	binary := filepath.Join(b.Path, b.Name)
 	return os.Remove(binary)
 }
 
-func NewBuild(opts ...build.Option) build.Builder {
+func (g *goBuild) String() string {
+	return "golang"
+}
+
+func NewBuild(opts ...build.Option) build.Build {
 	options := build.Options{
 		Path: os.TempDir(),
 	}
 	for _, o := range opts {
 		o(&options)
 	}
-	return &Builder{
+	return &goBuild{
 		Options: options,
 		Cmd:     whichGo(),
 		Path:    options.Path,

build/tar/tar.go

@@ -0,0 +1,45 @@
+// Package tar basically tarballs source code
+package tar
+
+import (
+	"os"
+	"path/filepath"
+
+	"github.com/micro/go-micro/v3/build"
+)
+
+type tarBuild struct{}
+
+func (t *tarBuild) Package(name string, src *build.Source) (*build.Package, error) {
+	pkg := name + ".tar.gz"
+	// path to the tarball
+	path := filepath.Join(os.TempDir(), src.Path, pkg)
+
+	// create a temp directory
+	if err := os.MkdirAll(filepath.Join(os.TempDir(), src.Path), 0755); err != nil {
+		return nil, err
+	}
+
+	if err := Compress(src.Path, path); err != nil {
+		return nil, err
+	}
+
+	return &build.Package{
+		Name:   name,
+		Path:   path,
+		Type:   t.String(),
+		Source: src,
+	}, nil
+}
+
+func (t *tarBuild) Remove(b *build.Package) error {
+	return os.Remove(b.Path)
+}
+
+func (t *tarBuild) String() string {
+	return "tar.gz"
+}
+
+func NewBuild(opts ...build.Option) build.Build {
+	return new(tarBuild)
+}

build/tar/util.go

@@ -0,0 +1,92 @@
+package tar
+
+import (
+	"archive/tar"
+	"bytes"
+	"compress/gzip"
+	"io"
+	"os"
+	"path/filepath"
+	"strings"
+)
+
+func Compress(source, dest string) error {
+	// tar + gzip
+	var buf bytes.Buffer
+	_ = compress(source, &buf)
+
+	// write the .tar.gzip
+	fileToWrite, err := os.OpenFile(dest, os.O_CREATE|os.O_RDWR, 0666)
+	if err != nil {
+		return err
+	}
+	_, err = io.Copy(fileToWrite, &buf)
+	return err
+}
+
+func compress(src string, buf io.Writer) error {
+	// tar > gzip > buf
+	zr := gzip.NewWriter(buf)
+	tw := tar.NewWriter(zr)
+
+	// walk through every file in the folder
+	filepath.Walk(src, func(file string, fi os.FileInfo, err error) error {
+
+		// generate tar header
+		header, err := tar.FileInfoHeader(fi, file)
+		if err != nil {
+			return err
+		}
+
+		// must provide real name
+		// (see https://golang.org/src/archive/tar/common.go?#L626)
+
+		srcWithSlash := src
+		if !strings.HasSuffix(src, string(filepath.Separator)) {
+			srcWithSlash = src + string(filepath.Separator)
+		}
+		header.Name = strings.ReplaceAll(file, srcWithSlash, "")
+		if header.Name == src || len(strings.TrimSpace(header.Name)) == 0 {
+			return nil
+		}
+
+		// @todo This is a quick hack to speed up whole repo uploads
+		// https://github.com/micro/micro/pull/956
+		if !fi.IsDir() && !strings.HasSuffix(header.Name, ".go") &&
+			!strings.HasSuffix(header.Name, ".mod") &&
+			!strings.HasSuffix(header.Name, ".sum") {
+			return nil
+		}
+
+		// write header
+		if err := tw.WriteHeader(header); err != nil {
+			return err
+		}
+		if fi.IsDir() {
+			return nil
+		}
+
+		// if not a dir, write file content
+
+		data, err := os.Open(file)
+		if err != nil {
+			return err
+		}
+		if _, err := io.Copy(tw, data); err != nil {
+			return err
+		}
+
+		return nil
+	})
+
+	// produce tar
+	if err := tw.Close(); err != nil {
+		return err
+	}
+	// produce gzip
+	if err := zr.Close(); err != nil {
+		return err
+	}
+	//
+	return nil
+}

cache/cache.go

@@ -1,65 +1,29 @@
+// Package cache is a caching interface
 package cache
 
-import (
-	"context"
-	"errors"
-	"time"
-)
-
-var (
-	// DefaultCache is the default cache.
-	DefaultCache Cache = NewCache()
-	// DefaultExpiration is the default duration for items stored in
-	// the cache to expire.
-	DefaultExpiration time.Duration = 0
-
-	// ErrItemExpired is returned in Cache.Get when the item found in the cache
-	// has expired.
-	ErrItemExpired error = errors.New("item has expired")
-	// ErrKeyNotFound is returned in Cache.Get and Cache.Delete when the
-	// provided key could not be found in cache.
-	ErrKeyNotFound error = errors.New("key not found in cache")
-)
-
-// Cache is the interface that wraps the cache.
-//
-// Context specifies the context for the cache.
-// Get gets a cached value by key.
-// Put stores a key-value pair into cache.
-// Delete removes a key from cache.
+// Cache is an interface for caching
 type Cache interface {
-	Context(ctx context.Context) Cache
-	Get(key string) (interface{}, time.Time, error)
-	Put(key string, val interface{}, d time.Duration) error
+	// Initialise options
+	Init(...Option) error
+	// Get a value
+	Get(key string) (interface{}, error)
+	// Set a value
+	Set(key string, val interface{}) error
+	// Delete a value
 	Delete(key string) error
+	// Name of the implementation
+	String() string
 }
 
-// Item represents an item stored in the cache.
-type Item struct {
-	Value      interface{}
-	Expiration int64
+type Options struct {
+	Nodes []string
 }
 
-// Expired returns true if the item has expired.
-func (i *Item) Expired() bool {
-	if i.Expiration == 0 {
-		return false
-	}
+type Option func(o *Options)
 
-	return time.Now().UnixNano() > i.Expiration
-}
-
-// NewCache returns a new cache.
-func NewCache(opts ...Option) Cache {
-	options := NewOptions(opts...)
-	items := make(map[string]Item)
-
-	if len(options.Items) > 0 {
-		items = options.Items
-	}
-
-	return &memCache{
-		opts:  options,
-		items: items,
+// Nodes sets the nodes for the cache
+func Nodes(v ...string) Option {
+	return func(o *Options) {
+		o.Nodes = v
 	}
 }

cache/cache_test.go

@@ -1,111 +0,0 @@
-package cache
-
-import (
-	"context"
-	"testing"
-	"time"
-)
-
-var (
-	ctx context.Context = context.TODO()
-	key string          = "test"
-	val interface{}     = "hello go-micro"
-)
-
-// TestMemCache tests the in-memory cache implementation.
-func TestCache(t *testing.T) {
-	t.Run("CacheGetMiss", func(t *testing.T) {
-		if _, _, err := NewCache().Context(ctx).Get(key); err == nil {
-			t.Error("expected to get no value from cache")
-		}
-	})
-
-	t.Run("CacheGetHit", func(t *testing.T) {
-		c := NewCache()
-
-		if err := c.Context(ctx).Put(key, val, 0); err != nil {
-			t.Error(err)
-		}
-
-		if a, _, err := c.Context(ctx).Get(key); err != nil {
-			t.Errorf("Expected a value, got err: %s", err)
-		} else if a != val {
-			t.Errorf("Expected '%v', got '%v'", val, a)
-		}
-	})
-
-	t.Run("CacheGetExpired", func(t *testing.T) {
-		c := NewCache()
-		e := 20 * time.Millisecond
-
-		if err := c.Context(ctx).Put(key, val, e); err != nil {
-			t.Error(err)
-		}
-
-		<-time.After(25 * time.Millisecond)
-		if _, _, err := c.Context(ctx).Get(key); err == nil {
-			t.Error("expected to get no value from cache")
-		}
-	})
-
-	t.Run("CacheGetValid", func(t *testing.T) {
-		c := NewCache()
-		e := 25 * time.Millisecond
-
-		if err := c.Context(ctx).Put(key, val, e); err != nil {
-			t.Error(err)
-		}
-
-		<-time.After(20 * time.Millisecond)
-		if _, _, err := c.Context(ctx).Get(key); err != nil {
-			t.Errorf("expected a value, got err: %s", err)
-		}
-	})
-
-	t.Run("CacheDeleteMiss", func(t *testing.T) {
-		if err := NewCache().Context(ctx).Delete(key); err == nil {
-			t.Error("expected to delete no value from cache")
-		}
-	})
-
-	t.Run("CacheDeleteHit", func(t *testing.T) {
-		c := NewCache()
-
-		if err := c.Context(ctx).Put(key, val, 0); err != nil {
-			t.Error(err)
-		}
-
-		if err := c.Context(ctx).Delete(key); err != nil {
-			t.Errorf("Expected to delete an item, got err: %s", err)
-		}
-
-		if _, _, err := c.Context(ctx).Get(key); err == nil {
-			t.Errorf("Expected error")
-		}
-	})
-}
-
-func TestCacheWithOptions(t *testing.T) {
-	t.Run("CacheWithExpiration", func(t *testing.T) {
-		c := NewCache(Expiration(20 * time.Millisecond))
-
-		if err := c.Context(ctx).Put(key, val, 0); err != nil {
-			t.Error(err)
-		}
-
-		<-time.After(25 * time.Millisecond)
-		if _, _, err := c.Context(ctx).Get(key); err == nil {
-			t.Error("expected to get no value from cache")
-		}
-	})
-
-	t.Run("CacheWithItems", func(t *testing.T) {
-		c := NewCache(Items(map[string]Item{key: {val, 0}}))
-
-		if a, _, err := c.Context(ctx).Get(key); err != nil {
-			t.Errorf("Expected a value, got err: %s", err)
-		} else if a != val {
-			t.Errorf("Expected '%v', got '%v'", val, a)
-		}
-	})
-}

cache/default.go

@@ -1,68 +0,0 @@
-package cache
-
-import (
-	"context"
-	"sync"
-	"time"
-)
-
-type memCache struct {
-	opts Options
-	sync.RWMutex
-	ctx context.Context
-
-	items map[string]Item
-}
-
-func (c *memCache) Context(ctx context.Context) Cache {
-	c.ctx = ctx
-	return c
-}
-
-func (c *memCache) Get(key string) (interface{}, time.Time, error) {
-	c.RWMutex.Lock()
-	defer c.RWMutex.Unlock()
-
-	item, found := c.items[key]
-	if !found {
-		return nil, time.Time{}, ErrKeyNotFound
-	}
-	if item.Expired() {
-		return nil, time.Time{}, ErrItemExpired
-	}
-
-	return item.Value, time.Unix(0, item.Expiration), nil
-}
-
-func (c *memCache) Put(key string, val interface{}, d time.Duration) error {
-	var e int64
-	if d == DefaultExpiration {
-		d = c.opts.Expiration
-	}
-	if d > 0 {
-		e = time.Now().Add(d).UnixNano()
-	}
-
-	c.RWMutex.Lock()
-	defer c.RWMutex.Unlock()
-
-	c.items[key] = Item{
-		Value:      val,
-		Expiration: e,
-	}
-
-	return nil
-}
-
-func (c *memCache) Delete(key string) error {
-	c.RWMutex.Lock()
-	defer c.RWMutex.Unlock()
-
-	_, found := c.items[key]
-	if !found {
-		return ErrKeyNotFound
-	}
-
-	delete(c.items, key)
-	return nil
-}

cache/memcache/memcache.go

@@ -0,0 +1,80 @@
+// Package memcache is a memcache implementation of the Cache
+package memcache
+
+import (
+	"encoding/json"
+
+	"github.com/bradfitz/gomemcache/memcache"
+	"github.com/micro/go-micro/v3/cache"
+)
+
+type memcacheCache struct {
+	options cache.Options
+	client  *memcache.Client
+}
+
+type memcacheItem struct {
+	Key   string
+	Value interface{}
+}
+
+func (m *memcacheCache) Init(opts ...cache.Option) error {
+	for _, o := range opts {
+		o(&m.options)
+	}
+	return nil
+}
+
+func (m *memcacheCache) Get(key string) (interface{}, error) {
+	item, err := m.client.Get(key)
+	if err != nil {
+		return nil, err
+	}
+
+	var mc *memcacheItem
+
+	if err := json.Unmarshal(item.Value, &mc); err != nil {
+		return nil, err
+	}
+
+	return mc.Value, nil
+}
+
+func (m *memcacheCache) Set(key string, val interface{}) error {
+	b, err := json.Marshal(val)
+	if err != nil {
+		return err
+	}
+
+	return m.client.Set(&memcache.Item{
+		Key:   key,
+		Value: b,
+	})
+}
+
+func (m *memcacheCache) Delete(key string) error {
+	return m.client.Delete(key)
+}
+
+func (m *memcacheCache) String() string {
+	return "memcache"
+}
+
+// NewCache returns a new memcache Cache
+func NewCache(opts ...cache.Option) cache.Cache {
+	var options cache.Options
+	for _, o := range opts {
+		o(&options)
+	}
+
+	// get and set the nodes
+	nodes := options.Nodes
+	if len(nodes) == 0 {
+		nodes = []string{"localhost:11211"}
+	}
+
+	return &memcacheCache{
+		options: options,
+		client:  memcache.New(nodes...),
+	}
+}

cache/memory/memory.go

@@ -0,0 +1,56 @@
+// Package memory is an in memory cache
+package memory
+
+import (
+	"sync"
+
+	"github.com/micro/go-micro/v3/cache"
+	"github.com/micro/go-micro/v3/errors"
+)
+
+type memoryCache struct {
+	// TODO: use a decent caching library
+	sync.RWMutex
+	values map[string]interface{}
+}
+
+func (m *memoryCache) Init(opts ...cache.Option) error {
+	// TODO: implement
+	return nil
+}
+
+func (m *memoryCache) Get(key string) (interface{}, error) {
+	m.RLock()
+	defer m.RUnlock()
+
+	v, ok := m.values[key]
+	if !ok {
+		return nil, errors.NotFound("go.micro.cache", key+" not found")
+	}
+
+	return v, nil
+}
+
+func (m *memoryCache) Set(key string, val interface{}) error {
+	m.Lock()
+	m.values[key] = val
+	m.Unlock()
+	return nil
+}
+
+func (m *memoryCache) Delete(key string) error {
+	m.Lock()
+	delete(m.values, key)
+	m.Unlock()
+	return nil
+}
+
+func (m *memoryCache) String() string {
+	return "memory"
+}
+
+func NewCache(opts ...cache.Option) cache.Cache {
+	return &memoryCache{
+		values: make(map[string]interface{}),
+	}
+}

cache/options.go

@@ -1,40 +0,0 @@
-package cache
-
-import "time"
-
-// Options represents the options for the cache.
-type Options struct {
-	Expiration time.Duration
-	Items      map[string]Item
-}
-
-// Option manipulates the Options passed.
-type Option func(o *Options)
-
-// Expiration sets the duration for items stored in the cache to expire.
-func Expiration(d time.Duration) Option {
-	return func(o *Options) {
-		o.Expiration = d
-	}
-}
-
-// Items initializes the cache with preconfigured items.
-func Items(i map[string]Item) Option {
-	return func(o *Options) {
-		o.Items = i
-	}
-}
-
-// NewOptions returns a new options struct.
-func NewOptions(opts ...Option) Options {
-	options := Options{
-		Expiration: DefaultExpiration,
-		Items:      make(map[string]Item),
-	}
-
-	for _, o := range opts {
-		o(&options)
-	}
-
-	return options
-}

cache/options_test.go

@@ -1,43 +0,0 @@
-package cache
-
-import (
-	"testing"
-	"time"
-)
-
-func TestOptions(t *testing.T) {
-	testData := map[string]struct {
-		set        bool
-		expiration time.Duration
-		items      map[string]Item
-	}{
-		"DefaultOptions":  {false, DefaultExpiration, map[string]Item{}},
-		"ModifiedOptions": {true, time.Second, map[string]Item{"test": {"hello go-micro", 0}}},
-	}
-
-	for k, d := range testData {
-		t.Run(k, func(t *testing.T) {
-			var opts Options
-
-			if d.set {
-				opts = NewOptions(
-					Expiration(d.expiration),
-					Items(d.items),
-				)
-			} else {
-				opts = NewOptions()
-			}
-
-			// test options
-			for _, o := range []Options{opts} {
-				if o.Expiration != d.expiration {
-					t.Fatalf("Expected expiration '%v', got '%v'", d.expiration, o.Expiration)
-				}
-
-				if o.Items["test"] != d.items["test"] {
-					t.Fatalf("Expected items %#v, got %#v", d.items, o.Items)
-				}
-			}
-		})
-	}
-}

client/backoff.go

@@ -4,7 +4,7 @@ import (
 	"context"
 	"time"
 
-	"github.com/asim/go-micro/v3/util/backoff"
+	"github.com/micro/go-micro/v3/util/backoff"
 )
 
 type BackoffFunc func(ctx context.Context, req Request, attempts int) (time.Duration, error)

client/backoff_test.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"testing"
 	"time"
+
+	"github.com/micro/go-micro/v3/codec"
 )
 
 func TestBackoff(t *testing.T) {
@@ -16,10 +18,13 @@ func TestBackoff(t *testing.T) {
 		7900 * time.Millisecond,
 	}
 
-	c := NewClient()
+	r := &testRequest{
+		service: "test",
+		method:  "test",
+	}
 
 	for i := 0; i < 5; i++ {
-		d, err := exponentialBackoff(context.TODO(), c.NewRequest("test", "test", nil), i)
+		d, err := exponentialBackoff(context.TODO(), r, i)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -29,3 +34,63 @@ func TestBackoff(t *testing.T) {
 		}
 	}
 }
+
+type testRequest struct {
+	service     string
+	method      string
+	endpoint    string
+	contentType string
+	codec       codec.Codec
+	body        interface{}
+	opts        RequestOptions
+}
+
+func newRequest(service, endpoint 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 &testRequest{
+		service:     service,
+		method:      endpoint,
+		endpoint:    endpoint,
+		body:        request,
+		contentType: contentType,
+		opts:        opts,
+	}
+}
+
+func (r *testRequest) ContentType() string {
+	return r.contentType
+}
+
+func (r *testRequest) Service() string {
+	return r.service
+}
+
+func (r *testRequest) Method() string {
+	return r.method
+}
+
+func (r *testRequest) Endpoint() string {
+	return r.endpoint
+}
+
+func (r *testRequest) Body() interface{} {
+	return r.body
+}
+
+func (r *testRequest) Codec() codec.Writer {
+	return r.codec
+}
+
+func (r *testRequest) Stream() bool {
+	return r.opts.Stream
+}

client/cache.go

@@ -1,66 +0,0 @@
-package client
-
-import (
-	"context"
-	"encoding/json"
-	"fmt"
-	"hash/fnv"
-	"time"
-
-	"github.com/asim/go-micro/v3/metadata"
-	cache "github.com/patrickmn/go-cache"
-)
-
-// NewCache returns an initialised cache.
-func NewCache() *Cache {
-	return &Cache{
-		cache: cache.New(cache.NoExpiration, 30*time.Second),
-	}
-}
-
-// Cache for responses
-type Cache struct {
-	cache *cache.Cache
-}
-
-// Get a response from the cache
-func (c *Cache) Get(ctx context.Context, req *Request) (interface{}, bool) {
-	return c.cache.Get(key(ctx, req))
-}
-
-// Set a response in the cache
-func (c *Cache) Set(ctx context.Context, req *Request, rsp interface{}, expiry time.Duration) {
-	c.cache.Set(key(ctx, req), rsp, expiry)
-}
-
-// List the key value pairs in the cache
-func (c *Cache) List() map[string]string {
-	items := c.cache.Items()
-
-	rsp := make(map[string]string, len(items))
-	for k, v := range items {
-		bytes, _ := json.Marshal(v.Object)
-		rsp[k] = string(bytes)
-	}
-
-	return rsp
-}
-
-// key returns a hash for the context and request
-func key(ctx context.Context, req *Request) string {
-	ns, _ := metadata.Get(ctx, "Micro-Namespace")
-
-	bytes, _ := json.Marshal(map[string]interface{}{
-		"namespace": ns,
-		"request": map[string]interface{}{
-			"service":  (*req).Service(),
-			"endpoint": (*req).Endpoint(),
-			"method":   (*req).Method(),
-			"body":     (*req).Body(),
-		},
-	})
-
-	h := fnv.New64()
-	h.Write(bytes)
-	return fmt.Sprintf("%x", h.Sum(nil))
-}

client/cache_test.go

@@ -1,76 +0,0 @@
-package client
-
-import (
-	"context"
-	"testing"
-	"time"
-
-	"github.com/asim/go-micro/v3/metadata"
-)
-
-func TestCache(t *testing.T) {
-	ctx := context.TODO()
-	req := NewRequest("go.micro.service.foo", "Foo.Bar", nil)
-
-	t.Run("CacheMiss", func(t *testing.T) {
-		if _, ok := NewCache().Get(ctx, &req); ok {
-			t.Errorf("Expected to get no result from Get")
-		}
-	})
-
-	t.Run("CacheHit", func(t *testing.T) {
-		c := NewCache()
-
-		rsp := "theresponse"
-		c.Set(ctx, &req, rsp, time.Minute)
-
-		if res, ok := c.Get(ctx, &req); !ok {
-			t.Errorf("Expected a result, got nothing")
-		} else if res != rsp {
-			t.Errorf("Expected '%v' result, got '%v'", rsp, res)
-		}
-	})
-}
-
-func TestCacheKey(t *testing.T) {
-	ctx := context.TODO()
-	req1 := NewRequest("go.micro.service.foo", "Foo.Bar", nil)
-	req2 := NewRequest("go.micro.service.foo", "Foo.Baz", nil)
-	req3 := NewRequest("go.micro.service.foo", "Foo.Baz", "customquery")
-
-	t.Run("IdenticalRequests", func(t *testing.T) {
-		key1 := key(ctx, &req1)
-		key2 := key(ctx, &req1)
-		if key1 != key2 {
-			t.Errorf("Expected the keys to match for identical requests and context")
-		}
-	})
-
-	t.Run("DifferentRequestEndpoints", func(t *testing.T) {
-		key1 := key(ctx, &req1)
-		key2 := key(ctx, &req2)
-
-		if key1 == key2 {
-			t.Errorf("Expected the keys to differ for different request endpoints")
-		}
-	})
-
-	t.Run("DifferentRequestBody", func(t *testing.T) {
-		key1 := key(ctx, &req2)
-		key2 := key(ctx, &req3)
-
-		if key1 == key2 {
-			t.Errorf("Expected the keys to differ for different request bodies")
-		}
-	})
-
-	t.Run("DifferentMetadata", func(t *testing.T) {
-		mdCtx := metadata.Set(context.TODO(), "Micro-Namespace", "bar")
-		key1 := key(mdCtx, &req1)
-		key2 := key(ctx, &req1)
-
-		if key1 == key2 {
-			t.Errorf("Expected the keys to differ for different metadata")
-		}
-	})
-}

client/client.go

@@ -5,7 +5,7 @@ import (
 	"context"
 	"time"
 
-	"github.com/asim/go-micro/v3/codec"
+	"github.com/micro/go-micro/v3/codec"
 )
 
 // Client is the interface used to make requests to services.
@@ -22,11 +22,6 @@ type Client interface {
 	String() string
 }
 
-// Router manages request routing
-type Router interface {
-	SendRequest(context.Context, Request) (Response, error)
-}
-
 // Message is the interface for publishing asynchronously
 type Message interface {
 	Topic() string
@@ -62,7 +57,7 @@ type Response interface {
 	Read() ([]byte, error)
 }
 
-// Stream is the inteface for a bidirectional synchronous stream
+// Stream is the interface for a bidirectional synchronous stream
 type Stream interface {
 	// Context for the stream
 	Context() context.Context
@@ -96,8 +91,6 @@ type MessageOption func(*MessageOptions)
 type RequestOption func(*RequestOptions)
 
 var (
-	// DefaultClient is a default client to use out of the box
-	DefaultClient Client = newRpcClient()
 	// DefaultBackoff is the default backoff function for retries
 	DefaultBackoff = exponentialBackoff
 	// DefaultRetry is the default check-for-retry function for retries
@@ -110,39 +103,4 @@ var (
 	DefaultPoolSize = 100
 	// DefaultPoolTTL sets the connection pool ttl
 	DefaultPoolTTL = time.Minute
-
-	// NewClient returns a new client
-	NewClient func(...Option) Client = newRpcClient
 )
-
-// Makes a synchronous call to a service using the default client
-func Call(ctx context.Context, request Request, response interface{}, opts ...CallOption) error {
-	return DefaultClient.Call(ctx, request, response, opts...)
-}
-
-// Publishes a publication using the default client. Using the underlying broker
-// set within the options.
-func Publish(ctx context.Context, msg Message, opts ...PublishOption) error {
-	return DefaultClient.Publish(ctx, msg, opts...)
-}
-
-// Creates a new message using the default client
-func NewMessage(topic string, payload interface{}, opts ...MessageOption) Message {
-	return DefaultClient.NewMessage(topic, payload, opts...)
-}
-
-// 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, endpoint string, request interface{}, reqOpts ...RequestOption) Request {
-	return DefaultClient.NewRequest(service, endpoint, 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 {
-	return DefaultClient.String()
-}

client/context.go

@@ -1,16 +0,0 @@
-package client
-
-import (
-	"context"
-)
-
-type clientKey struct{}
-
-func FromContext(ctx context.Context) (Client, bool) {
-	c, ok := ctx.Value(clientKey{}).(Client)
-	return c, ok
-}
-
-func NewContext(ctx context.Context, c Client) context.Context {
-	return context.WithValue(ctx, clientKey{}, c)
-}

client/grpc/codec.go

@@ -9,8 +9,8 @@ import (
 
 	"github.com/golang/protobuf/jsonpb"
 	"github.com/golang/protobuf/proto"
-	"github.com/asim/go-micro/v3/codec"
-	"github.com/asim/go-micro/v3/codec/bytes"
+	"github.com/micro/go-micro/v3/codec"
+	"github.com/micro/go-micro/v3/codec/bytes"
 	"github.com/oxtoacart/bpool"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/encoding"

client/grpc/error.go

@@ -0,0 +1,39 @@
+package grpc
+
+import (
+	"github.com/micro/go-micro/v3/errors"
+	"google.golang.org/grpc/status"
+)
+
+func microError(err error) error {
+	// no error
+	switch err {
+	case nil:
+		return nil
+	}
+
+	if verr, ok := err.(*errors.Error); ok {
+		return verr
+	}
+
+	// grpc error
+	s, ok := status.FromError(err)
+	if !ok {
+		return err
+	}
+
+	// return first error from details
+	if details := s.Details(); len(details) > 0 {
+		if verr, ok := details[0].(error); ok {
+			return microError(verr)
+		}
+	}
+
+	// try to decode micro *errors.Error
+	if e := errors.Parse(s.Message()); e.Code > 0 {
+		return e // actually a micro error
+	}
+
+	// fallback
+	return errors.InternalServerError("go.micro.client", s.Message())
+}

client/grpc/grpc.go

@@ -11,15 +11,11 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/asim/go-micro/v3/broker"
-	"github.com/asim/go-micro/v3/cmd"
-	"github.com/asim/go-micro/v3/client"
-	raw "github.com/asim/go-micro/v3/codec/bytes"
-	"github.com/asim/go-micro/v3/errors"
-	"github.com/asim/go-micro/v3/metadata"
-	"github.com/asim/go-micro/v3/registry"
-	"github.com/asim/go-micro/v3/selector"
-	pnet "github.com/asim/go-micro/v3/util/net"
+	"github.com/micro/go-micro/v3/broker"
+	"github.com/micro/go-micro/v3/client"
+	raw "github.com/micro/go-micro/v3/codec/bytes"
+	"github.com/micro/go-micro/v3/errors"
+	"github.com/micro/go-micro/v3/metadata"
 
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
@@ -34,8 +30,6 @@ type grpcClient struct {
 }
 
 func init() {
-	cmd.DefaultClients["grpc"] = NewClient
-
 	encoding.RegisterCodec(wrapCodec{jsonCodec{}})
 	encoding.RegisterCodec(wrapCodec{protoCodec{}})
 	encoding.RegisterCodec(wrapCodec{bytesCodec{}})
@@ -75,35 +69,9 @@ func (g *grpcClient) secure(addr string) grpc.DialOption {
 	return grpc.WithInsecure()
 }
 
-func (g *grpcClient) next(request client.Request, opts client.CallOptions) (selector.Next, error) {
-	service, address, _ := pnet.Proxy(request.Service(), opts.Address)
-
-	// return remote address
-	if len(address) > 0 {
-		return func() (*registry.Node, error) {
-			return &registry.Node{
-				Address: address[0],
-			}, nil
-		}, nil
-	}
-
-	// get next nodes from the selector
-	next, err := g.opts.Selector.Select(service, opts.SelectOptions...)
-	if err != nil {
-		if err == selector.ErrNotFound {
-			return nil, errors.InternalServerError("go.micro.client", "service %s: %s", service, err.Error())
-		}
-		return nil, errors.InternalServerError("go.micro.client", "error selecting %s node: %s", service, err.Error())
-	}
-
-	return next, nil
-}
-
-func (g *grpcClient) call(ctx context.Context, node *registry.Node, req client.Request, rsp interface{}, opts client.CallOptions) error {
+func (g *grpcClient) call(ctx context.Context, addr string, req client.Request, rsp interface{}, opts client.CallOptions) error {
 	var header map[string]string
 
-	address := node.Address
-
 	header = make(map[string]string)
 	if md, ok := metadata.FromContext(ctx); ok {
 		header = make(map[string]string, len(md))
@@ -134,7 +102,7 @@ func (g *grpcClient) call(ctx context.Context, node *registry.Node, req client.R
 
 	grpcDialOptions := []grpc.DialOption{
 		grpc.WithTimeout(opts.DialTimeout),
-		g.secure(address),
+		g.secure(addr),
 		grpc.WithDefaultCallOptions(
 			grpc.MaxCallRecvMsgSize(maxRecvMsgSize),
 			grpc.MaxCallSendMsgSize(maxSendMsgSize),
@@ -145,13 +113,13 @@ func (g *grpcClient) call(ctx context.Context, node *registry.Node, req client.R
 		grpcDialOptions = append(grpcDialOptions, opts...)
 	}
 
-	cc, err := g.pool.getConn(address, grpcDialOptions...)
+	cc, err := g.pool.getConn(addr, grpcDialOptions...)
 	if err != nil {
 		return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error sending request: %v", err))
 	}
 	defer func() {
 		// defer execution of release
-		g.pool.release(address, cc, grr)
+		g.pool.release(addr, cc, grr)
 	}()
 
 	ch := make(chan error, 1)
@@ -177,11 +145,9 @@ func (g *grpcClient) call(ctx context.Context, node *registry.Node, req client.R
 	return grr
 }
 
-func (g *grpcClient) stream(ctx context.Context, node *registry.Node, req client.Request, rsp interface{}, opts client.CallOptions) error {
+func (g *grpcClient) stream(ctx context.Context, addr string, req client.Request, rsp interface{}, opts client.CallOptions) error {
 	var header map[string]string
 
-	address := node.Address
-
 	if md, ok := metadata.FromContext(ctx); ok {
 		header = make(map[string]string, len(md))
 		for k, v := range md {
@@ -219,14 +185,14 @@ func (g *grpcClient) stream(ctx context.Context, node *registry.Node, req client
 
 	grpcDialOptions := []grpc.DialOption{
 		grpc.WithTimeout(opts.DialTimeout),
-		g.secure(address),
+		g.secure(addr),
 	}
 
 	if opts := g.getGrpcDialOptions(); opts != nil {
 		grpcDialOptions = append(grpcDialOptions, opts...)
 	}
 
-	cc, err := grpc.DialContext(dialCtx, address, grpcDialOptions...)
+	cc, err := grpc.DialContext(dialCtx, addr, grpcDialOptions...)
 	if err != nil {
 		return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error sending request: %v", err))
 	}
@@ -271,15 +237,15 @@ func (g *grpcClient) stream(ctx context.Context, node *registry.Node, req client
 
 	// setup the stream response
 	stream := &grpcStream{
-		context: ctx,
-		request: req,
+		ClientStream: st,
+		context:      ctx,
+		request:      req,
 		response: &response{
 			conn:   cc,
 			stream: st,
 			codec:  cf,
 			gcodec: codec,
 		},
-		stream: st,
 		conn:   cc,
 		cancel: cancel,
 	}
@@ -393,11 +359,6 @@ func (g *grpcClient) Call(ctx context.Context, req client.Request, rsp interface
 		opt(&callOpts)
 	}
 
-	next, err := g.next(req, callOpts)
-	if err != nil {
-		return err
-	}
-
 	// check if we already have a deadline
 	d, ok := ctx.Deadline()
 	if !ok {
@@ -427,6 +388,34 @@ func (g *grpcClient) Call(ctx context.Context, req client.Request, rsp interface
 		gcall = callOpts.CallWrappers[i-1](gcall)
 	}
 
+	// use the router passed as a call option, or fallback to the rpc clients router
+	if callOpts.Router == nil {
+		callOpts.Router = g.opts.Router
+	}
+
+	if callOpts.Selector == nil {
+		callOpts.Selector = g.opts.Selector
+	}
+
+	// inject proxy address
+	// TODO: don't even bother using Lookup/Select in this case
+	if len(g.opts.Proxy) > 0 {
+		callOpts.Address = []string{g.opts.Proxy}
+	}
+
+	// lookup the route to send the reques to
+	// TODO apply any filtering here
+	routes, err := g.opts.Lookup(ctx, req, callOpts)
+	if err != nil {
+		return errors.InternalServerError("go.micro.client", err.Error())
+	}
+
+	// balance the list of nodes
+	next, err := callOpts.Selector.Select(routes)
+	if err != nil {
+		return err
+	}
+
 	// return errors.New("go.micro.client", "request timeout", 408)
 	call := func(i int) error {
 		// call backoff first. Someone may want an initial start delay
@@ -440,19 +429,16 @@ func (g *grpcClient) Call(ctx context.Context, req client.Request, rsp interface
 			time.Sleep(t)
 		}
 
-		// select next node
-		node, err := next()
-		service := req.Service()
-		if err != nil {
-			if err == selector.ErrNotFound {
-				return errors.InternalServerError("go.micro.client", "service %s: %s", service, err.Error())
-			}
-			return errors.InternalServerError("go.micro.client", "error selecting %s node: %s", service, err.Error())
-		}
+		// get the next node
+		node := next()
 
 		// make the call
 		err = gcall(ctx, node, req, rsp, callOpts)
-		g.opts.Selector.Mark(service, node, err)
+
+		// record the result of the call to inform future routing decisions
+		g.opts.Selector.Record(node, err)
+
+		// try and transform the error to a go-micro error
 		if verr, ok := err.(*errors.Error); ok {
 			return verr
 		}
@@ -500,11 +486,6 @@ func (g *grpcClient) Stream(ctx context.Context, req client.Request, opts ...cli
 		opt(&callOpts)
 	}
 
-	next, err := g.next(req, callOpts)
-	if err != nil {
-		return nil, err
-	}
-
 	// #200 - streams shouldn't have a request timeout set on the context
 
 	// should we noop right here?
@@ -522,6 +503,34 @@ func (g *grpcClient) Stream(ctx context.Context, req client.Request, opts ...cli
 		gstream = callOpts.CallWrappers[i-1](gstream)
 	}
 
+	// use the router passed as a call option, or fallback to the rpc clients router
+	if callOpts.Router == nil {
+		callOpts.Router = g.opts.Router
+	}
+
+	if callOpts.Selector == nil {
+		callOpts.Selector = g.opts.Selector
+	}
+
+	// inject proxy address
+	// TODO: don't even bother using Lookup/Select in this case
+	if len(g.opts.Proxy) > 0 {
+		callOpts.Address = []string{g.opts.Proxy}
+	}
+
+	// lookup the route to send the reques to
+	// TODO: move to internal lookup func
+	routes, err := g.opts.Lookup(ctx, req, callOpts)
+	if err != nil {
+		return nil, errors.InternalServerError("go.micro.client", err.Error())
+	}
+
+	// balance the list of nodes
+	next, err := callOpts.Selector.Select(routes)
+	if err != nil {
+		return nil, err
+	}
+
 	call := func(i int) (client.Stream, error) {
 		// call backoff first. Someone may want an initial start delay
 		t, err := callOpts.Backoff(ctx, req, i)
@@ -534,20 +543,22 @@ func (g *grpcClient) Stream(ctx context.Context, req client.Request, opts ...cli
 			time.Sleep(t)
 		}
 
-		node, err := next()
-		service := req.Service()
-		if err != nil {
-			if err == selector.ErrNotFound {
-				return nil, errors.InternalServerError("go.micro.client", "service %s: %s", service, err.Error())
-			}
-			return nil, errors.InternalServerError("go.micro.client", "error selecting %s node: %s", service, err.Error())
-		}
+		// get the next node
+		node := next()
 
 		// make the call
 		stream := &grpcStream{}
 		err = g.stream(ctx, node, req, stream, callOpts)
 
-		g.opts.Selector.Mark(service, node, err)
+		// record the result of the call to inform future routing decisions
+		g.opts.Selector.Record(node, err)
+
+		// try and transform the error to a go-micro error
+		if verr, ok := err.(*errors.Error); ok {
+			return nil, verr
+		}
+
+		g.opts.Selector.Record(node, err)
 		return stream, err
 	}
 
@@ -574,7 +585,7 @@ func (g *grpcClient) Stream(ctx context.Context, req client.Request, opts ...cli
 				return rsp.stream, nil
 			}
 
-			retry, rerr := callOpts.Retry(ctx, req, i, err)
+			retry, rerr := callOpts.Retry(ctx, req, i, grr)
 			if rerr != nil {
 				return nil, rerr
 			}
@@ -592,6 +603,16 @@ func (g *grpcClient) Stream(ctx context.Context, req client.Request, opts ...cli
 
 func (g *grpcClient) Publish(ctx context.Context, p client.Message, opts ...client.PublishOption) error {
 	var options client.PublishOptions
+	var body []byte
+
+	// fail early on connect error
+	if !g.once.Load().(bool) {
+		if err := g.opts.Broker.Connect(); err != nil {
+			return errors.InternalServerError("go.micro.client", err.Error())
+		}
+		g.once.Store(true)
+	}
+
 	for _, o := range opts {
 		o(&options)
 	}
@@ -603,17 +624,15 @@ func (g *grpcClient) Publish(ctx context.Context, p client.Message, opts ...clie
 	md["Content-Type"] = p.ContentType()
 	md["Micro-Topic"] = p.Topic()
 
-	cf, err := g.newGRPCCodec(p.ContentType())
-	if err != nil {
-		return errors.InternalServerError("go.micro.client", err.Error())
-	}
-
-	var body []byte
-
 	// passed in raw data
 	if d, ok := p.Payload().(*raw.Frame); ok {
 		body = d.Data
 	} else {
+		// use codec for payload
+		cf, err := g.newGRPCCodec(p.ContentType())
+		if err != nil {
+			return errors.InternalServerError("go.micro.client", err.Error())
+		}
 		// set the body
 		b, err := cf.Marshal(p.Payload())
 		if err != nil {
@@ -622,13 +641,6 @@ func (g *grpcClient) Publish(ctx context.Context, p client.Message, opts ...clie
 		body = b
 	}
 
-	if !g.once.Load().(bool) {
-		if err = g.opts.Broker.Connect(); err != nil {
-			return errors.InternalServerError("go.micro.client", err.Error())
-		}
-		g.once.Store(true)
-	}
-
 	topic := p.Topic()
 
 	// get the exchange

client/grpc/grpc_test.go

@@ -5,19 +5,18 @@ import (
 	"net"
 	"testing"
 
-	"github.com/asim/go-micro/plugins/registry/memory/v3"
-	"github.com/asim/go-micro/v3/client"
-	"github.com/asim/go-micro/v3/errors"
-	"github.com/asim/go-micro/v3/registry"
-	"github.com/asim/go-micro/v3/selector"
+	"github.com/micro/go-micro/v3/client"
+	"github.com/micro/go-micro/v3/errors"
+	"github.com/micro/go-micro/v3/registry"
+	"github.com/micro/go-micro/v3/registry/memory"
+	"github.com/micro/go-micro/v3/router"
+	regRouter "github.com/micro/go-micro/v3/router/registry"
 	pgrpc "google.golang.org/grpc"
 	pb "google.golang.org/grpc/examples/helloworld/helloworld"
 )
 
 // server is used to implement helloworld.GreeterServer.
-type greeterServer struct {
-	pb.UnimplementedGreeterServer
-}
+type greeterServer struct{}
 
 // SayHello implements helloworld.GreeterServer
 func (g *greeterServer) SayHello(ctx context.Context, in *pb.HelloRequest) (*pb.HelloReply, error) {
@@ -58,16 +57,11 @@ func TestGRPCClient(t *testing.T) {
 		},
 	})
 
-	// create selector
-	se := selector.NewSelector(
-		selector.Registry(r),
-	)
+	// create router
+	rtr := regRouter.NewRouter(router.Registry(r))
 
 	// create client
-	c := NewClient(
-		client.Registry(r),
-		client.Selector(se),
-	)
+	c := NewClient(client.Router(rtr))
 
 	testMethods := []string{
 		"/helloworld.Greeter/SayHello",

client/grpc/message.go

@@ -1,7 +1,7 @@
 package grpc
 
 import (
-	"github.com/asim/go-micro/v3/client"
+	"github.com/micro/go-micro/v3/client"
 )
 
 type grpcEvent struct {

client/grpc/options.go

@@ -5,7 +5,7 @@ import (
 	"context"
 	"crypto/tls"
 
-	"github.com/asim/go-micro/v3/client"
+	"github.com/micro/go-micro/v3/client"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/encoding"
 )

client/grpc/request.go

@@ -4,8 +4,8 @@ import (
 	"fmt"
 	"strings"
 
-	"github.com/asim/go-micro/v3/client"
-	"github.com/asim/go-micro/v3/codec"
+	"github.com/micro/go-micro/v3/client"
+	"github.com/micro/go-micro/v3/codec"
 )
 
 type grpcRequest struct {

client/grpc/response.go

@@ -3,8 +3,8 @@ package grpc
 import (
 	"strings"
 
-	"github.com/asim/go-micro/v3/codec"
-	"github.com/asim/go-micro/v3/codec/bytes"
+	"github.com/micro/go-micro/v3/codec"
+	"github.com/micro/go-micro/v3/codec/bytes"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/encoding"
 )

client/grpc/stream.go

@@ -5,17 +5,19 @@ import (
 	"io"
 	"sync"
 
-	"github.com/asim/go-micro/v3/client"
+	"github.com/micro/go-micro/v3/client"
 	"google.golang.org/grpc"
 )
 
 // Implements the streamer interface
 type grpcStream struct {
+	// embed so we can access if need be
+	grpc.ClientStream
+
 	sync.RWMutex
 	closed   bool
 	err      error
 	conn     *grpc.ClientConn
-	stream   grpc.ClientStream
 	request  client.Request
 	response client.Response
 	context  context.Context
@@ -35,7 +37,7 @@ func (g *grpcStream) Response() client.Response {
 }
 
 func (g *grpcStream) Send(msg interface{}) error {
-	if err := g.stream.SendMsg(msg); err != nil {
+	if err := g.ClientStream.SendMsg(msg); err != nil {
 		g.setError(err)
 		return err
 	}
@@ -44,7 +46,8 @@ func (g *grpcStream) Send(msg interface{}) error {
 
 func (g *grpcStream) Recv(msg interface{}) (err error) {
 	defer g.setError(err)
-	if err = g.stream.RecvMsg(msg); err != nil {
+
+	if err = g.ClientStream.RecvMsg(msg); err != nil {
 		// #202 - inconsistent gRPC stream behavior
 		// the only way to tell if the stream is done is when we get a EOF on the Recv
 		// here we should close the underlying gRPC ClientConn
@@ -52,7 +55,10 @@ func (g *grpcStream) Recv(msg interface{}) (err error) {
 		if err == io.EOF && closeErr != nil {
 			err = closeErr
 		}
+
+		return err
 	}
+
 	return
 }
 
@@ -83,6 +89,6 @@ func (g *grpcStream) Close() error {
 	// cancel the context
 	g.cancel()
 	g.closed = true
-	g.stream.CloseSend()
+	g.ClientStream.CloseSend()
 	return g.conn.Close()
 }

client/lookup.go

@@ -0,0 +1,50 @@
+package client
+
+import (
+	"context"
+	"sort"
+
+	"github.com/micro/go-micro/v3/errors"
+	"github.com/micro/go-micro/v3/router"
+)
+
+// LookupFunc is used to lookup routes for a service
+type LookupFunc func(context.Context, Request, CallOptions) ([]string, error)
+
+// LookupRoute for a request using the router and then choose one using the selector
+func LookupRoute(ctx context.Context, req Request, opts CallOptions) ([]string, error) {
+	// check to see if an address was provided as a call option
+	if len(opts.Address) > 0 {
+		return opts.Address, nil
+	}
+
+	// construct the router query
+	query := []router.LookupOption{}
+
+	// if a custom network was requested, pass this to the router. By default the router will use it's
+	// own network, which is set during initialisation.
+	if len(opts.Network) > 0 {
+		query = append(query, router.LookupNetwork(opts.Network))
+	}
+
+	// lookup the routes which can be used to execute the request
+	routes, err := opts.Router.Lookup(req.Service(), query...)
+	if err == router.ErrRouteNotFound {
+		return nil, errors.InternalServerError("go.micro.client", "service %s: %s", req.Service(), err.Error())
+	} else if err != nil {
+		return nil, errors.InternalServerError("go.micro.client", "error getting next %s node: %s", req.Service(), err.Error())
+	}
+
+	// sort by lowest metric first
+	sort.Slice(routes, func(i, j int) bool {
+		return routes[i].Metric < routes[j].Metric
+	})
+
+	var addrs []string
+
+	for _, route := range routes {
+		addrs = append(addrs, route.Address)
+	}
+
+	return addrs, nil
+}

client/mucp/common_test.go

@@ -1,7 +1,7 @@
-package client
+package mucp
 
 import (
-	"github.com/asim/go-micro/v3/registry"
+	"github.com/micro/go-micro/v3/registry"
 )
 
 var (

client/mucp/mucp.go

@@ -1,4 +1,5 @@
-package client
+// Package mucp provides a transport agnostic RPC client
+package mucp
 
 import (
 	"context"
@@ -6,29 +7,28 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/asim/go-micro/v3/broker"
-	"github.com/asim/go-micro/v3/codec"
-	raw "github.com/asim/go-micro/v3/codec/bytes"
-	"github.com/asim/go-micro/v3/errors"
-	"github.com/asim/go-micro/v3/metadata"
-	"github.com/asim/go-micro/v3/registry"
-	"github.com/asim/go-micro/v3/selector"
-	"github.com/asim/go-micro/v3/transport"
-	"github.com/asim/go-micro/v3/util/buf"
-	"github.com/asim/go-micro/v3/util/net"
-	"github.com/asim/go-micro/v3/util/pool"
 	"github.com/google/uuid"
+	"github.com/micro/go-micro/v3/broker"
+	"github.com/micro/go-micro/v3/client"
+	"github.com/micro/go-micro/v3/codec"
+	raw "github.com/micro/go-micro/v3/codec/bytes"
+	"github.com/micro/go-micro/v3/errors"
+	"github.com/micro/go-micro/v3/metadata"
+	"github.com/micro/go-micro/v3/network/transport"
+	"github.com/micro/go-micro/v3/util/buf"
+	"github.com/micro/go-micro/v3/util/pool"
 )
 
 type rpcClient struct {
-	seq  uint64
 	once atomic.Value
-	opts Options
+	opts client.Options
 	pool pool.Pool
+	seq  uint64
 }
 
-func newRpcClient(opt ...Option) Client {
-	opts := NewOptions(opt...)
+// NewClient returns a new micro client interface
+func NewClient(opt ...client.Option) client.Client {
+	opts := client.NewOptions(opt...)
 
 	p := pool.NewPool(
 		pool.Size(opts.PoolSize),
@@ -43,7 +43,7 @@ func newRpcClient(opt ...Option) Client {
 	}
 	rc.once.Store(false)
 
-	c := Client(rc)
+	c := client.Client(rc)
 
 	// wrap in reverse
 	for i := len(opts.Wrappers); i > 0; i-- {
@@ -63,9 +63,7 @@ func (r *rpcClient) newCodec(contentType string) (codec.NewCodec, error) {
 	return nil, fmt.Errorf("Unsupported Content-Type: %s", contentType)
 }
 
-func (r *rpcClient) call(ctx context.Context, node *registry.Node, req Request, resp interface{}, opts CallOptions) error {
-	address := node.Address
-
+func (r *rpcClient) call(ctx context.Context, addr string, req client.Request, resp interface{}, opts client.CallOptions) error {
 	msg := &transport.Message{
 		Header: make(map[string]string),
 	}
@@ -89,16 +87,9 @@ func (r *rpcClient) call(ctx context.Context, node *registry.Node, req Request,
 	// set the accept header
 	msg.Header["Accept"] = req.ContentType()
 
-	// setup old protocol
-	cf := setupProtocol(msg, node)
-
-	// no codec specified
-	if cf == nil {
-		var err error
-		cf, err = r.newCodec(req.ContentType())
-		if err != nil {
-			return errors.InternalServerError("go.micro.client", err.Error())
-		}
+	cf, err := r.newCodec(req.ContentType())
+	if err != nil {
+		return errors.InternalServerError("go.micro.client", err.Error())
 	}
 
 	dOpts := []transport.DialOption{
@@ -109,7 +100,7 @@ func (r *rpcClient) call(ctx context.Context, node *registry.Node, req Request,
 		dOpts = append(dOpts, transport.WithTimeout(opts.DialTimeout))
 	}
 
-	c, err := r.pool.Get(address, dOpts...)
+	c, err := r.pool.Get(addr, dOpts...)
 	if err != nil {
 		return errors.InternalServerError("go.micro.client", "connection error: %v", err)
 	}
@@ -182,9 +173,7 @@ func (r *rpcClient) call(ctx context.Context, node *registry.Node, req Request,
 	return nil
 }
 
-func (r *rpcClient) stream(ctx context.Context, node *registry.Node, req Request, opts CallOptions) (Stream, error) {
-	address := node.Address
-
+func (r *rpcClient) stream(ctx context.Context, addr string, req client.Request, opts client.CallOptions) (client.Stream, error) {
 	msg := &transport.Message{
 		Header: make(map[string]string),
 	}
@@ -205,16 +194,9 @@ func (r *rpcClient) stream(ctx context.Context, node *registry.Node, req Request
 	// set the accept header
 	msg.Header["Accept"] = req.ContentType()
 
-	// set old codecs
-	cf := setupProtocol(msg, node)
-
-	// no codec specified
-	if cf == nil {
-		var err error
-		cf, err = r.newCodec(req.ContentType())
-		if err != nil {
-			return nil, errors.InternalServerError("go.micro.client", err.Error())
-		}
+	cf, err := r.newCodec(req.ContentType())
+	if err != nil {
+		return nil, errors.InternalServerError("go.micro.client", err.Error())
 	}
 
 	dOpts := []transport.DialOption{
@@ -225,7 +207,7 @@ func (r *rpcClient) stream(ctx context.Context, node *registry.Node, req Request
 		dOpts = append(dOpts, transport.WithTimeout(opts.DialTimeout))
 	}
 
-	c, err := r.opts.Transport.Dial(address, dOpts...)
+	c, err := r.opts.Transport.Dial(addr, dOpts...)
 	if err != nil {
 		return nil, errors.InternalServerError("go.micro.client", "connection error: %v", err)
 	}
@@ -292,7 +274,7 @@ func (r *rpcClient) stream(ctx context.Context, node *registry.Node, req Request
 	return stream, nil
 }
 
-func (r *rpcClient) Init(opts ...Option) error {
+func (r *rpcClient) Init(opts ...client.Option) error {
 	size := r.opts.PoolSize
 	ttl := r.opts.PoolTTL
 	tr := r.opts.Transport
@@ -316,62 +298,19 @@ func (r *rpcClient) Init(opts ...Option) error {
 	return nil
 }
 
-func (r *rpcClient) Options() Options {
+func (r *rpcClient) Options() client.Options {
 	return r.opts
 }
 
-// next returns an iterator for the next nodes to call
-func (r *rpcClient) next(request Request, opts CallOptions) (selector.Next, error) {
-	// try get the proxy
-	service, address, _ := net.Proxy(request.Service(), opts.Address)
-
-	// return remote address
-	if len(address) > 0 {
-		nodes := make([]*registry.Node, len(address))
-
-		for i, addr := range address {
-			nodes[i] = &registry.Node{
-				Address: addr,
-				// Set the protocol
-				Metadata: map[string]string{
-					"protocol": "mucp",
-				},
-			}
-		}
-
-		// crude return method
-		return func() (*registry.Node, error) {
-			return nodes[time.Now().Unix()%int64(len(nodes))], nil
-		}, nil
-	}
-
-	// get next nodes from the selector
-	next, err := r.opts.Selector.Select(service, opts.SelectOptions...)
-	if err != nil {
-		if err == selector.ErrNotFound {
-			return nil, errors.InternalServerError("go.micro.client", "service %s: %s", service, err.Error())
-		}
-		return nil, errors.InternalServerError("go.micro.client", "error selecting %s node: %s", service, err.Error())
-	}
-
-	return next, nil
-}
-
-func (r *rpcClient) Call(ctx context.Context, request Request, response interface{}, opts ...CallOption) error {
+func (r *rpcClient) Call(ctx context.Context, request client.Request, response interface{}, opts ...client.CallOption) error {
 	// make a copy of call opts
 	callOpts := r.opts.CallOptions
 	for _, opt := range opts {
 		opt(&callOpts)
 	}
 
-	next, err := r.next(request, callOpts)
-	if err != nil {
-		return err
-	}
-
 	// check if we already have a deadline
-	d, ok := ctx.Deadline()
-	if !ok {
+	if d, ok := ctx.Deadline(); !ok {
 		// no deadline so we create a new one
 		var cancel context.CancelFunc
 		ctx, cancel = context.WithTimeout(ctx, callOpts.RequestTimeout)
@@ -379,8 +318,8 @@ func (r *rpcClient) Call(ctx context.Context, request Request, response interfac
 	} else {
 		// got a deadline so no need to setup context
 		// but we need to set the timeout we pass along
-		opt := WithRequestTimeout(time.Until(d))
-		opt(&callOpts)
+		remaining := d.Sub(time.Now())
+		client.WithRequestTimeout(remaining)(&callOpts)
 	}
 
 	// should we noop right here?
@@ -398,6 +337,34 @@ func (r *rpcClient) Call(ctx context.Context, request Request, response interfac
 		rcall = callOpts.CallWrappers[i-1](rcall)
 	}
 
+	// use the router passed as a call option, or fallback to the rpc clients router
+	if callOpts.Router == nil {
+		callOpts.Router = r.opts.Router
+	}
+
+	if callOpts.Selector == nil {
+		callOpts.Selector = r.opts.Selector
+	}
+
+	// inject proxy address
+	// TODO: don't even bother using Lookup/Select in this case
+	if len(r.opts.Proxy) > 0 {
+		callOpts.Address = []string{r.opts.Proxy}
+	}
+
+	// lookup the route to send the reques to
+	// TODO apply any filtering here
+	routes, err := r.opts.Lookup(ctx, request, callOpts)
+	if err != nil {
+		return errors.InternalServerError("go.micro.client", err.Error())
+	}
+
+	// balance the list of nodes
+	next, err := callOpts.Selector.Select(routes)
+	if err != nil {
+		return err
+	}
+
 	// return errors.New("go.micro.client", "request timeout", 408)
 	call := func(i int) error {
 		// call backoff first. Someone may want an initial start delay
@@ -411,19 +378,15 @@ func (r *rpcClient) Call(ctx context.Context, request Request, response interfac
 			time.Sleep(t)
 		}
 
-		// select next node
-		node, err := next()
-		service := request.Service()
-		if err != nil {
-			if err == selector.ErrNotFound {
-				return errors.InternalServerError("go.micro.client", "service %s: %s", service, err.Error())
-			}
-			return errors.InternalServerError("go.micro.client", "error getting next %s node: %s", service, err.Error())
-		}
+		// get the next node
+		node := next()
 
 		// make the call
 		err = rcall(ctx, node, request, response, callOpts)
-		r.opts.Selector.Mark(service, node, err)
+
+		// record the result of the call to inform future routing decisions
+		r.opts.Selector.Record(node, err)
+
 		return err
 	}
 
@@ -431,7 +394,7 @@ func (r *rpcClient) Call(ctx context.Context, request Request, response interfac
 	retries := callOpts.Retries
 
 	// disable retries when using a proxy
-	if _, _, ok := net.Proxy(request.Service(), callOpts.Address); ok {
+	if len(r.opts.Proxy) > 0 {
 		retries = 0
 	}
 
@@ -468,18 +431,13 @@ func (r *rpcClient) Call(ctx context.Context, request Request, response interfac
 	return gerr
 }
 
-func (r *rpcClient) Stream(ctx context.Context, request Request, opts ...CallOption) (Stream, error) {
+func (r *rpcClient) Stream(ctx context.Context, request client.Request, opts ...client.CallOption) (client.Stream, error) {
 	// make a copy of call opts
 	callOpts := r.opts.CallOptions
 	for _, opt := range opts {
 		opt(&callOpts)
 	}
 
-	next, err := r.next(request, callOpts)
-	if err != nil {
-		return nil, err
-	}
-
 	// should we noop right here?
 	select {
 	case <-ctx.Done():
@@ -487,7 +445,35 @@ func (r *rpcClient) Stream(ctx context.Context, request Request, opts ...CallOpt
 	default:
 	}
 
-	call := func(i int) (Stream, error) {
+	// use the router passed as a call option, or fallback to the rpc clients router
+	if callOpts.Router == nil {
+		callOpts.Router = r.opts.Router
+	}
+
+	if callOpts.Selector == nil {
+		callOpts.Selector = r.opts.Selector
+	}
+
+	// inject proxy address
+	// TODO: don't even bother using Lookup/Select in this case
+	if len(r.opts.Proxy) > 0 {
+		callOpts.Address = []string{r.opts.Proxy}
+	}
+
+	// lookup the route to send the reques to
+	// TODO apply any filtering here
+	routes, err := r.opts.Lookup(ctx, request, callOpts)
+	if err != nil {
+		return nil, errors.InternalServerError("go.micro.client", err.Error())
+	}
+
+	// balance the list of nodes
+	next, err := callOpts.Selector.Select(routes)
+	if err != nil {
+		return nil, err
+	}
+
+	call := func(i int) (client.Stream, error) {
 		// call backoff first. Someone may want an initial start delay
 		t, err := callOpts.Backoff(ctx, request, i)
 		if err != nil {
@@ -499,22 +485,20 @@ func (r *rpcClient) Stream(ctx context.Context, request Request, opts ...CallOpt
 			time.Sleep(t)
 		}
 
-		node, err := next()
-		service := request.Service()
-		if err != nil {
-			if err == selector.ErrNotFound {
-				return nil, errors.InternalServerError("go.micro.client", "service %s: %s", service, err.Error())
-			}
-			return nil, errors.InternalServerError("go.micro.client", "error getting next %s node: %s", service, err.Error())
-		}
+		// get the next node
+		node := next()
 
+		// perform the call
 		stream, err := r.stream(ctx, node, request, callOpts)
-		r.opts.Selector.Mark(service, node, err)
+
+		// record the result of the call to inform future routing decisions
+		r.opts.Selector.Record(node, err)
+
 		return stream, err
 	}
 
 	type response struct {
-		stream Stream
+		stream client.Stream
 		err    error
 	}
 
@@ -522,7 +506,7 @@ func (r *rpcClient) Stream(ctx context.Context, request Request, opts ...CallOpt
 	retries := callOpts.Retries
 
 	// disable retries when using a proxy
-	if _, _, ok := net.Proxy(request.Service(), callOpts.Address); ok {
+	if len(r.opts.Proxy) > 0 {
 		retries = 0
 	}
 
@@ -560,8 +544,8 @@ func (r *rpcClient) Stream(ctx context.Context, request Request, opts ...CallOpt
 	return nil, grr
 }
 
-func (r *rpcClient) Publish(ctx context.Context, msg Message, opts ...PublishOption) error {
-	options := PublishOptions{
+func (r *rpcClient) Publish(ctx context.Context, msg client.Message, opts ...client.PublishOption) error {
+	options := client.PublishOptions{
 		Context: context.Background(),
 	}
 	for _, o := range opts {
@@ -629,11 +613,11 @@ func (r *rpcClient) Publish(ctx context.Context, msg Message, opts ...PublishOpt
 	}, broker.PublishContext(options.Context))
 }
 
-func (r *rpcClient) NewMessage(topic string, message interface{}, opts ...MessageOption) Message {
+func (r *rpcClient) NewMessage(topic string, message interface{}, opts ...client.MessageOption) client.Message {
 	return newMessage(topic, message, r.opts.ContentType, opts...)
 }
 
-func (r *rpcClient) NewRequest(service, method string, request interface{}, reqOpts ...RequestOption) Request {
+func (r *rpcClient) NewRequest(service, method string, request interface{}, reqOpts ...client.RequestOption) client.Request {
 	return newRequest(service, method, request, r.opts.ContentType, reqOpts...)
 }
 

client/mucp/mucp_codec.go

@@ -1,19 +1,19 @@
-package client
+package mucp
 
 import (
 	"bytes"
 	errs "errors"
 
-	"github.com/asim/go-micro/v3/codec"
-	raw "github.com/asim/go-micro/v3/codec/bytes"
-	"github.com/asim/go-micro/v3/codec/grpc"
-	"github.com/asim/go-micro/v3/codec/json"
-	"github.com/asim/go-micro/v3/codec/jsonrpc"
-	"github.com/asim/go-micro/v3/codec/proto"
-	"github.com/asim/go-micro/v3/codec/protorpc"
-	"github.com/asim/go-micro/v3/errors"
-	"github.com/asim/go-micro/v3/registry"
-	"github.com/asim/go-micro/v3/transport"
+	"github.com/micro/go-micro/v3/codec"
+	raw "github.com/micro/go-micro/v3/codec/bytes"
+	"github.com/micro/go-micro/v3/codec/grpc"
+	"github.com/micro/go-micro/v3/codec/json"
+	"github.com/micro/go-micro/v3/codec/jsonrpc"
+	"github.com/micro/go-micro/v3/codec/proto"
+	"github.com/micro/go-micro/v3/codec/protorpc"
+	"github.com/micro/go-micro/v3/errors"
+	"github.com/micro/go-micro/v3/network/transport"
+	"github.com/micro/go-micro/v3/registry"
 )
 
 const (
@@ -129,10 +129,8 @@ func setHeaders(m *codec.Message, stream string) {
 
 // setupProtocol sets up the old protocol
 func setupProtocol(msg *transport.Message, node *registry.Node) codec.NewCodec {
-	protocol := node.Metadata["protocol"]
-
-	// got protocol
-	if len(protocol) > 0 {
+	// get the protocol from node metadata
+	if protocol := node.Metadata["protocol"]; len(protocol) > 0 {
 		return nil
 	}