298 lines
6.8 KiB
Go
298 lines
6.8 KiB
Go
package tcp
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"fmt"
|
|
"reflect"
|
|
"strings"
|
|
"unicode"
|
|
"unicode/utf8"
|
|
|
|
"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/register"
|
|
"github.com/unistack-org/micro/v3/server"
|
|
)
|
|
|
|
const (
|
|
subSig = "func(context.Context, interface{}) error"
|
|
)
|
|
|
|
var typeOfError = reflect.TypeOf((*error)(nil)).Elem()
|
|
|
|
type handler struct {
|
|
reqType reflect.Type
|
|
ctxType reflect.Type
|
|
method reflect.Value
|
|
}
|
|
|
|
type tcpSubscriber struct {
|
|
topic string
|
|
rcvr reflect.Value
|
|
typ reflect.Type
|
|
subscriber interface{}
|
|
handlers []*handler
|
|
endpoints []*register.Endpoint
|
|
opts server.SubscriberOptions
|
|
}
|
|
|
|
// Is this an exported - upper case - name?
|
|
func isExported(name string) bool {
|
|
r, _ := utf8.DecodeRuneInString(name)
|
|
return unicode.IsUpper(r)
|
|
}
|
|
|
|
// Is this type exported or a builtin?
|
|
func isExportedOrBuiltinType(t reflect.Type) bool {
|
|
for t.Kind() == reflect.Ptr {
|
|
t = t.Elem()
|
|
}
|
|
// PkgPath will be non-empty even for an exported type,
|
|
// so we need to check the type name as well.
|
|
return isExported(t.Name()) || t.PkgPath() == ""
|
|
}
|
|
|
|
func newSubscriber(topic string, sub interface{}, opts ...server.SubscriberOption) server.Subscriber {
|
|
options := server.SubscriberOptions{
|
|
AutoAck: true,
|
|
}
|
|
|
|
for _, o := range opts {
|
|
o(&options)
|
|
}
|
|
|
|
var endpoints []*register.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 := ®ister.Endpoint{
|
|
Name: "Func",
|
|
Request: register.ExtractSubValue(typ),
|
|
Metadata: metadata.New(2),
|
|
}
|
|
ep.Metadata.Set("topic", topic)
|
|
ep.Metadata.Set("subscriber", "true")
|
|
endpoints = append(endpoints, ep)
|
|
} 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 := ®ister.Endpoint{
|
|
Name: name + "." + method.Name,
|
|
Request: register.ExtractSubValue(method.Type),
|
|
Metadata: metadata.New(2),
|
|
}
|
|
ep.Metadata.Set("topic", topic)
|
|
ep.Metadata.Set("subscriber", "true")
|
|
endpoints = append(endpoints, ep)
|
|
}
|
|
}
|
|
|
|
return &tcpSubscriber{
|
|
rcvr: reflect.ValueOf(sub),
|
|
typ: reflect.TypeOf(sub),
|
|
topic: topic,
|
|
subscriber: sub,
|
|
handlers: handlers,
|
|
endpoints: endpoints,
|
|
opts: options,
|
|
}
|
|
}
|
|
|
|
func validateSubscriber(sub server.Subscriber) error {
|
|
typ := reflect.TypeOf(sub.Subscriber())
|
|
var argType reflect.Type
|
|
|
|
switch typ.Kind() {
|
|
case 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 {
|
|
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())
|
|
}
|
|
default:
|
|
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 *tcpServer) createSubHandler(sb *tcpSubscriber, opts server.Options) broker.Handler {
|
|
return func(p broker.Event) error {
|
|
msg := p.Message()
|
|
ct := msg.Header["Content-Type"]
|
|
cf, err := s.newCodec(ct)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
hdr := make(map[string]string, len(msg.Header)-1)
|
|
for k, v := range msg.Header {
|
|
if k == "Content-Type" {
|
|
continue
|
|
}
|
|
hdr[k] = v
|
|
}
|
|
ctx := metadata.NewIncomingContext(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)
|
|
|
|
if err := cf.ReadHeader(buf, &codec.Message{}, codec.Event); err != nil {
|
|
return err
|
|
}
|
|
|
|
if err := cf.ReadBody(buf, req.Interface()); err != nil {
|
|
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() {
|
|
results <- fn(ctx, &tcpMessage{
|
|
topic: sb.topic,
|
|
contentType: ct,
|
|
payload: req.Interface(),
|
|
header: msg.Header,
|
|
body: msg.Body,
|
|
codec: cf,
|
|
})
|
|
}()
|
|
}
|
|
|
|
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
|
|
}
|
|
}
|
|
|
|
func (s *tcpSubscriber) Topic() string {
|
|
return s.topic
|
|
}
|
|
|
|
func (s *tcpSubscriber) Subscriber() interface{} {
|
|
return s.subscriber
|
|
}
|
|
|
|
func (s *tcpSubscriber) Endpoints() []*register.Endpoint {
|
|
return s.endpoints
|
|
}
|
|
|
|
func (s *tcpSubscriber) Options() server.SubscriberOptions {
|
|
return s.opts
|
|
}
|