Merge branch 'master' into micro

broker/broker.go

@@ -4,28 +4,35 @@ type Broker interface {
 	Address() string
 	Connect() error
 	Disconnect() error
-	Init() error
-	Publish(string, *Message) error
-	Subscribe(string, Handler) (Subscriber, error)
+	Init(...Option) error
+	Publish(string, *Message, ...PublishOption) error
+	Subscribe(string, Handler, ...SubscribeOption) (Subscriber, error)
 	String() string
 }
 
-type Handler func(*Message)
+// 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(Publication) error
 
 type Message struct {
 	Header map[string]string
 	Body   []byte
 }
 
+// Publication is given to a subscription handler for processing
+type Publication interface {
+	Topic() string
+	Message() *Message
+	Ack() error
+}
+
 type Subscriber interface {
+	Config() SubscribeOptions
 	Topic() string
 	Unsubscribe() error
 }
 
-type options struct{}
-
-type Option func(*options)
-
 var (
 	DefaultBroker Broker = newHttpBroker([]string{})
 )
@@ -34,8 +41,8 @@ func NewBroker(addrs []string, opt ...Option) Broker {
 	return newHttpBroker(addrs, opt...)
 }
 
-func Init() error {
-	return DefaultBroker.Init()
+func Init(opts ...Option) error {
+	return DefaultBroker.Init(opts...)
 }
 
 func Connect() error {
@@ -46,12 +53,12 @@ func Disconnect() error {
 	return DefaultBroker.Disconnect()
 }
 
-func Publish(topic string, msg *Message) error {
-	return DefaultBroker.Publish(topic, msg)
+func Publish(topic string, msg *Message, opts ...PublishOption) error {
+	return DefaultBroker.Publish(topic, msg, opts...)
 }
 
-func Subscribe(topic string, handler Handler) (Subscriber, error) {
-	return DefaultBroker.Subscribe(topic, handler)
+func Subscribe(topic string, handler Handler, opts ...SubscribeOption) (Subscriber, error) {
+	return DefaultBroker.Subscribe(topic, handler, opts...)
 }
 
 func String() string {

broker/http_broker.go

@@ -4,12 +4,15 @@ import (
 	"bytes"
 	"encoding/json"
 	"fmt"
+	"io"
 	"io/ioutil"
+	"math/rand"
 	"net"
 	"net/http"
 	"strconv"
 	"strings"
 	"sync"
+	"time"
 
 	log "github.com/golang/glog"
 	"github.com/micro/go-micro/errors"
@@ -17,6 +20,10 @@ import (
 	"github.com/pborman/uuid"
 )
 
+// HTTP Broker is a placeholder for actual message brokers.
+// This should not really be used in production but useful
+// in developer where you want zero dependencies.
+
 type httpBroker struct {
 	id          string
 	address     string
@@ -29,6 +36,7 @@ type httpBroker struct {
 }
 
 type httpSubscriber struct {
+	opts  SubscribeOptions
 	id    string
 	topic string
 	ch    chan *httpSubscriber
@@ -36,10 +44,20 @@ type httpSubscriber struct {
 	svc   *registry.Service
 }
 
+type httpPublication struct {
+	m *Message
+	t string
+}
+
 var (
-	DefaultSubPath = "/_sub"
+	DefaultSubPath   = "/_sub"
+	broadcastVersion = "ff.http.broadcast"
 )
 
+func init() {
+	rand.Seed(time.Now().Unix())
+}
+
 func newHttpBroker(addrs []string, opt ...Option) Broker {
 	addr := ":0"
 	if len(addrs) > 0 && len(addrs[0]) > 0 {
@@ -55,6 +73,22 @@ func newHttpBroker(addrs []string, opt ...Option) Broker {
 	}
 }
 
+func (h *httpPublication) Ack() error {
+	return nil
+}
+
+func (h *httpPublication) Message() *Message {
+	return h.m
+}
+
+func (h *httpPublication) Topic() string {
+	return h.t
+}
+
+func (h *httpSubscriber) Config() SubscribeOptions {
+	return h.opts
+}
+
 func (h *httpSubscriber) Topic() string {
 	return h.topic
 }
@@ -150,9 +184,10 @@ func (h *httpBroker) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 		return
 	}
 
+	p := &httpPublication{m: m, t: topic}
 	h.RLock()
 	for _, subscriber := range h.subscribers[topic] {
-		subscriber.fn(m)
+		subscriber.fn(p)
 	}
 	h.RUnlock()
 }
@@ -169,7 +204,7 @@ func (h *httpBroker) Disconnect() error {
 	return h.stop()
 }
 
-func (h *httpBroker) Init() error {
+func (h *httpBroker) Init(opts ...Option) error {
 	if len(h.id) == 0 {
 		h.id = "broker-" + uuid.NewUUID().String()
 	}
@@ -178,7 +213,7 @@ func (h *httpBroker) Init() error {
 	return nil
 }
 
-func (h *httpBroker) Publish(topic string, msg *Message) error {
+func (h *httpBroker) Publish(topic string, msg *Message, opts ...PublishOption) error {
 	s, err := registry.GetService("topic:" + topic)
 	if err != nil {
 		return err
@@ -190,37 +225,70 @@ func (h *httpBroker) Publish(topic string, msg *Message) error {
 		return err
 	}
 
-	for _, service := range s {
-		for _, node := range service.Nodes {
-			r, err := http.Post(fmt.Sprintf("http://%s:%d%s", node.Address, node.Port, DefaultSubPath), "application/json", bytes.NewBuffer(b))
-			if err == nil {
-				r.Body.Close()
-			}
+	fn := func(node *registry.Node, b io.Reader) {
+		r, err := http.Post(fmt.Sprintf("http://%s:%d%s", node.Address, node.Port, DefaultSubPath), "application/json", b)
+		if err == nil {
+			r.Body.Close()
 		}
 	}
+
+	buf := bytes.NewBuffer(nil)
+
+	for _, service := range s {
+		// broadcast version means broadcast to all nodes
+		if service.Version == broadcastVersion {
+			for _, node := range service.Nodes {
+				buf.Reset()
+				buf.Write(b)
+				fn(node, buf)
+			}
+			return nil
+		}
+
+		node := service.Nodes[rand.Int()%len(service.Nodes)]
+		buf.Reset()
+		buf.Write(b)
+		fn(node, buf)
+		return nil
+	}
+
+	buf.Reset()
+	buf = nil
+
 	return nil
 }
 
-func (h *httpBroker) Subscribe(topic string, handler Handler) (Subscriber, error) {
+func (h *httpBroker) Subscribe(topic string, handler Handler, opts ...SubscribeOption) (Subscriber, error) {
+	opt := newSubscribeOptions(opts...)
+
 	// parse address for host, port
 	parts := strings.Split(h.Address(), ":")
 	host := strings.Join(parts[:len(parts)-1], ":")
 	port, _ := strconv.Atoi(parts[len(parts)-1])
 
+	id := uuid.NewUUID().String()
+
 	// register service
 	node := &registry.Node{
-		Id:      h.id,
+		Id:      topic + "." + h.id + "." + id,
 		Address: host,
 		Port:    port,
 	}
 
+	version := opt.Queue
+	if len(version) == 0 {
+		version = broadcastVersion
+	}
+
 	service := &registry.Service{
-		Name:  "topic:" + topic,
-		Nodes: []*registry.Node{node},
+		Name:    "topic:" + topic,
+		Version: version,
+		Nodes:   []*registry.Node{node},
 	}
 
 	subscriber := &httpSubscriber{
-		id:    uuid.NewUUID().String(),
+		opts:  opt,
+		id:    id,
 		topic: topic,
 		ch:    h.unsubscribe,
 		fn:    handler,
@@ -234,7 +302,6 @@ func (h *httpBroker) Subscribe(topic string, handler Handler) (Subscriber, error
 	h.Lock()
 	h.subscribers[topic] = append(h.subscribers[topic], subscriber)
 	h.Unlock()
-
 	return subscriber, nil
 }
 

broker/options.go

@@ -0,0 +1,48 @@
+package broker
+
+type Options struct{}
+
+type PublishOptions struct{}
+
+type SubscribeOptions struct {
+	// AutoAck defaults to true. When a handler returns
+	// with a nil error the message is acked.
+	AutoAck bool
+	// Subscribers with the same queue name
+	// will create a shared subscription where each
+	// receives a subset of messages.
+	Queue string
+}
+
+type Option func(*Options)
+
+type PublishOption func(*PublishOptions)
+
+type SubscribeOption func(*SubscribeOptions)
+
+// DisableAutoAck will disable auto acking of messages
+// after they have been handled.
+func DisableAutoAck() SubscribeOption {
+	return func(o *SubscribeOptions) {
+		o.AutoAck = false
+	}
+}
+
+// QueueName sets the name of the queue to share messages on
+func QueueName(name string) SubscribeOption {
+	return func(o *SubscribeOptions) {
+		o.Queue = name
+	}
+}
+
+func newSubscribeOptions(opts ...SubscribeOption) SubscribeOptions {
+	opt := SubscribeOptions{
+		AutoAck: true,
+	}
+
+	for _, o := range opts {
+		o(&opt)
+	}
+
+	return opt
+}

client/rpc_client.go

@@ -136,9 +136,12 @@ func (r *rpcClient) stream(ctx context.Context, address string, req Request) (St
 		return nil, errors.InternalServerError("go.micro.client", fmt.Sprintf("Error sending request: %v", err))
 	}
 
+	var once sync.Once
 	stream := &rpcStream{
 		context: ctx,
 		request: req,
+		once:    once,
+		closed:  make(chan bool),
 		codec:   newRpcPlusCodec(msg, c, cf),
 	}
 

client/rpc_stream.go

@@ -13,13 +13,25 @@ import (
 type rpcStream struct {
 	sync.RWMutex
 	seq     uint64
-	closed  bool
+	once    sync.Once
+	closed  chan bool
 	err     error
 	request Request
 	codec   clientCodec
 	context context.Context
 }
 
+func (r *rpcStream) isClosed() bool {
+	select {
+	case _, ok := <-r.closed:
+		if !ok {
+			return true
+		}
+	default:
+	}
+	return false
+}
+
 func (r *rpcStream) Context() context.Context {
 	return r.context
 }
@@ -32,7 +44,7 @@ func (r *rpcStream) Send(msg interface{}) error {
 	r.Lock()
 	defer r.Unlock()
 
-	if r.closed {
+	if r.isClosed() {
 		r.err = errShutdown
 		return errShutdown
 	}
@@ -57,14 +69,14 @@ func (r *rpcStream) Recv(msg interface{}) error {
 	r.Lock()
 	defer r.Unlock()
 
-	if r.closed {
+	if r.isClosed() {
 		r.err = errShutdown
 		return errShutdown
 	}
 
 	var resp response
 	if err := r.codec.ReadResponseHeader(&resp); err != nil {
-		if err == io.EOF && !r.closed {
+		if err == io.EOF && !r.isClosed() {
 			r.err = io.ErrUnexpectedEOF
 			return io.ErrUnexpectedEOF
 		}
@@ -91,7 +103,7 @@ func (r *rpcStream) Recv(msg interface{}) error {
 		}
 	}
 
-	if r.err != nil && r.err != io.EOF && !r.closed {
+	if r.err != nil && r.err != io.EOF && !r.isClosed() {
 		log.Println("rpc: client protocol error:", r.err)
 	}
 
@@ -105,8 +117,8 @@ func (r *rpcStream) Error() error {
 }
 
 func (r *rpcStream) Close() error {
-	r.Lock()
-	defer r.Unlock()
-	r.closed = true
+	r.once.Do(func() {
+		close(r.closed)
+	})
 	return r.codec.Close()
 }

cmd/cmd.go

@@ -16,6 +16,7 @@ import (
 	"time"
 
 	"github.com/codegangsta/cli"
+	log "github.com/golang/glog"
 	"github.com/micro/go-micro/broker"
 	"github.com/micro/go-micro/client"
 	"github.com/micro/go-micro/registry"
@@ -26,6 +27,8 @@ import (
 )
 
 var (
+	Actions = []func(*cli.Context){}
+
 	Flags = []cli.Flag{
 		cli.StringFlag{
 			Name:   "server_name",
@@ -100,9 +103,9 @@ var (
 		},
 
 		cli.BoolFlag{
-			Name:   "disable_ping",
-			EnvVar: "MICRO_DISABLE_PING",
-			Usage:  "Disable ping",
+			Name:   "enable_ping",
+			EnvVar: "MICRO_ENABLE_PING",
+			Usage:  "Enable ping",
 		},
 
 		// logging flags
@@ -180,6 +183,7 @@ func ping() {
 	cl := &http.Client{}
 
 	fn := func() {
+		log.Infof("Ping micro-services.co")
 		p.Timestamp = time.Now().Unix()
 		b, err := json.Marshal(p)
 		if err != nil {
@@ -255,7 +259,7 @@ func Setup(c *cli.Context) error {
 
 	client.DefaultClient = client.NewClient()
 
-	if !c.Bool("disable_ping") {
+	if c.Bool("enable_ping") {
 		go ping()
 	}
 
@@ -283,7 +287,11 @@ GLOBAL OPTIONS:
 	app := cli.NewApp()
 	app.HideVersion = true
 	app.Usage = "a go micro app"
-	app.Action = func(c *cli.Context) {}
+	app.Action = func(c *cli.Context) {
+		for _, action := range Actions {
+			action(c)
+		}
+	}
 	app.Before = Setup
 	app.Flags = Flags
 	app.RunAndExitOnError()

examples/pubsub/consumer/consumer.go

@@ -0,0 +1,52 @@
+package main
+
+import (
+	"fmt"
+
+	log "github.com/golang/glog"
+	"github.com/micro/go-micro/broker"
+	"github.com/micro/go-micro/cmd"
+
+	// To enable rabbitmq plugin uncomment
+	//_ "github.com/micro/go-plugins/broker/rabbitmq"
+)
+
+var (
+	topic = "go.micro.topic.foo"
+)
+
+// Example of a shared subscription which receives a subset of messages
+func sharedSub() {
+	_, err := broker.Subscribe(topic, func(p broker.Publication) error {
+		fmt.Println("[sub] received message:", string(p.Message().Body), "header", p.Message().Header)
+		return nil
+	}, broker.QueueName("consumer"))
+	if err != nil {
+		fmt.Println(err)
+	}
+}
+
+// Example of a subscription which receives all the messages
+func sub() {
+	_, err := broker.Subscribe(topic, func(p broker.Publication) error {
+		fmt.Println("[sub] received message:", string(p.Message().Body), "header", p.Message().Header)
+		return nil
+	})
+	if err != nil {
+		fmt.Println(err)
+	}
+}
+
+func main() {
+	cmd.Init()
+
+	if err := broker.Init(); err != nil {
+		log.Fatalf("Broker Init error: %v", err)
+	}
+	if err := broker.Connect(); err != nil {
+		log.Fatalf("Broker Connect error: %v", err)
+	}
+
+	sub()
+	select {}
+}

examples/pubsub/main.go

@@ -33,8 +33,9 @@ func pub() {
 }
 
 func sub() {
-	_, err := broker.Subscribe(topic, func(msg *broker.Message) {
-		fmt.Println("[sub] received message:", string(msg.Body), "header", msg.Header)
+	_, err := broker.Subscribe(topic, func(p broker.Publication) error {
+		fmt.Println("[sub] received message:", string(p.Message().Body), "header", p.Message().Header)
+		return nil
 	})
 	if err != nil {
 		fmt.Println(err)

examples/pubsub/producer/producer.go

@@ -0,0 +1,50 @@
+package main
+
+import (
+	"fmt"
+	"time"
+
+	log "github.com/golang/glog"
+	"github.com/micro/go-micro/broker"
+	"github.com/micro/go-micro/cmd"
+
+	// To enable rabbitmq plugin uncomment
+	//_ "github.com/micro/go-plugins/broker/rabbitmq"
+)
+
+var (
+	topic = "go.micro.topic.foo"
+)
+
+func pub() {
+	tick := time.NewTicker(time.Second)
+	i := 0
+	for _ = range tick.C {
+		msg := &broker.Message{
+			Header: map[string]string{
+				"id": fmt.Sprintf("%d", i),
+			},
+			Body: []byte(fmt.Sprintf("%d: %s", i, time.Now().String())),
+		}
+		if err := broker.Publish(topic, msg); err != nil {
+			log.Errorf("[pub] failed: %v", err)
+		} else {
+			fmt.Println("[pub] pubbed message:", string(msg.Body))
+		}
+		i++
+	}
+}
+
+func main() {
+	cmd.Init()
+
+	if err := broker.Init(); err != nil {
+		log.Fatalf("Broker Init error: %v", err)
+	}
+
+	if err := broker.Connect(); err != nil {
+		log.Fatalf("Broker Connect error: %v", err)
+	}
+
+	pub()
+}

server/extractor.go

@@ -20,7 +20,10 @@ func init() {
 	}
 }
 
-func extractValue(v reflect.Type) *registry.Value {
+func extractValue(v reflect.Type, d int) *registry.Value {
+	if d == 3 {
+		return nil
+	}
 	if v == nil {
 		return nil
 	}
@@ -37,7 +40,10 @@ func extractValue(v reflect.Type) *registry.Value {
 	switch v.Kind() {
 	case reflect.Struct:
 		for i := 0; i < v.NumField(); i++ {
-			val := extractValue(v.Field(i).Type)
+			val := extractValue(v.Field(i).Type, d+1)
+			if val == nil {
+				continue
+			}
 			val.Name = v.Field(i).Name
 			arg.Values = append(arg.Values, val)
 		}
@@ -47,7 +53,10 @@ func extractValue(v reflect.Type) *registry.Value {
 			p = p.Elem()
 		}
 		arg.Type = "[]" + p.Name()
-		arg.Values = append(arg.Values, extractValue(v.Elem()))
+		val := extractValue(v.Elem(), d+1)
+		if val != nil {
+			arg.Values = append(arg.Values, val)
+		}
 	}
 
 	return arg
@@ -77,8 +86,8 @@ func extractEndpoint(method reflect.Method) *registry.Endpoint {
 		stream = true
 	}
 
-	request := extractValue(reqType)
-	response := extractValue(rspType)
+	request := extractValue(reqType, 0)
+	response := extractValue(rspType, 0)
 
 	return &registry.Endpoint{
 		Name:     method.Name,
@@ -102,7 +111,7 @@ func extractSubValue(typ reflect.Type) *registry.Value {
 	default:
 		return nil
 	}
-	return extractValue(reqType)
+	return extractValue(reqType, 0)
 }
 
 func extractAddress(addr string) (string, error) {

server/options.go

@@ -24,7 +24,8 @@ type options struct {
 
 func newOptions(opt ...Option) options {
 	opts := options{
-		codecs: make(map[string]codec.NewCodec),
+		codecs:   make(map[string]codec.NewCodec),
+		metadata: map[string]string{},
 	}
 
 	for _, o := range opt {

server/rpc_server.go

@@ -2,6 +2,7 @@ package server
 
 import (
 	"fmt"
+	"runtime/debug"
 	"strconv"
 	"strings"
 	"sync"
@@ -43,6 +44,13 @@ func newRpcServer(opts ...Option) Server {
 }
 
 func (s *rpcServer) accept(sock transport.Socket) {
+	defer func() {
+		if r := recover(); r != nil {
+			log.Error(r, string(debug.Stack()))
+			sock.Close()
+		}
+	}()
+
 	var msg transport.Message
 	if err := sock.Recv(&msg); err != nil {
 		return

server/subscriber.go

@@ -155,11 +155,12 @@ func validateSubscriber(sub Subscriber) error {
 }
 
 func (s *rpcServer) createSubHandler(sb *subscriber, opts options) broker.Handler {
-	return func(msg *broker.Message) {
+	return func(p broker.Publication) error {
+		msg := p.Message()
 		ct := msg.Header["Content-Type"]
 		cf, err := s.newCodec(ct)
 		if err != nil {
-			return
+			return err
 		}
 
 		hdr := make(map[string]string)
@@ -190,11 +191,11 @@ func (s *rpcServer) createSubHandler(sb *subscriber, opts options) broker.Handle
 			defer co.Close()
 
 			if err := co.ReadHeader(&codec.Message{}, codec.Publication); err != nil {
-				continue
+				return err
 			}
 
 			if err := co.ReadBody(req.Interface()); err != nil {
-				continue
+				return err
 			}
 
 			fn := func(ctx context.Context, msg Publication) error {
@@ -225,6 +226,7 @@ func (s *rpcServer) createSubHandler(sb *subscriber, opts options) broker.Handle
 				message:     req.Interface(),
 			})
 		}
+		return nil
 	}
 }