Merge branch 'master' into selector

service.go

@@ -14,6 +14,8 @@ import (
 
 type service struct {
 	opts Options
+
+	init chan bool
 }
 
 func newService(opts ...Option) Service {
@@ -28,6 +30,7 @@ func newService(opts ...Option) Service {
 
 	return &service{
 		opts: options,
+		init: make(chan bool),
 	}
 }
 
@@ -49,24 +52,36 @@ func (s *service) run(exit chan bool) {
 	}
 }
 
+// Init initialises options. Additionally it calls cmd.Init
+// which parses command line flags. cmd.Init is only called
+// on first Init.
 func (s *service) Init(opts ...Option) {
-	// We might get more command flags or the action here
-	// This is pretty ugly, find a better way
-	options := newOptions()
-	options.Cmd = s.opts.Cmd
-	for _, o := range opts {
-		o(&options)
-	}
-	s.opts.Cmd = options.Cmd
+	// If <-s.init blocks, Init has not been called yet
+	// so we can call cmd.Init once.
+	select {
+	case <-s.init:
+	default:
+		// close init
+		close(s.init)
 
-	// Initialise the command flags, overriding new service
-	s.opts.Cmd.Init(
-		cmd.Broker(&s.opts.Broker),
-		cmd.Registry(&s.opts.Registry),
-		cmd.Transport(&s.opts.Transport),
-		cmd.Client(&s.opts.Client),
-		cmd.Server(&s.opts.Server),
-	)
+		// We might get more command flags or the action here
+		// This is pretty ugly, find a better way
+		options := newOptions()
+		options.Cmd = s.opts.Cmd
+		for _, o := range opts {
+			o(&options)
+		}
+		s.opts.Cmd = options.Cmd
+
+		// Initialise the command flags, overriding new service
+		s.opts.Cmd.Init(
+			cmd.Broker(&s.opts.Broker),
+			cmd.Registry(&s.opts.Registry),
+			cmd.Transport(&s.opts.Transport),
+			cmd.Client(&s.opts.Client),
+			cmd.Server(&s.opts.Server),
+		)
+	}
 
 	// Update any options to override command flags
 	for _, o := range opts {

transport/http_transport.go

@@ -174,6 +174,10 @@ func (h *httpTransportClient) Recv(m *Message) error {
 		return err
 	}
 
+	if rsp.StatusCode != 200 {
+		return errors.New(rsp.Status + ": " + string(b))
+	}
+
 	mr := &Message{
 		Header: make(map[string]string),
 		Body:   b,

transport/http_transport_test.go

@@ -1,13 +1,11 @@
-package transport_test
+package transport
 
 import (
 	"strings"
 	"testing"
-
-	"github.com/micro/go-micro/transport"
 )
 
-func expectedPort(t *testing.T, expected string, lsn transport.Listener) {
+func expectedPort(t *testing.T, expected string, lsn Listener) {
 	parts := strings.Split(lsn.Addr(), ":")
 	port := parts[len(parts)-1]
 
@@ -18,7 +16,7 @@ func expectedPort(t *testing.T, expected string, lsn transport.Listener) {
 }
 
 func TestHTTPTransportPortRange(t *testing.T) {
-	tp := transport.NewTransport()
+	tp := NewTransport()
 
 	lsn1, err := tp.Listen(":44444-44448")
 	if err != nil {
@@ -43,7 +41,7 @@ func TestHTTPTransportPortRange(t *testing.T) {
 }
 
 func TestHTTPTransportCommunication(t *testing.T) {
-	tr := transport.NewTransport()
+	tr := NewTransport()
 
 	l, err := tr.Listen(":0")
 	if err != nil {
@@ -51,17 +49,15 @@ func TestHTTPTransportCommunication(t *testing.T) {
 	}
 	defer l.Close()
 
-	fn := func(sock transport.Socket) {
+	fn := func(sock Socket) {
 		defer sock.Close()
 
 		for {
-			var m transport.Message
+			var m Message
 			if err := sock.Recv(&m); err != nil {
 				return
 			}
 
-			t.Logf("Successfully received %+v", m)
-
 			if err := sock.Send(&m); err != nil {
 				return
 			}
@@ -86,7 +82,7 @@ func TestHTTPTransportCommunication(t *testing.T) {
 	}
 	defer c.Close()
 
-	m := transport.Message{
+	m := Message{
 		Header: map[string]string{
 			"Content-Type": "application/json",
 		},
@@ -97,7 +93,7 @@ func TestHTTPTransportCommunication(t *testing.T) {
 		t.Errorf("Unexpected send err: %v", err)
 	}
 
-	var rm transport.Message
+	var rm Message
 
 	if err := c.Recv(&rm); err != nil {
 		t.Errorf("Unexpected recv err: %v", err)
@@ -109,3 +105,70 @@ func TestHTTPTransportCommunication(t *testing.T) {
 
 	close(done)
 }
+
+func TestHTTPTransportError(t *testing.T) {
+	tr := NewTransport()
+
+	l, err := tr.Listen(":0")
+	if err != nil {
+		t.Errorf("Unexpected listen err: %v", err)
+	}
+	defer l.Close()
+
+	fn := func(sock Socket) {
+		defer sock.Close()
+
+		for {
+			var m Message
+			if err := sock.Recv(&m); err != nil {
+				t.Fatal(err)
+			}
+
+			sock.(*httpTransportSocket).error(&Message{
+				Body: []byte(`an error occurred`),
+			})
+		}
+	}
+
+	done := make(chan bool)
+
+	go func() {
+		if err := l.Accept(fn); err != nil {
+			select {
+			case <-done:
+			default:
+				t.Errorf("Unexpected accept err: %v", err)
+			}
+		}
+	}()
+
+	c, err := tr.Dial(l.Addr())
+	if err != nil {
+		t.Errorf("Unexpected dial err: %v", err)
+	}
+	defer c.Close()
+
+	m := Message{
+		Header: map[string]string{
+			"Content-Type": "application/json",
+		},
+		Body: []byte(`{"message": "Hello World"}`),
+	}
+
+	if err := c.Send(&m); err != nil {
+		t.Errorf("Unexpected send err: %v", err)
+	}
+
+	var rm Message
+
+	err = c.Recv(&rm)
+	if err == nil {
+		t.Fatal("Expected error but got nil")
+	}
+
+	if err.Error() != "500 Internal Server Error: an error occurred" {
+		t.Fatalf("Did not receive expected error, got: %v", err)
+	}
+
+	close(done)
+}