Add pub/sub to client/server and make broker more low level

server/extractor.go

@@ -1,6 +1,7 @@
 package server
 
 import (
+	"fmt"
 	"reflect"
 
 	"github.com/myodc/go-micro/registry"
@@ -37,7 +38,7 @@ func extractEndpoint(method reflect.Method) *registry.Endpoint {
 	}
 
 	var rspType, reqType reflect.Type
-	//	var stream bool
+	var stream bool
 	mt := method.Type
 
 	switch mt.NumIn() {
@@ -51,9 +52,9 @@ func extractEndpoint(method reflect.Method) *registry.Endpoint {
 		return nil
 	}
 
-	//	if rspType.Kind() == reflect.Func {
-	//		stream = true
-	//	}
+	if rspType.Kind() == reflect.Func {
+		stream = true
+	}
 
 	request := extractValue(reqType)
 	response := extractValue(rspType)
@@ -62,5 +63,21 @@ func extractEndpoint(method reflect.Method) *registry.Endpoint {
 		Name:     method.Name,
 		Request:  request,
 		Response: response,
+		Metadata: map[string]string{
+			"stream": fmt.Sprintf("%v", stream),
+		},
 	}
 }
+
+func extractSubValue(typ reflect.Type) *registry.Value {
+	var reqType reflect.Type
+	switch typ.NumIn() {
+	case 1:
+		reqType = typ.In(0)
+	case 2:
+		reqType = typ.In(1)
+	default:
+		return nil
+	}
+	return extractValue(reqType)
+}

server/handler.go

@@ -9,3 +9,9 @@ type Handler interface {
 	Handler() interface{}
 	Endpoints() []*registry.Endpoint
 }
+
+type Subscriber interface {
+	Topic() string
+	Subscriber() interface{}
+	Endpoints() []*registry.Endpoint
+}

server/options.go

@@ -1,11 +1,13 @@
 package server
 
 import (
+	"github.com/myodc/go-micro/broker"
 	"github.com/myodc/go-micro/registry"
 	"github.com/myodc/go-micro/transport"
 )
 
 type options struct {
+	broker    broker.Broker
 	registry  registry.Registry
 	transport transport.Transport
 	metadata  map[string]string
@@ -22,6 +24,10 @@ func newOptions(opt ...Option) options {
 		o(&opts)
 	}
 
+	if opts.broker == nil {
+		opts.broker = broker.DefaultBroker
+	}
+
 	if opts.registry == nil {
 		opts.registry = registry.DefaultRegistry
 	}
@@ -93,6 +99,12 @@ func Address(a string) Option {
 	}
 }
 
+func Broker(b broker.Broker) Option {
+	return func(o *options) {
+		o.broker = b
+	}
+}
+
 func Registry(r registry.Registry) Option {
 	return func(o *options) {
 		o.registry = r

server/rpc_server.go

@@ -1,10 +1,12 @@
 package server
 
 import (
+	"fmt"
 	"strconv"
 	"strings"
 	"sync"
 
+	"github.com/myodc/go-micro/broker"
 	c "github.com/myodc/go-micro/context"
 	"github.com/myodc/go-micro/registry"
 	"github.com/myodc/go-micro/transport"
@@ -20,16 +22,18 @@ type rpcServer struct {
 	exit chan chan error
 
 	sync.RWMutex
-	opts     options
-	handlers map[string]Handler
+	opts        options
+	handlers    map[string]Handler
+	subscribers map[*subscriber][]broker.Subscriber
 }
 
 func newRpcServer(opts ...Option) Server {
 	return &rpcServer{
-		opts:     newOptions(opts...),
-		rpc:      rpc.NewServer(),
-		handlers: make(map[string]Handler),
-		exit:     make(chan chan error),
+		opts:        newOptions(opts...),
+		rpc:         rpc.NewServer(),
+		handlers:    make(map[string]Handler),
+		subscribers: make(map[*subscriber][]broker.Subscriber),
+		exit:        make(chan chan error),
 	}
 }
 
@@ -84,6 +88,29 @@ func (s *rpcServer) Handle(h Handler) error {
 	return nil
 }
 
+func (s *rpcServer) NewSubscriber(topic string, sb interface{}) Subscriber {
+	return newSubscriber(topic, sb)
+}
+
+func (s *rpcServer) Subscribe(sb Subscriber) error {
+	sub, ok := sb.(*subscriber)
+	if !ok {
+		return fmt.Errorf("invalid subscriber: expected *subscriber")
+	}
+	if len(sub.handlers) == 0 {
+		return fmt.Errorf("invalid subscriber: no handler functions")
+	}
+
+	s.Lock()
+	_, ok = s.subscribers[sub]
+	if ok {
+		return fmt.Errorf("subscriber %v already exists", s)
+	}
+	s.subscribers[sub] = nil
+	s.Unlock()
+	return nil
+}
+
 func (s *rpcServer) Register() error {
 	// parse address for host, port
 	config := s.Config()
@@ -110,6 +137,9 @@ func (s *rpcServer) Register() error {
 	for _, e := range s.handlers {
 		endpoints = append(endpoints, e.Endpoints()...)
 	}
+	for e, _ := range s.subscribers {
+		endpoints = append(endpoints, e.Endpoints()...)
+	}
 	s.RUnlock()
 
 	service := &registry.Service{
@@ -120,7 +150,23 @@ func (s *rpcServer) Register() error {
 	}
 
 	log.Infof("Registering node: %s", node.Id)
-	return config.registry.Register(service)
+	if err := config.registry.Register(service); err != nil {
+		return err
+	}
+
+	s.Lock()
+	defer s.Unlock()
+
+	for sb, _ := range s.subscribers {
+		handler := createSubHandler(sb)
+		sub, err := config.broker.Subscribe(sb.Topic(), handler)
+		if err != nil {
+			return err
+		}
+		s.subscribers[sb] = []broker.Subscriber{sub}
+	}
+
+	return nil
 }
 
 func (s *rpcServer) Deregister() error {
@@ -147,7 +193,21 @@ func (s *rpcServer) Deregister() error {
 		Nodes:   []*registry.Node{node},
 	}
 
-	return config.registry.Deregister(service)
+	log.Infof("Deregistering node: %s", node.Id)
+	if err := config.registry.Deregister(service); err != nil {
+		return err
+	}
+
+	s.Lock()
+	for sb, subs := range s.subscribers {
+		for _, sub := range subs {
+			log.Infof("Unsubscribing from topic: %s", sub.Topic())
+			sub.Unsubscribe()
+		}
+		s.subscribers[sb] = nil
+	}
+	s.Unlock()
+	return nil
 }
 
 func (s *rpcServer) Start() error {
@@ -169,9 +229,11 @@ func (s *rpcServer) Start() error {
 	go func() {
 		ch := <-s.exit
 		ch <- ts.Close()
+		config.broker.Disconnect()
 	}()
 
-	return nil
+	// TODO: subscribe to cruft
+	return config.broker.Connect()
 }
 
 func (s *rpcServer) Stop() error {

server/server.go

@@ -14,6 +14,8 @@ type Server interface {
 	Init(...Option)
 	Handle(Handler) error
 	NewHandler(interface{}) Handler
+	NewSubscriber(string, interface{}) Subscriber
+	Subscribe(Subscriber) error
 	Register() error
 	Deregister() error
 	Start() error
@@ -45,6 +47,10 @@ func NewServer(opt ...Option) Server {
 	return newRpcServer(opt...)
 }
 
+func NewSubscriber(topic string, h interface{}) Subscriber {
+	return DefaultServer.NewSubscriber(topic, h)
+}
+
 func NewHandler(h interface{}) Handler {
 	return DefaultServer.NewHandler(h)
 }
@@ -53,6 +59,10 @@ func Handle(h Handler) error {
 	return DefaultServer.Handle(h)
 }
 
+func Subscribe(s Subscriber) error {
+	return DefaultServer.Subscribe(s)
+}
+
 func Register() error {
 	return DefaultServer.Register()
 }
@@ -78,9 +88,6 @@ func Run() error {
 		return err
 	}
 
-	log.Infof("Deregistering %s", DefaultServer.Config().Id())
-	DefaultServer.Deregister()
-
 	return Stop()
 }
 

server/subscriber.go

@@ -0,0 +1,154 @@
+package server
+
+import (
+	"encoding/json"
+	"reflect"
+
+	"github.com/golang/protobuf/proto"
+	"github.com/myodc/go-micro/broker"
+	c "github.com/myodc/go-micro/context"
+	"github.com/myodc/go-micro/registry"
+	"golang.org/x/net/context"
+)
+
+type handler struct {
+	method  reflect.Value
+	reqType reflect.Type
+	ctxType reflect.Type
+}
+
+type subscriber struct {
+	topic      string
+	rcvr       reflect.Value
+	typ        reflect.Type
+	subscriber interface{}
+	handlers   []*handler
+	endpoints  []*registry.Endpoint
+}
+
+func newSubscriber(topic string, sub interface{}) Subscriber {
+	var endpoints []*registry.Endpoint
+	var handlers []*handler
+
+	if typ := reflect.TypeOf(sub); typ.Kind() == reflect.Func {
+		h := &handler{
+			method: reflect.ValueOf(sub),
+		}
+
+		switch typ.NumIn() {
+		case 1:
+			h.reqType = typ.In(0)
+		case 2:
+			h.ctxType = typ.In(0)
+			h.reqType = typ.In(1)
+		}
+
+		handlers = append(handlers, h)
+
+		endpoints = append(endpoints, &registry.Endpoint{
+			Name:    "Func",
+			Request: extractSubValue(typ),
+			Metadata: map[string]string{
+				"topic": topic,
+			},
+		})
+	} else {
+		for m := 0; m < typ.NumMethod(); m++ {
+			method := typ.Method(m)
+			h := &handler{
+				method: method.Func,
+			}
+
+			switch method.Type.NumIn() {
+			case 2:
+				h.reqType = method.Type.In(1)
+			case 3:
+				h.ctxType = method.Type.In(1)
+				h.reqType = method.Type.In(2)
+			}
+
+			handlers = append(handlers, h)
+
+			endpoints = append(endpoints, &registry.Endpoint{
+				Name:    method.Name,
+				Request: extractSubValue(method.Type),
+				Metadata: map[string]string{
+					"topic": topic,
+				},
+			})
+		}
+	}
+
+	return &subscriber{
+		rcvr:       reflect.ValueOf(sub),
+		typ:        reflect.TypeOf(sub),
+		topic:      topic,
+		subscriber: sub,
+		handlers:   handlers,
+		endpoints:  endpoints,
+	}
+}
+
+func createSubHandler(sb *subscriber) broker.Handler {
+	return func(msg *broker.Message) {
+		hdr := make(map[string]string)
+		for k, v := range msg.Header {
+			hdr[k] = v
+		}
+		delete(hdr, "Content-Type")
+		ctx := c.WithMetadata(context.Background(), hdr)
+		rctx := reflect.ValueOf(ctx)
+
+		for _, handler := range sb.handlers {
+			var isVal bool
+			var req reflect.Value
+			var uerr error
+
+			if handler.reqType.Kind() == reflect.Ptr {
+				req = reflect.New(handler.reqType.Elem())
+			} else {
+				req = reflect.New(handler.reqType)
+				isVal = true
+			}
+
+			switch msg.Header["Content-Type"] {
+			case "application/octet-stream":
+				uerr = proto.Unmarshal(msg.Body, req.Interface().(proto.Message))
+			case "application/json":
+				uerr = json.Unmarshal(msg.Body, req.Interface())
+			}
+
+			if uerr != nil {
+				continue
+			}
+
+			if isVal {
+				req = req.Elem()
+			}
+
+			var vals []reflect.Value
+			if sb.typ.Kind() != reflect.Func {
+				vals = append(vals, sb.rcvr)
+			}
+
+			if handler.ctxType != nil {
+				vals = append(vals, rctx)
+			}
+
+			vals = append(vals, req)
+			go handler.method.Call(vals)
+		}
+	}
+}
+
+func (s *subscriber) Topic() string {
+	return s.topic
+}
+
+func (s *subscriber) Subscriber() interface{} {
+	return s.subscriber
+}
+
+func (s *subscriber) Endpoints() []*registry.Endpoint {
+	return s.endpoints
+}