2016-06-30 20:21:57 +01:00
|
|
|
// Package http implements a go-micro.Server
|
2021-10-26 22:55:51 +03:00
|
|
|
package http // import "go.unistack.org/micro-server-http/v3"
|
2016-06-30 20:21:57 +01:00
|
|
|
|
|
|
|
import (
|
2021-03-01 20:23:22 +03:00
|
|
|
"context"
|
2020-12-15 11:54:15 +03:00
|
|
|
"crypto/tls"
|
2022-01-24 20:29:52 +03:00
|
|
|
"errors"
|
2018-12-19 15:33:23 +08:00
|
|
|
"fmt"
|
2016-06-30 20:21:57 +01:00
|
|
|
"net"
|
|
|
|
"net/http"
|
2021-02-05 18:31:51 +03:00
|
|
|
"reflect"
|
2018-12-19 15:33:23 +08:00
|
|
|
"sort"
|
2021-02-05 18:31:51 +03:00
|
|
|
"strings"
|
2016-06-30 20:21:57 +01:00
|
|
|
"sync"
|
2019-02-01 17:11:25 +08:00
|
|
|
"time"
|
2016-06-30 20:21:57 +01:00
|
|
|
|
2021-10-26 22:36:04 +03:00
|
|
|
"go.unistack.org/micro/v3/broker"
|
|
|
|
"go.unistack.org/micro/v3/codec"
|
|
|
|
"go.unistack.org/micro/v3/logger"
|
|
|
|
"go.unistack.org/micro/v3/register"
|
|
|
|
"go.unistack.org/micro/v3/server"
|
|
|
|
rhttp "go.unistack.org/micro/v3/util/http"
|
2020-12-15 11:54:15 +03:00
|
|
|
"golang.org/x/net/netutil"
|
2016-06-30 20:21:57 +01:00
|
|
|
)
|
|
|
|
|
2023-05-09 18:34:24 +03:00
|
|
|
var _ server.Server = (*Server)(nil)
|
2022-01-23 02:00:15 +03:00
|
|
|
|
2023-05-09 18:34:24 +03:00
|
|
|
type Server struct {
|
2022-01-22 01:10:24 +03:00
|
|
|
hd server.Handler
|
|
|
|
rsvc *register.Service
|
|
|
|
handlers map[string]server.Handler
|
|
|
|
exit chan chan error
|
|
|
|
subscribers map[*httpSubscriber][]broker.Subscriber
|
|
|
|
errorHandler func(context.Context, server.Handler, http.ResponseWriter, *http.Request, error, int)
|
|
|
|
pathHandlers *rhttp.Trie
|
|
|
|
opts server.Options
|
|
|
|
registerRPC bool
|
2021-04-25 12:02:18 +03:00
|
|
|
sync.RWMutex
|
|
|
|
registered bool
|
|
|
|
init bool
|
2016-06-30 20:21:57 +01:00
|
|
|
}
|
|
|
|
|
2023-05-09 18:34:24 +03:00
|
|
|
func (h *Server) newCodec(ct string) (codec.Codec, error) {
|
2021-04-16 17:14:27 +03:00
|
|
|
if idx := strings.IndexRune(ct, ';'); idx >= 0 {
|
|
|
|
ct = ct[:idx]
|
|
|
|
}
|
2021-07-06 19:10:54 +03:00
|
|
|
h.RLock()
|
|
|
|
cf, ok := h.opts.Codecs[ct]
|
|
|
|
h.RUnlock()
|
|
|
|
if ok {
|
2018-12-19 15:33:23 +08:00
|
|
|
return cf, nil
|
|
|
|
}
|
2020-11-25 23:44:32 +03:00
|
|
|
return nil, codec.ErrUnknownContentType
|
2018-12-19 15:33:23 +08:00
|
|
|
}
|
|
|
|
|
2023-05-09 18:34:24 +03:00
|
|
|
func (h *Server) Options() server.Options {
|
2016-06-30 20:21:57 +01:00
|
|
|
h.Lock()
|
|
|
|
opts := h.opts
|
|
|
|
h.Unlock()
|
|
|
|
return opts
|
|
|
|
}
|
|
|
|
|
2023-05-09 18:34:24 +03:00
|
|
|
func (h *Server) Init(opts ...server.Option) error {
|
2021-03-24 15:22:20 +03:00
|
|
|
if len(opts) == 0 && h.init {
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2016-06-30 20:21:57 +01:00
|
|
|
h.Lock()
|
2021-04-13 11:32:56 +03:00
|
|
|
|
2016-06-30 20:21:57 +01:00
|
|
|
for _, o := range opts {
|
|
|
|
o(&h.opts)
|
|
|
|
}
|
2021-03-09 14:19:59 +03:00
|
|
|
if fn, ok := h.opts.Context.Value(errorHandlerKey{}).(func(ctx context.Context, s server.Handler, w http.ResponseWriter, r *http.Request, err error, status int)); ok && fn != nil {
|
|
|
|
h.errorHandler = fn
|
|
|
|
}
|
2021-03-24 15:22:20 +03:00
|
|
|
if h.handlers == nil {
|
|
|
|
h.handlers = make(map[string]server.Handler)
|
|
|
|
}
|
2021-04-13 11:32:56 +03:00
|
|
|
if h.pathHandlers == nil {
|
2021-09-30 03:00:24 +03:00
|
|
|
h.pathHandlers = rhttp.NewTrie()
|
2021-04-13 11:32:56 +03:00
|
|
|
}
|
2021-07-14 13:43:45 +03:00
|
|
|
|
|
|
|
if v, ok := h.opts.Context.Value(registerRPCHandlerKey{}).(bool); ok {
|
|
|
|
h.registerRPC = v
|
|
|
|
}
|
|
|
|
|
2021-04-13 11:32:56 +03:00
|
|
|
if phs, ok := h.opts.Context.Value(pathHandlerKey{}).(*pathHandlerVal); ok && phs.h != nil {
|
2022-01-22 01:10:24 +03:00
|
|
|
for pm, ps := range phs.h {
|
|
|
|
for pp, ph := range ps {
|
|
|
|
if err := h.pathHandlers.Insert([]string{pm}, pp, ph); err != nil {
|
|
|
|
h.Unlock()
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
}
|
2021-04-13 23:42:33 +03:00
|
|
|
}
|
|
|
|
}
|
2021-07-06 19:10:54 +03:00
|
|
|
h.Unlock()
|
2021-04-13 23:42:33 +03:00
|
|
|
|
2021-07-06 19:10:54 +03:00
|
|
|
h.RLock()
|
2021-03-24 15:22:20 +03:00
|
|
|
if err := h.opts.Register.Init(); err != nil {
|
2021-07-06 19:10:54 +03:00
|
|
|
h.RUnlock()
|
2021-03-24 15:22:20 +03:00
|
|
|
return err
|
|
|
|
}
|
|
|
|
if err := h.opts.Broker.Init(); err != nil {
|
2021-07-06 19:10:54 +03:00
|
|
|
h.RUnlock()
|
2021-03-24 15:22:20 +03:00
|
|
|
return err
|
|
|
|
}
|
|
|
|
if err := h.opts.Tracer.Init(); err != nil {
|
2021-07-06 19:10:54 +03:00
|
|
|
h.RUnlock()
|
2021-03-24 15:22:20 +03:00
|
|
|
return err
|
|
|
|
}
|
|
|
|
if err := h.opts.Logger.Init(); err != nil {
|
2021-07-06 19:10:54 +03:00
|
|
|
h.RUnlock()
|
2021-03-24 15:22:20 +03:00
|
|
|
return err
|
|
|
|
}
|
|
|
|
if err := h.opts.Meter.Init(); err != nil {
|
2021-07-06 19:10:54 +03:00
|
|
|
h.RUnlock()
|
2021-03-24 15:22:20 +03:00
|
|
|
return err
|
|
|
|
}
|
|
|
|
if err := h.opts.Transport.Init(); err != nil {
|
2021-07-06 19:10:54 +03:00
|
|
|
h.RUnlock()
|
2021-03-24 15:22:20 +03:00
|
|
|
return err
|
|
|
|
}
|
2021-07-06 19:10:54 +03:00
|
|
|
h.RUnlock()
|
2021-03-24 15:22:20 +03:00
|
|
|
|
2021-07-06 19:10:54 +03:00
|
|
|
h.Lock()
|
2021-03-24 15:22:20 +03:00
|
|
|
h.init = true
|
2021-07-06 19:10:54 +03:00
|
|
|
h.Unlock()
|
2021-03-24 15:22:20 +03:00
|
|
|
|
2016-06-30 20:21:57 +01:00
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2023-05-09 18:34:24 +03:00
|
|
|
func (h *Server) Handle(handler server.Handler) error {
|
2022-01-22 01:10:24 +03:00
|
|
|
// passed unknown handler
|
2021-04-26 00:43:06 +03:00
|
|
|
hdlr, ok := handler.(*httpHandler)
|
|
|
|
if !ok {
|
2021-07-06 19:10:54 +03:00
|
|
|
h.Lock()
|
2021-03-09 14:19:59 +03:00
|
|
|
h.hd = handler
|
2021-07-06 19:10:54 +03:00
|
|
|
h.Unlock()
|
2021-04-26 00:43:06 +03:00
|
|
|
return nil
|
2021-03-09 14:19:59 +03:00
|
|
|
}
|
2021-04-26 00:43:06 +03:00
|
|
|
|
2022-01-22 01:10:24 +03:00
|
|
|
// passed http.Handler like some muxer
|
2021-04-26 00:43:06 +03:00
|
|
|
if _, ok := hdlr.hd.(http.Handler); ok {
|
2021-07-06 19:10:54 +03:00
|
|
|
h.Lock()
|
2021-04-26 00:43:06 +03:00
|
|
|
h.hd = handler
|
2021-07-06 19:10:54 +03:00
|
|
|
h.Unlock()
|
2021-04-26 00:43:06 +03:00
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2022-01-22 01:10:24 +03:00
|
|
|
// passed micro compat handler
|
2021-07-06 19:10:54 +03:00
|
|
|
h.Lock()
|
2021-04-26 00:43:06 +03:00
|
|
|
if h.handlers == nil {
|
|
|
|
h.handlers = make(map[string]server.Handler)
|
|
|
|
}
|
|
|
|
h.handlers[handler.Name()] = handler
|
2021-07-06 19:10:54 +03:00
|
|
|
h.Unlock()
|
2021-04-26 00:43:06 +03:00
|
|
|
|
2016-06-30 20:21:57 +01:00
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2023-05-09 18:34:24 +03:00
|
|
|
func (h *Server) NewHandler(handler interface{}, opts ...server.HandlerOption) server.Handler {
|
2020-11-18 13:08:50 +03:00
|
|
|
options := server.NewHandlerOptions(opts...)
|
2016-06-30 20:21:57 +01:00
|
|
|
|
2021-04-20 12:53:29 +03:00
|
|
|
eps := make([]*register.Endpoint, 0, len(options.Metadata))
|
|
|
|
for name, metadata := range options.Metadata {
|
|
|
|
eps = append(eps, ®ister.Endpoint{
|
|
|
|
Name: name,
|
|
|
|
Metadata: metadata,
|
|
|
|
})
|
2017-04-03 15:03:46 +01:00
|
|
|
}
|
|
|
|
|
2021-02-05 18:31:51 +03:00
|
|
|
hdlr := &httpHandler{
|
2022-01-22 01:10:24 +03:00
|
|
|
eps: eps,
|
|
|
|
hd: handler,
|
|
|
|
opts: options,
|
|
|
|
sopts: h.opts,
|
|
|
|
handlers: rhttp.NewTrie(),
|
2021-02-05 18:31:51 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
tp := reflect.TypeOf(handler)
|
|
|
|
|
2022-03-21 15:41:40 +03:00
|
|
|
/*
|
|
|
|
if len(options.Metadata) == 0 {
|
|
|
|
if h.registerRPC {
|
|
|
|
h.opts.Logger.Infof(h.opts.Context, "register rpc handler for http.MethodPost %s /%s", hn, hn)
|
|
|
|
if err := hdlr.handlers.Insert([]string{http.MethodPost}, "/"+hn, pth); err != nil {
|
|
|
|
h.opts.Logger.Errorf(h.opts.Context, "cant add rpc handler for http.MethodPost %s /%s", hn, hn)
|
|
|
|
}
|
2022-03-05 19:10:02 +03:00
|
|
|
}
|
|
|
|
}
|
2022-03-21 15:41:40 +03:00
|
|
|
*/
|
|
|
|
|
2024-03-20 12:27:14 +03:00
|
|
|
registerCORS := false
|
|
|
|
if v, ok := options.Context.Value(registerCORSHandlerKey{}).(bool); ok && v {
|
|
|
|
registerCORS = true
|
|
|
|
}
|
|
|
|
|
2021-02-05 18:31:51 +03:00
|
|
|
for hn, md := range options.Metadata {
|
|
|
|
var method reflect.Method
|
|
|
|
mname := hn[strings.Index(hn, ".")+1:]
|
|
|
|
for m := 0; m < tp.NumMethod(); m++ {
|
|
|
|
mn := tp.Method(m)
|
|
|
|
if mn.Name != mname {
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
method = mn
|
|
|
|
break
|
|
|
|
}
|
|
|
|
|
|
|
|
if method.Name == "" && h.opts.Logger.V(logger.ErrorLevel) {
|
2024-04-24 18:25:31 +03:00
|
|
|
h.opts.Logger.Error(h.opts.Context, fmt.Sprintf("nil method for %s", mname))
|
2021-02-05 18:31:51 +03:00
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
|
|
|
mtype, err := prepareEndpoint(method)
|
|
|
|
if err != nil && h.opts.Logger.V(logger.ErrorLevel) {
|
|
|
|
h.opts.Logger.Errorf(h.opts.Context, "%v", err)
|
|
|
|
continue
|
|
|
|
} else if mtype == nil {
|
2024-04-24 18:25:31 +03:00
|
|
|
h.opts.Logger.Error(h.opts.Context, fmt.Sprintf("nil mtype for %s", mname))
|
2021-02-05 18:31:51 +03:00
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
|
|
|
rcvr := reflect.ValueOf(handler)
|
|
|
|
name := reflect.Indirect(rcvr).Type().Name()
|
|
|
|
|
2022-01-22 01:10:24 +03:00
|
|
|
pth := &patHandler{mtype: mtype, name: name, rcvr: rcvr}
|
2021-02-05 18:31:51 +03:00
|
|
|
hdlr.name = name
|
2021-07-14 13:43:45 +03:00
|
|
|
|
2024-03-18 16:25:52 +03:00
|
|
|
methods := []string{md["Method"]}
|
2024-03-20 12:27:14 +03:00
|
|
|
if registerCORS {
|
2024-03-18 16:25:52 +03:00
|
|
|
methods = append(methods, http.MethodOptions)
|
|
|
|
}
|
2024-03-18 16:38:18 +03:00
|
|
|
|
2024-03-18 16:25:52 +03:00
|
|
|
if err := hdlr.handlers.Insert(methods, md["Path"], pth); err != nil {
|
2024-04-24 18:25:31 +03:00
|
|
|
h.opts.Logger.Error(h.opts.Context, fmt.Sprintf("cant add handler for %v %s", methods, md["Path"]))
|
2024-03-18 14:51:40 +03:00
|
|
|
}
|
|
|
|
|
2022-01-22 01:10:24 +03:00
|
|
|
if h.registerRPC {
|
2024-03-18 16:25:52 +03:00
|
|
|
methods := []string{http.MethodPost}
|
2024-03-20 12:27:14 +03:00
|
|
|
if registerCORS {
|
2024-03-18 16:25:52 +03:00
|
|
|
methods = append(methods, http.MethodOptions)
|
2023-02-21 23:56:07 +03:00
|
|
|
}
|
2024-03-18 15:45:43 +03:00
|
|
|
|
2024-03-18 16:25:52 +03:00
|
|
|
if err := hdlr.handlers.Insert(methods, "/"+hn, pth); err != nil {
|
2024-04-24 18:25:31 +03:00
|
|
|
h.opts.Logger.Error(h.opts.Context, fmt.Sprintf("cant add rpc handler for http.MethodPost %s /%s", hn, hn))
|
2024-03-18 15:45:43 +03:00
|
|
|
}
|
2023-02-21 23:56:07 +03:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
metadata, ok := options.Context.Value(handlerEndpointsKey{}).([]EndpointMetadata)
|
|
|
|
if !ok {
|
|
|
|
return hdlr
|
|
|
|
}
|
|
|
|
|
|
|
|
for _, md := range metadata {
|
|
|
|
hn := md.Name
|
|
|
|
var method reflect.Method
|
|
|
|
mname := hn[strings.Index(hn, ".")+1:]
|
|
|
|
for m := 0; m < tp.NumMethod(); m++ {
|
|
|
|
mn := tp.Method(m)
|
|
|
|
if mn.Name != mname {
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
method = mn
|
|
|
|
break
|
|
|
|
}
|
|
|
|
|
|
|
|
if method.Name == "" && h.opts.Logger.V(logger.ErrorLevel) {
|
2024-04-24 18:25:31 +03:00
|
|
|
h.opts.Logger.Error(h.opts.Context, fmt.Sprintf("nil method for %s", mname))
|
2023-02-21 23:56:07 +03:00
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
|
|
|
mtype, err := prepareEndpoint(method)
|
|
|
|
if err != nil && h.opts.Logger.V(logger.ErrorLevel) {
|
|
|
|
h.opts.Logger.Errorf(h.opts.Context, "%v", err)
|
|
|
|
continue
|
|
|
|
} else if mtype == nil {
|
2024-04-24 18:25:31 +03:00
|
|
|
h.opts.Logger.Error(h.opts.Context, fmt.Sprintf("nil mtype for %s", mname))
|
2023-02-21 23:56:07 +03:00
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
|
|
|
rcvr := reflect.ValueOf(handler)
|
|
|
|
name := reflect.Indirect(rcvr).Type().Name()
|
|
|
|
|
|
|
|
pth := &patHandler{mtype: mtype, name: name, rcvr: rcvr}
|
|
|
|
hdlr.name = name
|
|
|
|
|
2024-03-20 12:27:14 +03:00
|
|
|
methods := []string{md.Method}
|
|
|
|
if registerCORS {
|
|
|
|
methods = append(methods, http.MethodOptions)
|
|
|
|
}
|
|
|
|
|
|
|
|
if err := hdlr.handlers.Insert(methods, md.Path, pth); err != nil {
|
2024-04-24 18:25:31 +03:00
|
|
|
h.opts.Logger.Error(h.opts.Context, fmt.Sprintf("cant add handler for %s %s", md.Method, md.Path))
|
2023-02-21 23:56:07 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
if h.registerRPC {
|
2024-03-20 12:27:14 +03:00
|
|
|
methods := []string{http.MethodPost}
|
|
|
|
if registerCORS {
|
|
|
|
methods = append(methods, http.MethodOptions)
|
|
|
|
}
|
|
|
|
|
2024-04-24 18:25:31 +03:00
|
|
|
h.opts.Logger.Info(h.opts.Context, fmt.Sprintf("register rpc handler for http.MethodPost %s /%s", hn, hn))
|
2024-03-20 12:27:14 +03:00
|
|
|
if err := hdlr.handlers.Insert(methods, "/"+hn, pth); err != nil {
|
2024-04-24 18:25:31 +03:00
|
|
|
h.opts.Logger.Error(h.opts.Context, fmt.Sprintf("cant add rpc handler for http.MethodPost %s /%s", hn, hn))
|
2022-01-22 01:10:24 +03:00
|
|
|
}
|
|
|
|
}
|
2016-06-30 20:21:57 +01:00
|
|
|
}
|
2021-02-05 18:31:51 +03:00
|
|
|
|
|
|
|
return hdlr
|
2016-06-30 20:21:57 +01:00
|
|
|
}
|
|
|
|
|
2023-05-09 18:34:24 +03:00
|
|
|
func (h *Server) NewSubscriber(topic string, handler interface{}, opts ...server.SubscriberOption) server.Subscriber {
|
2018-12-19 15:33:23 +08:00
|
|
|
return newSubscriber(topic, handler, opts...)
|
|
|
|
}
|
|
|
|
|
2023-05-09 18:34:24 +03:00
|
|
|
func (h *Server) Subscribe(sb server.Subscriber) error {
|
2018-12-19 17:47:03 +08:00
|
|
|
sub, ok := sb.(*httpSubscriber)
|
2018-12-19 15:33:23 +08:00
|
|
|
if !ok {
|
2018-12-19 17:47:03 +08:00
|
|
|
return fmt.Errorf("invalid subscriber: expected *httpSubscriber")
|
2018-12-19 15:33:23 +08:00
|
|
|
}
|
|
|
|
if len(sub.handlers) == 0 {
|
|
|
|
return fmt.Errorf("invalid subscriber: no handler functions")
|
2016-06-30 20:21:57 +01:00
|
|
|
}
|
|
|
|
|
2020-10-10 00:38:35 +03:00
|
|
|
if err := server.ValidateSubscriber(sb); err != nil {
|
2018-12-19 15:33:23 +08:00
|
|
|
return err
|
2016-06-30 20:21:57 +01:00
|
|
|
}
|
|
|
|
|
2021-07-06 19:10:54 +03:00
|
|
|
h.RLock()
|
2018-12-19 15:33:23 +08:00
|
|
|
_, ok = h.subscribers[sub]
|
2021-07-06 19:10:54 +03:00
|
|
|
h.RUnlock()
|
2018-12-19 15:33:23 +08:00
|
|
|
if ok {
|
|
|
|
return fmt.Errorf("subscriber %v already exists", h)
|
|
|
|
}
|
2021-07-06 19:10:54 +03:00
|
|
|
h.Lock()
|
2018-12-19 15:33:23 +08:00
|
|
|
h.subscribers[sub] = nil
|
2021-07-06 19:10:54 +03:00
|
|
|
h.Unlock()
|
2018-12-19 15:33:23 +08:00
|
|
|
return nil
|
2016-06-30 20:21:57 +01:00
|
|
|
}
|
|
|
|
|
2023-05-09 18:34:24 +03:00
|
|
|
func (h *Server) Register() error {
|
2021-03-09 14:19:59 +03:00
|
|
|
var eps []*register.Endpoint
|
2020-11-05 01:55:38 +03:00
|
|
|
h.RLock()
|
2021-03-09 14:19:59 +03:00
|
|
|
for _, hdlr := range h.handlers {
|
|
|
|
eps = append(eps, hdlr.Endpoints()...)
|
|
|
|
}
|
2020-11-05 01:55:38 +03:00
|
|
|
rsvc := h.rsvc
|
|
|
|
config := h.opts
|
|
|
|
h.RUnlock()
|
|
|
|
|
|
|
|
// if service already filled, reuse it and return early
|
|
|
|
if rsvc != nil {
|
|
|
|
if err := server.DefaultRegisterFunc(rsvc, config); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
return nil
|
|
|
|
}
|
2016-06-30 20:21:57 +01:00
|
|
|
|
2021-01-29 14:32:32 +03:00
|
|
|
service, err := server.NewRegisterService(h)
|
2020-10-10 00:38:35 +03:00
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
2020-10-16 16:20:19 +03:00
|
|
|
service.Nodes[0].Metadata["protocol"] = "http"
|
2017-04-03 15:03:46 +01:00
|
|
|
service.Endpoints = eps
|
2016-06-30 20:21:57 +01:00
|
|
|
|
2018-12-19 15:33:23 +08:00
|
|
|
h.Lock()
|
2021-04-26 00:43:06 +03:00
|
|
|
subscriberList := make([]*httpSubscriber, 0, len(h.subscribers))
|
2018-12-19 15:33:23 +08:00
|
|
|
for e := range h.subscribers {
|
|
|
|
// Only advertise non internal subscribers
|
2021-04-20 12:53:29 +03:00
|
|
|
subscriberList = append(subscriberList, e)
|
2018-12-19 15:33:23 +08:00
|
|
|
}
|
|
|
|
sort.Slice(subscriberList, func(i, j int) bool {
|
|
|
|
return subscriberList[i].topic > subscriberList[j].topic
|
|
|
|
})
|
2020-11-05 01:55:38 +03:00
|
|
|
|
2018-12-19 15:33:23 +08:00
|
|
|
for _, e := range subscriberList {
|
|
|
|
service.Endpoints = append(service.Endpoints, e.Endpoints()...)
|
|
|
|
}
|
|
|
|
h.Unlock()
|
|
|
|
|
2020-11-05 01:55:38 +03:00
|
|
|
h.RLock()
|
|
|
|
registered := h.registered
|
|
|
|
h.RUnlock()
|
2016-06-30 20:21:57 +01:00
|
|
|
|
2020-11-05 01:55:38 +03:00
|
|
|
if !registered {
|
|
|
|
if config.Logger.V(logger.InfoLevel) {
|
2024-04-24 18:25:31 +03:00
|
|
|
config.Logger.Info(config.Context, fmt.Sprintf("Register [%s] Registering node: %s", config.Register.String(), service.Nodes[0].ID))
|
2020-11-05 01:55:38 +03:00
|
|
|
}
|
|
|
|
}
|
2018-08-09 18:31:22 +08:00
|
|
|
|
2020-11-05 01:55:38 +03:00
|
|
|
// register the service
|
|
|
|
if err := server.DefaultRegisterFunc(service, config); err != nil {
|
2018-12-19 18:22:27 +08:00
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
2020-11-05 01:55:38 +03:00
|
|
|
// already registered? don't need to register subscribers
|
|
|
|
if registered {
|
2018-12-19 18:22:27 +08:00
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2020-11-05 01:55:38 +03:00
|
|
|
h.Lock()
|
2023-05-13 16:09:21 +03:00
|
|
|
|
|
|
|
h.registered = true
|
|
|
|
h.rsvc = service
|
|
|
|
h.Unlock()
|
|
|
|
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func (h *Server) subscribe() error {
|
|
|
|
config := h.opts
|
|
|
|
|
2020-10-10 00:38:35 +03:00
|
|
|
for sb := range h.subscribers {
|
2020-11-05 01:55:38 +03:00
|
|
|
handler := h.createSubHandler(sb, config)
|
|
|
|
var opts []broker.SubscribeOption
|
2018-12-19 18:22:27 +08:00
|
|
|
if queue := sb.Options().Queue; len(queue) > 0 {
|
2020-11-05 01:55:38 +03:00
|
|
|
opts = append(opts, broker.SubscribeGroup(queue))
|
2018-12-19 18:22:27 +08:00
|
|
|
}
|
2020-10-10 00:38:35 +03:00
|
|
|
|
2020-11-05 01:55:38 +03:00
|
|
|
subCtx := config.Context
|
2020-10-16 16:20:19 +03:00
|
|
|
if cx := sb.Options().Context; cx != nil {
|
|
|
|
subCtx = cx
|
|
|
|
}
|
2020-11-05 01:55:38 +03:00
|
|
|
opts = append(opts, broker.SubscribeContext(subCtx))
|
|
|
|
opts = append(opts, broker.SubscribeAutoAck(sb.Options().AutoAck))
|
2022-06-27 00:41:18 +03:00
|
|
|
opts = append(opts, broker.SubscribeBodyOnly(sb.Options().BodyOnly))
|
2020-10-16 16:20:19 +03:00
|
|
|
|
2020-11-05 01:55:38 +03:00
|
|
|
sub, err := config.Broker.Subscribe(subCtx, sb.Topic(), handler, opts...)
|
2018-12-19 18:22:27 +08:00
|
|
|
if err != nil {
|
2021-07-06 19:10:54 +03:00
|
|
|
h.Unlock()
|
2018-12-19 18:22:27 +08:00
|
|
|
return err
|
|
|
|
}
|
|
|
|
h.subscribers[sb] = []broker.Subscriber{sub}
|
|
|
|
}
|
2020-11-05 01:55:38 +03:00
|
|
|
|
2018-12-19 18:22:27 +08:00
|
|
|
return nil
|
2016-06-30 20:21:57 +01:00
|
|
|
}
|
|
|
|
|
2023-05-09 18:34:24 +03:00
|
|
|
func (h *Server) Deregister() error {
|
2020-11-05 01:55:38 +03:00
|
|
|
h.RLock()
|
|
|
|
config := h.opts
|
|
|
|
h.RUnlock()
|
2020-10-10 00:38:35 +03:00
|
|
|
|
2021-01-29 14:32:32 +03:00
|
|
|
service, err := server.NewRegisterService(h)
|
2020-10-10 00:38:35 +03:00
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
2018-08-09 18:31:22 +08:00
|
|
|
|
2020-11-05 01:55:38 +03:00
|
|
|
if config.Logger.V(logger.InfoLevel) {
|
2024-04-24 18:25:31 +03:00
|
|
|
config.Logger.Info(config.Context, "Deregistering node: "+service.Nodes[0].ID)
|
2020-11-05 01:55:38 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
if err := server.DefaultDeregisterFunc(service, config); err != nil {
|
2018-12-19 17:56:52 +08:00
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
h.Lock()
|
2020-11-05 01:55:38 +03:00
|
|
|
h.rsvc = nil
|
|
|
|
|
2018-12-19 18:22:27 +08:00
|
|
|
if !h.registered {
|
|
|
|
h.Unlock()
|
|
|
|
return nil
|
|
|
|
}
|
2020-11-05 01:55:38 +03:00
|
|
|
|
2018-12-19 18:22:27 +08:00
|
|
|
h.registered = false
|
|
|
|
|
2020-10-16 16:20:19 +03:00
|
|
|
subCtx := h.opts.Context
|
2018-12-19 17:56:52 +08:00
|
|
|
for sb, subs := range h.subscribers {
|
2020-10-16 16:46:39 +03:00
|
|
|
if cx := sb.Options().Context; cx != nil {
|
|
|
|
subCtx = cx
|
|
|
|
}
|
|
|
|
|
2018-12-19 17:56:52 +08:00
|
|
|
for _, sub := range subs {
|
2024-04-24 18:25:31 +03:00
|
|
|
config.Logger.Info(config.Context, "Unsubscribing from topic: "+sub.Topic())
|
2020-10-16 16:20:19 +03:00
|
|
|
if err := sub.Unsubscribe(subCtx); err != nil {
|
2021-07-06 19:10:54 +03:00
|
|
|
h.Unlock()
|
2024-04-24 18:25:31 +03:00
|
|
|
config.Logger.Error(config.Context, fmt.Sprintf("failed to unsubscribe topic: %s, error", sb.Topic()), err)
|
2020-10-16 16:20:19 +03:00
|
|
|
return err
|
|
|
|
}
|
2018-12-19 17:56:52 +08:00
|
|
|
}
|
|
|
|
h.subscribers[sb] = nil
|
|
|
|
}
|
|
|
|
h.Unlock()
|
|
|
|
return nil
|
2016-06-30 20:21:57 +01:00
|
|
|
}
|
|
|
|
|
2023-05-09 18:34:24 +03:00
|
|
|
func (h *Server) Start() error {
|
2020-11-05 01:55:38 +03:00
|
|
|
h.RLock()
|
|
|
|
config := h.opts
|
|
|
|
h.RUnlock()
|
2016-06-30 20:21:57 +01:00
|
|
|
|
2020-12-15 11:54:15 +03:00
|
|
|
// micro: config.Transport.Listen(config.Address)
|
|
|
|
var ts net.Listener
|
|
|
|
|
|
|
|
if l := config.Listener; l != nil {
|
|
|
|
ts = l
|
|
|
|
} else {
|
|
|
|
var err error
|
|
|
|
|
|
|
|
// check the tls config for secure connect
|
|
|
|
if tc := config.TLSConfig; tc != nil {
|
|
|
|
ts, err = tls.Listen("tcp", config.Address, tc)
|
|
|
|
// otherwise just plain tcp listener
|
|
|
|
} else {
|
|
|
|
ts, err = net.Listen("tcp", config.Address)
|
|
|
|
}
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
if config.MaxConn > 0 {
|
|
|
|
ts = netutil.LimitListener(ts, config.MaxConn)
|
2016-06-30 20:21:57 +01:00
|
|
|
}
|
|
|
|
|
2020-11-05 01:55:38 +03:00
|
|
|
if config.Logger.V(logger.InfoLevel) {
|
2024-04-10 00:06:50 +03:00
|
|
|
config.Logger.Info(config.Context, "Listening on "+ts.Addr().String())
|
2020-11-05 01:55:38 +03:00
|
|
|
}
|
2018-08-09 18:31:22 +08:00
|
|
|
|
2016-06-30 20:21:57 +01:00
|
|
|
h.Lock()
|
2020-12-15 11:54:15 +03:00
|
|
|
h.opts.Address = ts.Addr().String()
|
2016-06-30 20:21:57 +01:00
|
|
|
h.Unlock()
|
|
|
|
|
2021-03-09 14:19:59 +03:00
|
|
|
var handler http.Handler
|
2021-03-09 23:54:46 +03:00
|
|
|
|
2021-04-26 00:43:06 +03:00
|
|
|
// nolint: nestif
|
2021-03-09 23:54:46 +03:00
|
|
|
if h.opts.Context != nil {
|
|
|
|
if hs, ok := h.opts.Context.Value(serverKey{}).(*http.Server); ok && hs != nil {
|
|
|
|
if hs.Handler == nil && h.hd != nil {
|
|
|
|
if hdlr, ok := h.hd.Handler().(http.Handler); ok {
|
|
|
|
hs.Handler = hdlr
|
|
|
|
handler = hs.Handler
|
|
|
|
}
|
|
|
|
} else {
|
|
|
|
handler = hs.Handler
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2022-01-23 02:00:15 +03:00
|
|
|
switch {
|
|
|
|
case handler == nil && h.hd == nil:
|
2021-03-09 14:19:59 +03:00
|
|
|
handler = h
|
2022-01-23 02:40:24 +03:00
|
|
|
case len(h.handlers) > 0 && h.hd != nil:
|
|
|
|
handler = h
|
2022-01-23 02:00:15 +03:00
|
|
|
case handler == nil && h.hd != nil:
|
2021-03-09 23:54:46 +03:00
|
|
|
if hdlr, ok := h.hd.Handler().(http.Handler); ok {
|
|
|
|
handler = hdlr
|
|
|
|
}
|
2021-02-05 18:31:51 +03:00
|
|
|
}
|
|
|
|
|
2021-03-09 23:54:46 +03:00
|
|
|
if handler == nil {
|
|
|
|
return fmt.Errorf("cant process with nil handler")
|
|
|
|
}
|
2016-06-30 20:21:57 +01:00
|
|
|
|
2020-12-15 11:54:15 +03:00
|
|
|
if err := config.Broker.Connect(h.opts.Context); err != nil {
|
2019-02-27 12:28:03 +08:00
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
2020-12-15 11:54:15 +03:00
|
|
|
if err := config.RegisterCheck(h.opts.Context); err != nil {
|
2020-11-05 01:55:38 +03:00
|
|
|
if config.Logger.V(logger.ErrorLevel) {
|
2024-04-10 00:15:54 +03:00
|
|
|
config.Logger.Error(config.Context, fmt.Sprintf("Server %s-%s register check error", config.Name, config.ID), err)
|
2020-10-10 00:38:35 +03:00
|
|
|
}
|
|
|
|
} else {
|
|
|
|
if err = h.Register(); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
2019-02-27 12:28:03 +08:00
|
|
|
}
|
2019-01-24 13:58:56 +00:00
|
|
|
|
2023-05-13 16:09:21 +03:00
|
|
|
if err := h.subscribe(); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
2021-02-07 13:27:28 +03:00
|
|
|
fn := handler
|
2021-02-07 19:03:38 +03:00
|
|
|
|
2024-03-09 10:06:13 +03:00
|
|
|
var hs *http.Server
|
2021-02-07 13:27:28 +03:00
|
|
|
if h.opts.Context != nil {
|
|
|
|
if mwf, ok := h.opts.Context.Value(middlewareKey{}).([]func(http.Handler) http.Handler); ok && len(mwf) > 0 {
|
|
|
|
// wrap the handler func
|
|
|
|
for i := len(mwf); i > 0; i-- {
|
|
|
|
fn = mwf[i-1](fn)
|
|
|
|
}
|
|
|
|
}
|
2024-03-09 10:06:13 +03:00
|
|
|
var ok bool
|
|
|
|
if hs, ok = h.opts.Context.Value(serverKey{}).(*http.Server); ok && hs != nil {
|
2021-02-07 19:03:38 +03:00
|
|
|
hs.Handler = fn
|
2024-03-12 16:44:34 +03:00
|
|
|
} else {
|
|
|
|
hs = &http.Server{Handler: fn}
|
2021-02-07 19:03:38 +03:00
|
|
|
}
|
2021-02-07 13:27:28 +03:00
|
|
|
}
|
|
|
|
|
2024-03-09 10:06:13 +03:00
|
|
|
go func() {
|
|
|
|
if cerr := hs.Serve(ts); cerr != nil && !errors.Is(cerr, net.ErrClosed) {
|
2024-04-10 00:06:50 +03:00
|
|
|
h.opts.Logger.Error(h.opts.Context, "serve error", cerr)
|
2024-03-09 10:06:13 +03:00
|
|
|
}
|
|
|
|
}()
|
2016-06-30 20:21:57 +01:00
|
|
|
|
|
|
|
go func() {
|
2019-02-01 17:11:25 +08:00
|
|
|
t := new(time.Ticker)
|
|
|
|
|
|
|
|
// only process if it exists
|
2020-11-05 01:55:38 +03:00
|
|
|
if config.RegisterInterval > time.Duration(0) {
|
2019-02-01 17:11:25 +08:00
|
|
|
// new ticker
|
2020-11-05 01:55:38 +03:00
|
|
|
t = time.NewTicker(config.RegisterInterval)
|
2019-02-01 17:11:25 +08:00
|
|
|
}
|
|
|
|
|
|
|
|
// return error chan
|
|
|
|
var ch chan error
|
|
|
|
|
|
|
|
Loop:
|
|
|
|
for {
|
|
|
|
select {
|
|
|
|
// register self on interval
|
|
|
|
case <-t.C:
|
2020-10-10 00:38:35 +03:00
|
|
|
h.RLock()
|
|
|
|
registered := h.registered
|
|
|
|
h.RUnlock()
|
2020-11-05 01:55:38 +03:00
|
|
|
rerr := config.RegisterCheck(h.opts.Context)
|
2021-04-26 00:43:06 +03:00
|
|
|
// nolint: nestif
|
2020-10-10 00:38:35 +03:00
|
|
|
if rerr != nil && registered {
|
2020-11-05 01:55:38 +03:00
|
|
|
if config.Logger.V(logger.ErrorLevel) {
|
2024-04-10 00:15:54 +03:00
|
|
|
config.Logger.Error(config.Context, fmt.Sprintf("Server %s-%s register check error, deregister it", config.Name, config.ID), rerr)
|
2020-10-10 00:38:35 +03:00
|
|
|
}
|
|
|
|
// deregister self in case of error
|
|
|
|
if err := h.Deregister(); err != nil {
|
2020-11-05 01:55:38 +03:00
|
|
|
if config.Logger.V(logger.ErrorLevel) {
|
2024-04-10 00:06:50 +03:00
|
|
|
config.Logger.Error(config.Context, fmt.Sprintf("Server %s-%s deregister error", config.Name, config.ID), err)
|
2020-10-10 00:38:35 +03:00
|
|
|
}
|
|
|
|
}
|
|
|
|
} else if rerr != nil && !registered {
|
2020-11-05 01:55:38 +03:00
|
|
|
if config.Logger.V(logger.ErrorLevel) {
|
2024-04-10 00:06:50 +03:00
|
|
|
config.Logger.Error(config.Context, fmt.Sprintf("Server %s-%s register check error", config.Name, config.ID), rerr)
|
2020-10-10 00:38:35 +03:00
|
|
|
}
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
if err := h.Register(); err != nil {
|
2020-11-05 01:55:38 +03:00
|
|
|
if config.Logger.V(logger.ErrorLevel) {
|
2024-04-10 00:15:54 +03:00
|
|
|
config.Logger.Error(config.Context, fmt.Sprintf("Server %s-%s register error", config.Name, config.ID), err)
|
2020-10-10 00:38:35 +03:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2019-02-01 17:11:25 +08:00
|
|
|
if err := h.Register(); err != nil {
|
2024-04-10 00:06:50 +03:00
|
|
|
config.Logger.Error(config.Context, "Server register error", err)
|
2019-02-01 17:11:25 +08:00
|
|
|
}
|
|
|
|
// wait for exit
|
|
|
|
case ch = <-h.exit:
|
|
|
|
break Loop
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2019-01-24 13:58:56 +00:00
|
|
|
// deregister
|
2020-12-15 11:54:15 +03:00
|
|
|
if err := h.Deregister(); err != nil {
|
2024-04-10 00:06:50 +03:00
|
|
|
config.Logger.Error(config.Context, "Server deregister error", err)
|
2020-12-15 11:54:15 +03:00
|
|
|
}
|
2019-01-24 13:58:56 +00:00
|
|
|
|
2020-12-15 11:54:15 +03:00
|
|
|
if err := config.Broker.Disconnect(config.Context); err != nil {
|
2024-04-10 00:06:50 +03:00
|
|
|
config.Logger.Error(config.Context, "Broker disconnect error", err)
|
2020-12-15 11:54:15 +03:00
|
|
|
}
|
2021-05-13 10:59:29 +03:00
|
|
|
|
2024-03-09 10:06:13 +03:00
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), h.opts.GracefulTimeout)
|
|
|
|
defer cancel()
|
|
|
|
|
|
|
|
err := hs.Shutdown(ctx)
|
|
|
|
if err != nil {
|
|
|
|
err = hs.Close()
|
|
|
|
}
|
|
|
|
|
|
|
|
ch <- err
|
2016-06-30 20:21:57 +01:00
|
|
|
}()
|
|
|
|
|
2019-02-27 12:28:03 +08:00
|
|
|
return nil
|
2016-06-30 20:21:57 +01:00
|
|
|
}
|
|
|
|
|
2023-05-09 18:34:24 +03:00
|
|
|
func (h *Server) Stop() error {
|
2016-06-30 20:21:57 +01:00
|
|
|
ch := make(chan error)
|
|
|
|
h.exit <- ch
|
|
|
|
return <-ch
|
|
|
|
}
|
|
|
|
|
2023-05-09 18:34:24 +03:00
|
|
|
func (h *Server) String() string {
|
2016-06-30 20:21:57 +01:00
|
|
|
return "http"
|
|
|
|
}
|
|
|
|
|
2023-05-09 18:34:24 +03:00
|
|
|
func (h *Server) Name() string {
|
2021-01-29 14:32:32 +03:00
|
|
|
return h.opts.Name
|
|
|
|
}
|
|
|
|
|
2023-05-09 18:34:24 +03:00
|
|
|
func NewServer(opts ...server.Option) *Server {
|
2020-10-10 00:38:35 +03:00
|
|
|
options := server.NewOptions(opts...)
|
2023-01-23 09:23:38 +03:00
|
|
|
eh := DefaultErrorHandler
|
|
|
|
if v, ok := options.Context.Value(errorHandlerKey{}).(errorHandler); ok && v != nil {
|
|
|
|
eh = v
|
|
|
|
}
|
2023-05-09 18:34:24 +03:00
|
|
|
return &Server{
|
2021-03-31 00:49:55 +03:00
|
|
|
opts: options,
|
|
|
|
exit: make(chan chan error),
|
|
|
|
subscribers: make(map[*httpSubscriber][]broker.Subscriber),
|
2023-01-23 09:23:38 +03:00
|
|
|
errorHandler: eh,
|
2021-09-30 03:00:24 +03:00
|
|
|
pathHandlers: rhttp.NewTrie(),
|
2016-06-30 20:21:57 +01:00
|
|
|
}
|
|
|
|
}
|