Merge pull request #598 from unistack-org/ipv6fix

bunch of other ipv6 fixes

broker/http_broker.go

@@ -627,12 +627,6 @@ func (h *httpBroker) Subscribe(topic string, handler Handler, opts ...SubscribeO
 		return nil, err
 	}
 
-	// ipv6 addr
-	if addr == "::" {
-		// ipv6 addr
-		addr = fmt.Sprintf("[%s]", addr)
-	}
-
 	// create unique id
 	id := h.id + "." + uuid.New().String()
 
@@ -645,7 +639,7 @@ func (h *httpBroker) Subscribe(topic string, handler Handler, opts ...SubscribeO
 	// register service
 	node := &registry.Node{
 		Id:      id,
-		Address: fmt.Sprintf("%s:%s", addr, port),
+		Address: mnet.HostPort(addr, port),
 		Metadata: map[string]string{
 			"secure": fmt.Sprintf("%t", secure),
 		},

broker/memory/memory.go

@@ -3,15 +3,20 @@ package memory
 
 import (
 	"errors"
+	"math/rand"
 	"sync"
+	"time"
 
 	"github.com/google/uuid"
 	"github.com/micro/go-micro/broker"
+	maddr "github.com/micro/go-micro/util/addr"
+	mnet "github.com/micro/go-micro/util/net"
 )
 
 type memoryBroker struct {
 	opts broker.Options
 
+	addr string
 	sync.RWMutex
 	connected   bool
 	Subscribers map[string][]*memorySubscriber
@@ -35,7 +40,7 @@ func (m *memoryBroker) Options() broker.Options {
 }
 
 func (m *memoryBroker) Address() string {
-	return ""
+	return m.addr
 }
 
 func (m *memoryBroker) Connect() error {
@@ -46,6 +51,15 @@ func (m *memoryBroker) Connect() error {
 		return nil
 	}
 
+	addr, err := maddr.Extract("::")
+	if err != nil {
+		return err
+	}
+	i := rand.Intn(20000)
+	// set addr with port
+	addr = mnet.HostPort(addr, 10000+i)
+
+	m.addr = addr
 	m.connected = true
 
 	return nil
@@ -171,6 +185,7 @@ func (m *memorySubscriber) Unsubscribe() error {
 
 func NewBroker(opts ...broker.Option) broker.Broker {
 	var options broker.Options
+	rand.Seed(time.Now().UnixNano())
 	for _, o := range opts {
 		o(&options)
 	}

server/grpc/grpc.go

@@ -22,6 +22,7 @@ import (
 	"github.com/micro/go-micro/util/addr"
 	mgrpc "github.com/micro/go-micro/util/grpc"
 	"github.com/micro/go-micro/util/log"
+	mnet "github.com/micro/go-micro/util/net"
 
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
@@ -519,15 +520,12 @@ func (g *grpcServer) Register() error {
 		advt = config.Address
 	}
 
-	if idx := strings.Count(advt, ":"); idx > 1 {
+	if cnt := strings.Count(advt, ":"); cnt >= 1 {
 		// ipv6 address in format [host]:port or ipv4 host:port
 		host, port, err = net.SplitHostPort(advt)
 		if err != nil {
 			return err
 		}
-		if host == "::" {
-			host = fmt.Sprintf("[%s]", host)
-		}
 	} else {
 		host = advt
 	}
@@ -540,7 +538,7 @@ func (g *grpcServer) Register() error {
 	// register service
 	node := &registry.Node{
 		Id:       config.Name + "-" + config.Id,
-		Address:  fmt.Sprintf("%s:%s", addr, port),
+		Address:  mnet.HostPort(addr, port),
 		Metadata: config.Metadata,
 	}
 
@@ -649,15 +647,12 @@ func (g *grpcServer) Deregister() error {
 		advt = config.Address
 	}
 
-	if idx := strings.Count(advt, ":"); idx > 1 {
+	if cnt := strings.Count(advt, ":"); cnt >= 1 {
 		// ipv6 address in format [host]:port or ipv4 host:port
 		host, port, err = net.SplitHostPort(advt)
 		if err != nil {
 			return err
 		}
-		if host == "::" {
-			host = fmt.Sprintf("[%s]", host)
-		}
 	} else {
 		host = advt
 	}
@@ -669,7 +664,7 @@ func (g *grpcServer) Deregister() error {
 
 	node := &registry.Node{
 		Id:      config.Name + "-" + config.Id,
-		Address: fmt.Sprintf("%s:%s", addr, port),
+		Address: mnet.HostPort(addr, port),
 	}
 
 	service := &registry.Service{

server/rpc_server.go

@@ -18,6 +18,7 @@ import (
 	"github.com/micro/go-micro/transport"
 	"github.com/micro/go-micro/util/addr"
 	log "github.com/micro/go-micro/util/log"
+	mnet "github.com/micro/go-micro/util/net"
 )
 
 type rpcServer struct {
@@ -293,15 +294,12 @@ func (s *rpcServer) Register() error {
 		advt = config.Address
 	}
 
-	if idx := strings.Count(advt, ":"); idx > 1 {
+	if cnt := strings.Count(advt, ":"); cnt >= 1 {
 		// ipv6 address in format [host]:port or ipv4 host:port
 		host, port, err = net.SplitHostPort(advt)
 		if err != nil {
 			return err
 		}
-		if host == "::" {
-			host = fmt.Sprintf("[%s]", host)
-		}
 	} else {
 		host = advt
 	}
@@ -320,7 +318,7 @@ func (s *rpcServer) Register() error {
 	// register service
 	node := &registry.Node{
 		Id:       config.Name + "-" + config.Id,
-		Address:  fmt.Sprintf("%s:%s", addr, port),
+		Address:  mnet.HostPort(addr, port),
 		Metadata: md,
 	}
 
@@ -434,15 +432,12 @@ func (s *rpcServer) Deregister() error {
 		advt = config.Address
 	}
 
-	if idx := strings.Count(advt, ":"); idx > 1 {
+	if cnt := strings.Count(advt, ":"); cnt >= 1 {
 		// ipv6 address in format [host]:port or ipv4 host:port
 		host, port, err = net.SplitHostPort(advt)
 		if err != nil {
 			return err
 		}
-		if host == "::" {
-			host = fmt.Sprintf("[%s]", host)
-		}
 	} else {
 		host = advt
 	}
@@ -454,7 +449,7 @@ func (s *rpcServer) Deregister() error {
 
 	node := &registry.Node{
 		Id:      config.Name + "-" + config.Id,
-		Address: fmt.Sprintf("%s:%s", addr, port),
+		Address: mnet.HostPort(addr, port),
 	}
 
 	service := &registry.Service{

transport/memory/memory.go

@@ -5,11 +5,13 @@ import (
 	"errors"
 	"fmt"
 	"math/rand"
-	"strings"
+	"net"
 	"sync"
 	"time"
 
 	"github.com/micro/go-micro/transport"
+	maddr "github.com/micro/go-micro/util/addr"
+	mnet "github.com/micro/go-micro/util/net"
 )
 
 type memorySocket struct {
@@ -170,15 +172,25 @@ func (m *memoryTransport) Listen(addr string, opts ...transport.ListenOption) (t
 		o(&options)
 	}
 
-	parts := strings.Split(addr, ":")
+	host, port, err := net.SplitHostPort(addr)
+	if err != nil {
+		return nil, err
+	}
+
+	addr, err = maddr.Extract(host)
+	if err != nil {
+		return nil, err
+	}
 
 	// if zero port then randomly assign one
-	if len(parts) > 1 && parts[len(parts)-1] == "0" {
+	if len(port) > 0 && port == "0" {
 		i := rand.Intn(20000)
-		// set addr with port
+		port = fmt.Sprintf("%d", 10000+i)
-		addr = fmt.Sprintf("%s:%d", parts[:len(parts)-1], 10000+i)
 	}
 
+	// set addr with port
+	addr = mnet.HostPort(addr, port)
+
 	if _, ok := m.listeners[addr]; ok {
 		return nil, errors.New("already listening on " + addr)
 	}

transport/memory/memory_test.go

@@ -10,7 +10,7 @@ func TestMemoryTransport(t *testing.T) {
 	tr := NewTransport()
 
 	// bind / listen
-	l, err := tr.Listen("localhost:8080")
+	l, err := tr.Listen("127.0.0.1:8080")
 	if err != nil {
 		t.Fatalf("Unexpected error listening %v", err)
 	}
@@ -37,7 +37,7 @@ func TestMemoryTransport(t *testing.T) {
 	}()
 
 	// dial
-	c, err := tr.Dial("localhost:8080")
+	c, err := tr.Dial("127.0.0.1:8080")
 	if err != nil {
 		t.Fatalf("Unexpected error dialing %v", err)
 	}

util/net/net.go

@@ -8,6 +8,15 @@ import (
 	"strings"
 )
 
+// HostPort format addr and port suitable for dial
+func HostPort(addr string, port interface{}) string {
+	host := addr
+	if strings.Count(addr, ":") > 0 {
+		host = fmt.Sprintf("[%s]", addr)
+	}
+	return fmt.Sprintf("%s:%v", host, port)
+}
+
 // Listen takes addr:portmin-portmax and binds to the first available port
 // Example: Listen("localhost:5000-6000", fn)
 func Listen(addr string, fn func(string) (net.Listener, error)) (net.Listener, error) {
@@ -22,10 +31,6 @@ func Listen(addr string, fn func(string) (net.Listener, error)) (net.Listener, e
 		return nil, err
 	}
 
-	if host == "::" {
-		host = fmt.Sprintf("[%s]", host)
-	}
-
 	// try to extract port range
 	prange := strings.Split(ports, "-")
 
@@ -51,7 +56,7 @@ func Listen(addr string, fn func(string) (net.Listener, error)) (net.Listener, e
 	// range the ports
 	for port := min; port <= max; port++ {
 		// try bind to host:port
-		ln, err := fn(fmt.Sprintf("%s:%d", host, port))
+		ln, err := fn(HostPort(host, port))
 		if err == nil {
 			return ln, nil
 		}