micro/server/subscriber.go

274 lines
6.0 KiB
Go
Raw Normal View History

package server
import (
"bytes"
2018-03-03 14:53:52 +03:00
"context"
"fmt"
"reflect"
2018-04-26 12:47:13 +03:00
"strings"
2015-11-20 19:17:33 +03:00
"github.com/micro/go-micro/broker"
"github.com/micro/go-micro/codec"
2016-01-28 20:55:28 +03:00
"github.com/micro/go-micro/metadata"
2015-11-20 19:17:33 +03:00
"github.com/micro/go-micro/registry"
)
const (
subSig = "func(context.Context, interface{}) error"
)
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
opts SubscriberOptions
}
func newSubscriber(topic string, sub interface{}, opts ...SubscriberOption) Subscriber {
var options SubscriberOptions
for _, o := range opts {
o(&options)
}
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{
2015-10-11 14:12:41 +03:00
"topic": topic,
"subscriber": "true",
},
})
} else {
2015-10-11 14:12:41 +03:00
hdlr := reflect.ValueOf(sub)
name := reflect.Indirect(hdlr).Type().Name()
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{
2015-10-11 14:12:41 +03:00
Name: name + "." + method.Name,
Request: extractSubValue(method.Type),
Metadata: map[string]string{
2015-10-11 14:12:41 +03:00
"topic": topic,
"subscriber": "true",
},
})
}
}
return &subscriber{
rcvr: reflect.ValueOf(sub),
typ: reflect.TypeOf(sub),
topic: topic,
subscriber: sub,
handlers: handlers,
endpoints: endpoints,
opts: options,
}
}
func validateSubscriber(sub Subscriber) error {
typ := reflect.TypeOf(sub.Subscriber())
var argType reflect.Type
if typ.Kind() == reflect.Func {
name := "Func"
switch typ.NumIn() {
case 2:
argType = typ.In(1)
default:
return fmt.Errorf("subscriber %v takes wrong number of args: %v required signature %s", name, typ.NumIn(), subSig)
}
if !isExportedOrBuiltinType(argType) {
return fmt.Errorf("subscriber %v argument type not exported: %v", name, argType)
}
if typ.NumOut() != 1 {
2016-04-06 20:03:27 +03:00
return fmt.Errorf("subscriber %v has wrong number of outs: %v require signature %s",
name, typ.NumOut(), subSig)
}
if returnType := typ.Out(0); returnType != typeOfError {
return fmt.Errorf("subscriber %v returns %v not error", name, returnType.String())
}
} else {
hdlr := reflect.ValueOf(sub.Subscriber())
name := reflect.Indirect(hdlr).Type().Name()
for m := 0; m < typ.NumMethod(); m++ {
method := typ.Method(m)
switch method.Type.NumIn() {
case 3:
argType = method.Type.In(2)
default:
return fmt.Errorf("subscriber %v.%v takes wrong number of args: %v required signature %s",
name, method.Name, method.Type.NumIn(), subSig)
}
if !isExportedOrBuiltinType(argType) {
return fmt.Errorf("%v argument type not exported: %v", name, argType)
}
if method.Type.NumOut() != 1 {
return fmt.Errorf(
"subscriber %v.%v has wrong number of outs: %v require signature %s",
name, method.Name, method.Type.NumOut(), subSig)
}
if returnType := method.Type.Out(0); returnType != typeOfError {
return fmt.Errorf("subscriber %v.%v returns %v not error", name, method.Name, returnType.String())
}
}
}
return nil
}
func (s *rpcServer) createSubHandler(sb *subscriber, opts Options) broker.Handler {
2015-12-23 22:16:55 +03:00
return func(p broker.Publication) error {
msg := p.Message()
2015-12-02 23:56:50 +03:00
ct := msg.Header["Content-Type"]
cf, err := s.newCodec(ct)
if err != nil {
2015-12-23 22:16:55 +03:00
return err
}
hdr := make(map[string]string)
for k, v := range msg.Header {
hdr[k] = v
}
delete(hdr, "Content-Type")
2016-01-28 20:55:28 +03:00
ctx := metadata.NewContext(context.Background(), hdr)
2018-04-26 12:47:13 +03:00
results := make(chan error, len(sb.handlers))
2015-12-02 22:56:57 +03:00
for i := 0; i < len(sb.handlers); i++ {
handler := sb.handlers[i]
var isVal bool
var req reflect.Value
if handler.reqType.Kind() == reflect.Ptr {
req = reflect.New(handler.reqType.Elem())
} else {
req = reflect.New(handler.reqType)
isVal = true
}
2015-12-02 22:56:57 +03:00
if isVal {
req = req.Elem()
}
2015-12-02 22:56:57 +03:00
b := &buffer{bytes.NewBuffer(msg.Body)}
co := cf(b)
defer co.Close()
if err := co.ReadHeader(&codec.Message{}, codec.Publication); err != nil {
2015-12-23 22:16:55 +03:00
return err
}
2015-12-02 22:56:57 +03:00
if err := co.ReadBody(req.Interface()); err != nil {
2015-12-23 22:16:55 +03:00
return err
}
2018-04-14 20:21:02 +03:00
fn := func(ctx context.Context, msg Message) error {
2015-12-02 22:56:57 +03:00
var vals []reflect.Value
if sb.typ.Kind() != reflect.Func {
vals = append(vals, sb.rcvr)
}
if handler.ctxType != nil {
vals = append(vals, reflect.ValueOf(ctx))
}
2018-04-14 20:21:02 +03:00
vals = append(vals, reflect.ValueOf(msg.Payload()))
2015-12-02 22:56:57 +03:00
returnValues := handler.method.Call(vals)
if err := returnValues[0].Interface(); err != nil {
return err.(error)
}
return nil
}
for i := len(opts.SubWrappers); i > 0; i-- {
fn = opts.SubWrappers[i-1](fn)
}
2017-05-31 21:47:41 +03:00
s.wg.Add(1)
go func() {
defer s.wg.Done()
2018-04-26 12:47:13 +03:00
results <- fn(ctx, &rpcMessage{
2017-05-31 21:47:41 +03:00
topic: sb.topic,
contentType: ct,
2018-04-14 20:21:02 +03:00
payload: req.Interface(),
2017-05-31 21:47:41 +03:00
})
}()
}
2018-04-26 12:47:13 +03:00
var errors []string
for i := 0; i < len(sb.handlers); i++ {
if err := <-results; err != nil {
errors = append(errors, err.Error())
}
}
if len(errors) > 0 {
return fmt.Errorf("subscriber error: %s", strings.Join(errors, "\n"))
}
2015-12-23 22:16:55 +03:00
return nil
}
}
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
}
func (s *subscriber) Options() SubscriberOptions {
return s.opts
}