micro-server-http/subscriber.go

214 lines
4.6 KiB
Go
Raw Normal View History

package http
import (
2018-12-19 10:33:23 +03:00
"bytes"
"context"
"fmt"
"reflect"
"strings"
"github.com/unistack-org/micro/v3/broker"
"github.com/unistack-org/micro/v3/codec"
"github.com/unistack-org/micro/v3/metadata"
"github.com/unistack-org/micro/v3/registry"
"github.com/unistack-org/micro/v3/server"
)
2018-12-19 10:33:23 +03:00
const (
subSig = "func(context.Context, interface{}) error"
)
var typeOfError = reflect.TypeOf((*error)(nil)).Elem()
type handler struct {
method reflect.Value
reqType reflect.Type
ctxType reflect.Type
}
2018-12-19 12:47:03 +03:00
type httpSubscriber struct {
2018-12-19 10:33:23 +03:00
topic string
rcvr reflect.Value
typ reflect.Type
subscriber interface{}
handlers []*handler
endpoints []*registry.Endpoint
opts server.SubscriberOptions
}
func newSubscriber(topic string, sub interface{}, opts ...server.SubscriberOption) server.Subscriber {
options := server.NewSubscriberOptions(opts...)
2018-12-19 10:33:23 +03:00
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)
ep := &registry.Endpoint{
Name: "Func",
Request: registry.ExtractSubValue(typ),
Metadata: metadata.New(2),
}
ep.Metadata.Set("topic", topic)
ep.Metadata.Set("subscriber", "true")
endpoints = append(endpoints, ep)
2018-12-19 10:33:23 +03:00
} else {
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)
ep := &registry.Endpoint{
Name: name + "." + method.Name,
Request: registry.ExtractSubValue(method.Type),
Metadata: metadata.New(2),
}
ep.Metadata.Set("topic", topic)
ep.Metadata.Set("subscriber", "true")
endpoints = append(endpoints, ep)
2018-12-19 10:33:23 +03:00
}
}
2018-12-19 12:47:03 +03:00
return &httpSubscriber{
2018-12-19 10:33:23 +03:00
rcvr: reflect.ValueOf(sub),
typ: reflect.TypeOf(sub),
topic: topic,
subscriber: sub,
handlers: handlers,
endpoints: endpoints,
opts: options,
}
}
2018-12-19 12:47:03 +03:00
func (s *httpServer) createSubHandler(sb *httpSubscriber, opts server.Options) broker.Handler {
2019-07-07 14:46:31 +03:00
return func(p broker.Event) error {
2018-12-19 10:33:23 +03:00
msg := p.Message()
ct := msg.Header["Content-Type"]
cf, err := s.newCodec(ct)
if err != nil {
return err
}
hdr := metadata.Copy(msg.Header)
2018-12-19 10:33:23 +03:00
delete(hdr, "Content-Type")
ctx := metadata.NewContext(context.Background(), hdr)
results := make(chan error, len(sb.handlers))
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
}
if isVal {
req = req.Elem()
}
buf := bytes.NewBuffer(msg.Body)
2018-12-19 10:33:23 +03:00
if err := cf.ReadHeader(buf, &codec.Message{}, codec.Event); err != nil {
2018-12-19 10:33:23 +03:00
return err
}
if err := cf.ReadBody(buf, req.Interface()); err != nil {
2018-12-19 10:33:23 +03:00
return err
}
fn := func(ctx context.Context, msg server.Message) error {
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))
}
vals = append(vals, reflect.ValueOf(msg.Payload()))
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)
}
go func() {
2018-12-19 12:47:03 +03:00
results <- fn(ctx, &httpMessage{
2018-12-19 10:33:23 +03:00
topic: sb.topic,
contentType: ct,
payload: req.Interface(),
header: msg.Header,
body: msg.Body,
codec: cf,
2018-12-19 10:33:23 +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"))
}
return nil
}
}
2018-12-19 12:47:03 +03:00
func (s *httpSubscriber) Topic() string {
2018-12-19 10:33:23 +03:00
return s.topic
}
2018-12-19 12:47:03 +03:00
func (s *httpSubscriber) Subscriber() interface{} {
2018-12-19 10:33:23 +03:00
return s.subscriber
}
2018-12-19 12:47:03 +03:00
func (s *httpSubscriber) Endpoints() []*registry.Endpoint {
2018-12-19 10:33:23 +03:00
return s.endpoints
}
2018-12-19 12:47:03 +03:00
func (s *httpSubscriber) Options() server.SubscriberOptions {
2018-12-19 10:33:23 +03:00
return s.opts
}