Compare commits
48 Commits
Author | SHA1 | Date |
---|---|---|
Василий Толстов | 0838d2ab9b | |
Василий Толстов | c95efdbf53 | |
Василий Толстов | c81960af51 | |
Василий Толстов | 1c36c1685f | |
Василий Толстов | e889545210 | |
Кирилл Горбунов | 526e414d60 | |
Василий Толстов | db423dff58 | |
Василий Толстов | 542d4cec00 | |
Василий Толстов | 0ecd1da0dc | |
Василий Толстов | 0a8755ecb7 | |
Василий Толстов | 9c29d92d7f | |
Василий Толстов | b871c1be38 | |
Василий Толстов | 74bb12e75e | |
Денис Евстигнеев | 49a95c183b | |
Денис Евстигнеев | 5e6bd93a6b | |
Василий Толстов | 9ef26caf40 | |
Василий Толстов | b3e58d2cb6 | |
Василий Толстов | b89d9fdc5b | |
Кирилл Горбунов | 95dcdd6025 | |
Василий Толстов | abe5be3ddc | |
Денис Евстигнеев | c3e6cdd973 | |
Василий Толстов | 76dcf3af67 | |
Василий Толстов | 3e30960694 | |
Василий Толстов | 1643393377 | |
Василий Толстов | 313ae201af | |
Василий Толстов | 069eaf4485 | |
Василий Толстов | fd670155aa | |
Василий Толстов | 0e3a199e16 | |
Василий Толстов | a665b69706 | |
Василий Толстов | 90b7b7ec1c | |
Василий Толстов | 63ae848043 | |
Василий Толстов | 4d378bbd51 | |
Василий Толстов | 60d165f907 | |
Василий Толстов | 5a37de7d74 | |
Василий Толстов | c7d8b6a3a4 | |
Василий Толстов | cd3e2526b4 | |
Василий Толстов | 10ae1741fc | |
Василий Толстов | ca6cea861f | |
Василий Толстов | 002d5c9c53 | |
Василий Толстов | 1d25e59860 | |
Василий Толстов | ce26e8bf63 | |
Василий Толстов | 8e09142b90 | |
Василий Толстов | 13bf2bbb1e | |
Василий Толстов | ba292901d7 | |
Василий Толстов | 6d1b6c7014 | |
Василий Толстов | 1276f99159 | |
Василий Толстов | ee5cbbbcb6 | |
Василий Толстов | ce0abd2de3 |
|
@ -0,0 +1,24 @@
|
|||
# Binaries for programs and plugins
|
||||
*.exe
|
||||
*.exe~
|
||||
*.dll
|
||||
*.so
|
||||
*.dylib
|
||||
bin
|
||||
|
||||
# Test binary, built with `go test -c`
|
||||
*.test
|
||||
|
||||
# Output of the go coverage tool, specifically when used with LiteIDE
|
||||
*.out
|
||||
|
||||
# Dependency directories (remove the comment below to include it)
|
||||
# vendor/
|
||||
|
||||
# Go workspace file
|
||||
go.work
|
||||
|
||||
# General
|
||||
.DS_Store
|
||||
.idea
|
||||
.vscode
|
18
go.mod
18
go.mod
|
@ -3,14 +3,20 @@ module go.unistack.org/micro-server-http/v3
|
|||
go 1.18
|
||||
|
||||
require (
|
||||
go.unistack.org/micro-proto/v3 v3.3.1
|
||||
go.unistack.org/micro/v3 v3.10.14
|
||||
golang.org/x/net v0.7.0
|
||||
go.unistack.org/micro-client-http/v3 v3.9.11
|
||||
go.unistack.org/micro-codec-yaml/v3 v3.10.0
|
||||
go.unistack.org/micro-proto/v3 v3.4.1
|
||||
go.unistack.org/micro/v3 v3.10.75
|
||||
golang.org/x/net v0.24.0
|
||||
)
|
||||
|
||||
require (
|
||||
github.com/golang/protobuf v1.5.2 // indirect
|
||||
github.com/google/gnostic v0.6.9 // indirect
|
||||
google.golang.org/protobuf v1.28.1 // indirect
|
||||
github.com/google/gnostic v0.7.0 // indirect
|
||||
github.com/google/gnostic-models v0.6.9-0.20230804172637-c7be7c783f49 // indirect
|
||||
golang.org/x/sys v0.20.0 // indirect
|
||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20240429193739-8cf5692501f6 // indirect
|
||||
google.golang.org/grpc v1.63.2 // indirect
|
||||
google.golang.org/protobuf v1.34.0 // indirect
|
||||
gopkg.in/yaml.v3 v3.0.1 // indirect
|
||||
sigs.k8s.io/yaml v1.4.0 // indirect
|
||||
)
|
||||
|
|
431
handler.go
431
handler.go
|
@ -6,14 +6,20 @@ import (
|
|||
"io"
|
||||
"net/http"
|
||||
"reflect"
|
||||
"slices"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"go.unistack.org/micro/v3/errors"
|
||||
"go.unistack.org/micro/v3/logger"
|
||||
"go.unistack.org/micro/v3/metadata"
|
||||
"go.unistack.org/micro/v3/options"
|
||||
"go.unistack.org/micro/v3/register"
|
||||
"go.unistack.org/micro/v3/semconv"
|
||||
"go.unistack.org/micro/v3/server"
|
||||
"go.unistack.org/micro/v3/tracer"
|
||||
rhttp "go.unistack.org/micro/v3/util/http"
|
||||
rflutil "go.unistack.org/micro/v3/util/reflect"
|
||||
)
|
||||
|
@ -60,20 +66,298 @@ func (h *httpHandler) Options() server.HandlerOptions {
|
|||
return h.opts
|
||||
}
|
||||
|
||||
func (h *httpServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
// check for http.HandlerFunc handlers
|
||||
if ph, _, err := h.pathHandlers.Search(r.Method, r.URL.Path); err == nil {
|
||||
ph.(http.HandlerFunc)(w, r)
|
||||
return
|
||||
func (h *Server) HTTPHandlerFunc(handler interface{}) (http.HandlerFunc, error) {
|
||||
if handler == nil {
|
||||
return nil, fmt.Errorf("invalid handler specified: %v", handler)
|
||||
}
|
||||
|
||||
rtype := reflect.TypeOf(handler)
|
||||
if rtype.NumIn() != 3 {
|
||||
return nil, fmt.Errorf("invalid handler, NumIn != 3: %v", rtype.NumIn())
|
||||
}
|
||||
|
||||
argType := rtype.In(1)
|
||||
replyType := rtype.In(2)
|
||||
|
||||
// First arg need not be a pointer.
|
||||
if !isExportedOrBuiltinType(argType) {
|
||||
return nil, fmt.Errorf("invalid handler, argument type not exported: %v", argType)
|
||||
}
|
||||
|
||||
if replyType.Kind() != reflect.Ptr {
|
||||
return nil, fmt.Errorf("invalid handler, reply type not a pointer: %v", replyType)
|
||||
}
|
||||
|
||||
// Reply type must be exported.
|
||||
if !isExportedOrBuiltinType(replyType) {
|
||||
return nil, fmt.Errorf("invalid handler, reply type not exported: %v", replyType)
|
||||
}
|
||||
|
||||
if rtype.NumOut() != 1 {
|
||||
return nil, fmt.Errorf("invalid handler, has wrong number of outs: %v", rtype.NumOut())
|
||||
}
|
||||
|
||||
// The return type of the method must be error.
|
||||
if returnType := rtype.Out(0); returnType != typeOfError {
|
||||
return nil, fmt.Errorf("invalid handler, returns %v not error", returnType.String())
|
||||
}
|
||||
|
||||
return func(w http.ResponseWriter, r *http.Request) {
|
||||
ct := DefaultContentType
|
||||
if htype := r.Header.Get(metadata.HeaderContentType); htype != "" {
|
||||
ct = htype
|
||||
}
|
||||
|
||||
ctx := context.WithValue(r.Context(), rspCodeKey{}, &rspCodeVal{})
|
||||
ctx = context.WithValue(ctx, rspHeaderKey{}, &rspHeaderVal{})
|
||||
md, ok := metadata.FromIncomingContext(ctx)
|
||||
if !ok {
|
||||
md = metadata.New(len(r.Header) + 8)
|
||||
}
|
||||
for k, v := range r.Header {
|
||||
md[k] = strings.Join(v, ", ")
|
||||
}
|
||||
md["RemoteAddr"] = r.RemoteAddr
|
||||
md["Method"] = r.Method
|
||||
md["URL"] = r.URL.String()
|
||||
md["Proto"] = r.Proto
|
||||
md["Content-Length"] = fmt.Sprintf("%d", r.ContentLength)
|
||||
md["Transfer-Encoding"] = strings.Join(r.TransferEncoding, ",")
|
||||
md["Host"] = r.Host
|
||||
md["RequestURI"] = r.RequestURI
|
||||
if r.TLS != nil {
|
||||
md["TLS"] = "true"
|
||||
md["TLS-ALPN"] = r.TLS.NegotiatedProtocol
|
||||
md["TLS-ServerName"] = r.TLS.ServerName
|
||||
}
|
||||
|
||||
ctx = metadata.NewIncomingContext(ctx, md)
|
||||
|
||||
path := r.URL.Path
|
||||
|
||||
if r.Body != nil {
|
||||
defer r.Body.Close()
|
||||
}
|
||||
|
||||
matches := make(map[string]interface{})
|
||||
var match bool
|
||||
var hldr *patHandler
|
||||
var handler *httpHandler
|
||||
|
||||
for _, shdlr := range h.handlers {
|
||||
hdlr := shdlr.(*httpHandler)
|
||||
fh, mp, err := hdlr.handlers.Search(r.Method, path)
|
||||
if err == nil {
|
||||
match = true
|
||||
for k, v := range mp {
|
||||
matches[k] = v
|
||||
}
|
||||
hldr = fh.(*patHandler)
|
||||
handler = hdlr
|
||||
break
|
||||
} else if err == rhttp.ErrMethodNotAllowed && !h.registerRPC {
|
||||
w.WriteHeader(http.StatusMethodNotAllowed)
|
||||
_, _ = w.Write([]byte("not matching route found"))
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
if !match && h.registerRPC {
|
||||
microMethod, mok := md.Get(metadata.HeaderEndpoint)
|
||||
if mok {
|
||||
serviceMethod := strings.Split(microMethod, ".")
|
||||
if len(serviceMethod) == 2 {
|
||||
if shdlr, ok := h.handlers[serviceMethod[0]]; ok {
|
||||
hdlr := shdlr.(*httpHandler)
|
||||
fh, mp, err := hdlr.handlers.Search(http.MethodPost, "/"+microMethod)
|
||||
if err == nil {
|
||||
match = true
|
||||
for k, v := range mp {
|
||||
matches[k] = v
|
||||
}
|
||||
hldr = fh.(*patHandler)
|
||||
handler = hdlr
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// get fields from url values
|
||||
if len(r.URL.RawQuery) > 0 {
|
||||
umd, cerr := rflutil.URLMap(r.URL.RawQuery)
|
||||
if cerr != nil {
|
||||
w.WriteHeader(http.StatusInternalServerError)
|
||||
_, _ = w.Write([]byte(cerr.Error()))
|
||||
return
|
||||
}
|
||||
for k, v := range umd {
|
||||
matches[k] = v
|
||||
}
|
||||
}
|
||||
|
||||
cf, err := h.newCodec(ct)
|
||||
if err != nil {
|
||||
w.WriteHeader(http.StatusBadRequest)
|
||||
_, _ = w.Write([]byte(err.Error()))
|
||||
return
|
||||
}
|
||||
|
||||
var argv, replyv reflect.Value
|
||||
|
||||
// Decode the argument value.
|
||||
argIsValue := false // if true, need to indirect before calling.
|
||||
if hldr.mtype.ArgType.Kind() == reflect.Ptr {
|
||||
argv = reflect.New(hldr.mtype.ArgType.Elem())
|
||||
} else {
|
||||
argv = reflect.New(hldr.mtype.ArgType)
|
||||
argIsValue = true
|
||||
}
|
||||
|
||||
if argIsValue {
|
||||
argv = argv.Elem()
|
||||
}
|
||||
|
||||
// reply value
|
||||
replyv = reflect.New(hldr.mtype.ReplyType.Elem())
|
||||
|
||||
function := hldr.mtype.method.Func
|
||||
var returnValues []reflect.Value
|
||||
|
||||
if r.Body != nil {
|
||||
var buf []byte
|
||||
buf, err = io.ReadAll(r.Body)
|
||||
if err != nil && err != io.EOF {
|
||||
h.errorHandler(ctx, handler, w, r, err, http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
if err = cf.Unmarshal(buf, argv.Interface()); err != nil {
|
||||
h.errorHandler(ctx, handler, w, r, err, http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
matches = rflutil.FlattenMap(matches)
|
||||
if err = rflutil.Merge(argv.Interface(), matches, rflutil.SliceAppend(true), rflutil.Tags([]string{"protobuf", "json"})); err != nil {
|
||||
h.errorHandler(ctx, handler, w, r, err, http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
hr := &rpcRequest{
|
||||
codec: cf,
|
||||
service: handler.sopts.Name,
|
||||
contentType: ct,
|
||||
method: fmt.Sprintf("%s.%s", hldr.name, hldr.mtype.method.Name),
|
||||
endpoint: fmt.Sprintf("%s.%s", hldr.name, hldr.mtype.method.Name),
|
||||
payload: argv.Interface(),
|
||||
header: md,
|
||||
}
|
||||
|
||||
// define the handler func
|
||||
fn := func(fctx context.Context, req server.Request, rsp interface{}) (err error) {
|
||||
returnValues = function.Call([]reflect.Value{hldr.rcvr, hldr.mtype.prepareContext(fctx), argv, reflect.ValueOf(rsp)})
|
||||
|
||||
// The return value for the method is an error.
|
||||
if rerr := returnValues[0].Interface(); rerr != nil {
|
||||
err = rerr.(error)
|
||||
}
|
||||
|
||||
md, ok := metadata.FromOutgoingContext(ctx)
|
||||
if !ok {
|
||||
md = metadata.New(0)
|
||||
}
|
||||
if nmd, ok := metadata.FromOutgoingContext(fctx); ok {
|
||||
for k, v := range nmd {
|
||||
md.Set(k, v)
|
||||
}
|
||||
}
|
||||
metadata.SetOutgoingContext(ctx, md)
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
// wrap the handler func
|
||||
h.opts.Hooks.EachNext(func(hook options.Hook) {
|
||||
if h, ok := hook.(server.HookHandler); ok {
|
||||
fn = h(fn)
|
||||
}
|
||||
})
|
||||
|
||||
if ct == "application/x-www-form-urlencoded" {
|
||||
cf, err = h.newCodec(DefaultContentType)
|
||||
if err != nil {
|
||||
h.errorHandler(ctx, handler, w, r, err, http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
ct = DefaultContentType
|
||||
}
|
||||
|
||||
scode := int(200)
|
||||
appErr := fn(ctx, hr, replyv.Interface())
|
||||
|
||||
w.Header().Set(metadata.HeaderContentType, ct)
|
||||
if md, ok := metadata.FromOutgoingContext(ctx); ok {
|
||||
for k, v := range md {
|
||||
w.Header().Set(k, v)
|
||||
}
|
||||
}
|
||||
if md := getRspHeader(ctx); md != nil {
|
||||
for k, v := range md {
|
||||
for _, vv := range v {
|
||||
w.Header().Add(k, vv)
|
||||
}
|
||||
}
|
||||
}
|
||||
if nct := w.Header().Get(metadata.HeaderContentType); nct != ct {
|
||||
if cf, err = h.newCodec(nct); err != nil {
|
||||
h.errorHandler(ctx, nil, w, r, err, http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
var buf []byte
|
||||
if appErr != nil {
|
||||
switch verr := appErr.(type) {
|
||||
case *errors.Error:
|
||||
scode = int(verr.Code)
|
||||
buf, err = cf.Marshal(verr)
|
||||
case *Error:
|
||||
buf, err = cf.Marshal(verr.err)
|
||||
default:
|
||||
buf, err = cf.Marshal(appErr)
|
||||
}
|
||||
} else {
|
||||
buf, err = cf.Marshal(replyv.Interface())
|
||||
}
|
||||
|
||||
if err != nil && handler.sopts.Logger.V(logger.ErrorLevel) {
|
||||
handler.sopts.Logger.Errorf(handler.sopts.Context, "handler err: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
if nscode := GetRspCode(ctx); nscode != 0 {
|
||||
scode = nscode
|
||||
}
|
||||
w.WriteHeader(scode)
|
||||
|
||||
if _, cerr := w.Write(buf); cerr != nil {
|
||||
handler.sopts.Logger.Errorf(ctx, "write failed: %v", cerr)
|
||||
}
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (h *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
ct := DefaultContentType
|
||||
if htype := r.Header.Get(metadata.HeaderContentType); htype != "" {
|
||||
ct = htype
|
||||
}
|
||||
|
||||
ts := time.Now()
|
||||
|
||||
ctx := context.WithValue(r.Context(), rspCodeKey{}, &rspCodeVal{})
|
||||
ctx = context.WithValue(ctx, rspHeaderKey{}, &rspHeaderVal{})
|
||||
|
||||
md, ok := metadata.FromIncomingContext(ctx)
|
||||
if !ok {
|
||||
md = metadata.New(len(r.Header) + 8)
|
||||
|
@ -82,17 +366,22 @@ func (h *httpServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
|||
md[k] = strings.Join(v, ", ")
|
||||
}
|
||||
md["RemoteAddr"] = r.RemoteAddr
|
||||
if r.TLS != nil {
|
||||
md["Scheme"] = "https"
|
||||
} else {
|
||||
md["Scheme"] = "http"
|
||||
}
|
||||
md["Method"] = r.Method
|
||||
md["URL"] = r.URL.String()
|
||||
md["Proto"] = r.Proto
|
||||
md["ContentLength"] = fmt.Sprintf("%d", r.ContentLength)
|
||||
md["TransferEncoding"] = strings.Join(r.TransferEncoding, ",")
|
||||
if len(r.TransferEncoding) > 0 {
|
||||
md["TransferEncoding"] = strings.Join(r.TransferEncoding, ",")
|
||||
}
|
||||
md["Host"] = r.Host
|
||||
md["RequestURI"] = r.RequestURI
|
||||
ctx = metadata.NewIncomingContext(ctx, md)
|
||||
|
||||
defer r.Body.Close()
|
||||
|
||||
path := r.URL.Path
|
||||
if !strings.HasPrefix(path, "/") {
|
||||
h.errorHandler(ctx, nil, w, r, fmt.Errorf("path must starts with /"), http.StatusBadRequest)
|
||||
|
@ -143,16 +432,87 @@ func (h *httpServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
|||
}
|
||||
}
|
||||
|
||||
var sp tracer.Span
|
||||
if !match && h.hd != nil {
|
||||
if hdlr, ok := h.hd.Handler().(http.Handler); ok {
|
||||
hdlr.ServeHTTP(w, r)
|
||||
h.opts.Meter.Counter(semconv.ServerRequestInflight, "endpoint", h.hd.Name()).Inc()
|
||||
ctx, sp = h.opts.Tracer.Start(ctx, h.hd.Name()+" rpc-server",
|
||||
tracer.WithSpanKind(tracer.SpanKindServer),
|
||||
tracer.WithSpanLabels(
|
||||
"endpoint", h.hd.Name(),
|
||||
),
|
||||
)
|
||||
hdlr.ServeHTTP(w, r.WithContext(ctx))
|
||||
n := GetRspCode(ctx)
|
||||
if n > 399 {
|
||||
h.opts.Meter.Counter(semconv.ServerRequestTotal, "endpoint", h.hd.Name(), "status", "success", "code", strconv.Itoa(n)).Inc()
|
||||
if s, _ := sp.Status(); s != tracer.SpanStatusError {
|
||||
sp.SetStatus(tracer.SpanStatusError, http.StatusText(n))
|
||||
}
|
||||
} else {
|
||||
h.opts.Meter.Counter(semconv.ServerRequestTotal, "endpoint", h.hd.Name(), "status", "failure", "code", strconv.Itoa(n)).Inc()
|
||||
}
|
||||
te := time.Since(ts)
|
||||
h.opts.Meter.Summary(semconv.ServerRequestLatencyMicroseconds, "endpoint", h.hd.Name()).Update(te.Seconds())
|
||||
h.opts.Meter.Histogram(semconv.ServerRequestDurationSeconds, "endpoint", h.hd.Name()).Update(te.Seconds())
|
||||
h.opts.Meter.Counter(semconv.ServerRequestInflight, "endpoint", h.hd.Name()).Dec()
|
||||
sp.Finish()
|
||||
return
|
||||
}
|
||||
} else if !match {
|
||||
// check for http.HandlerFunc handlers
|
||||
ctx, sp = h.opts.Tracer.Start(ctx, r.URL.Path+" rpc-server",
|
||||
tracer.WithSpanKind(tracer.SpanKindServer),
|
||||
tracer.WithSpanLabels(
|
||||
"endpoint", r.URL.Path,
|
||||
),
|
||||
)
|
||||
if ph, _, err := h.pathHandlers.Search(r.Method, r.URL.Path); err == nil {
|
||||
ph.(http.HandlerFunc)(w, r.WithContext(ctx))
|
||||
if n := GetRspCode(ctx); n > 399 {
|
||||
sp.SetStatus(tracer.SpanStatusError, http.StatusText(n))
|
||||
}
|
||||
sp.Finish()
|
||||
return
|
||||
}
|
||||
h.errorHandler(ctx, nil, w, r, fmt.Errorf("not matching route found"), http.StatusNotFound)
|
||||
sp.SetStatus(tracer.SpanStatusError, http.StatusText(http.StatusNotFound))
|
||||
sp.Finish()
|
||||
return
|
||||
}
|
||||
|
||||
endpointName := fmt.Sprintf("%s.%s", hldr.name, hldr.mtype.method.Name)
|
||||
topts := []tracer.SpanOption{
|
||||
tracer.WithSpanKind(tracer.SpanKindServer),
|
||||
tracer.WithSpanLabels(
|
||||
"endpoint", endpointName,
|
||||
),
|
||||
}
|
||||
|
||||
if slices.Contains(tracer.DefaultSkipEndpoints, endpointName) {
|
||||
topts = append(topts, tracer.WithSpanRecord(false))
|
||||
}
|
||||
|
||||
ctx, sp = h.opts.Tracer.Start(ctx, endpointName+" rpc-server", topts...)
|
||||
|
||||
defer func() {
|
||||
te := time.Since(ts)
|
||||
h.opts.Meter.Summary(semconv.ServerRequestLatencyMicroseconds, "endpoint", handler.name).Update(te.Seconds())
|
||||
h.opts.Meter.Histogram(semconv.ServerRequestDurationSeconds, "endpoint", handler.name).Update(te.Seconds())
|
||||
h.opts.Meter.Counter(semconv.ServerRequestInflight, "endpoint", handler.name).Dec()
|
||||
|
||||
n := GetRspCode(ctx)
|
||||
if n > 399 {
|
||||
if s, _ := sp.Status(); s != tracer.SpanStatusError {
|
||||
sp.SetStatus(tracer.SpanStatusError, http.StatusText(n))
|
||||
}
|
||||
h.opts.Meter.Counter(semconv.ServerRequestTotal, "endpoint", handler.name, "status", "failure", "code", strconv.Itoa(n)).Inc()
|
||||
} else {
|
||||
h.opts.Meter.Counter(semconv.ServerRequestTotal, "endpoint", handler.name, "status", "success", "code", strconv.Itoa(n)).Inc()
|
||||
}
|
||||
sp.Finish()
|
||||
}()
|
||||
|
||||
// get fields from url values
|
||||
if len(r.URL.RawQuery) > 0 {
|
||||
umd, cerr := rflutil.URLMap(r.URL.RawQuery)
|
||||
|
@ -165,6 +525,10 @@ func (h *httpServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
|||
}
|
||||
}
|
||||
|
||||
if r.Body != nil {
|
||||
defer r.Body.Close()
|
||||
}
|
||||
|
||||
cf, err := h.newCodec(ct)
|
||||
if err != nil {
|
||||
h.errorHandler(ctx, nil, w, r, err, http.StatusBadRequest)
|
||||
|
@ -192,21 +556,26 @@ func (h *httpServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
|||
function := hldr.mtype.method.Func
|
||||
var returnValues []reflect.Value
|
||||
|
||||
buf, err := io.ReadAll(r.Body)
|
||||
if err != nil && err != io.EOF {
|
||||
h.errorHandler(ctx, handler, w, r, err, http.StatusInternalServerError)
|
||||
return
|
||||
if r.Body != nil {
|
||||
var buf []byte
|
||||
buf, err = io.ReadAll(r.Body)
|
||||
if err != nil && err != io.EOF {
|
||||
h.errorHandler(ctx, handler, w, r, err, http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
if err = cf.Unmarshal(buf, argv.Interface()); err != nil {
|
||||
h.errorHandler(ctx, handler, w, r, err, http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
if err = cf.Unmarshal(buf, argv.Interface()); err != nil {
|
||||
h.errorHandler(ctx, handler, w, r, err, http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
matches = rflutil.FlattenMap(matches)
|
||||
if err = rflutil.Merge(argv.Interface(), matches, rflutil.SliceAppend(true), rflutil.Tags([]string{"protobuf", "json"})); err != nil {
|
||||
h.errorHandler(ctx, handler, w, r, err, http.StatusBadRequest)
|
||||
return
|
||||
if len(matches) > 0 {
|
||||
matches = rflutil.FlattenMap(matches)
|
||||
if err = rflutil.Merge(argv.Interface(), matches, rflutil.SliceAppend(true), rflutil.Tags([]string{"protobuf", "json"})); err != nil {
|
||||
h.errorHandler(ctx, handler, w, r, err, http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
hr := &rpcRequest{
|
||||
|
@ -239,13 +608,18 @@ func (h *httpServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
|||
}
|
||||
metadata.SetOutgoingContext(ctx, md)
|
||||
|
||||
if err != nil && sp != nil {
|
||||
sp.SetStatus(tracer.SpanStatusError, err.Error())
|
||||
}
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
// wrap the handler func
|
||||
for i := len(handler.sopts.HdlrWrappers); i > 0; i-- {
|
||||
fn = handler.sopts.HdlrWrappers[i-1](fn)
|
||||
}
|
||||
h.opts.Hooks.EachNext(func(hook options.Hook) {
|
||||
if h, ok := hook.(server.HookHandler); ok {
|
||||
fn = h(fn)
|
||||
}
|
||||
})
|
||||
|
||||
if ct == "application/x-www-form-urlencoded" {
|
||||
cf, err = h.newCodec(DefaultContentType)
|
||||
|
@ -279,6 +653,7 @@ func (h *httpServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
|||
}
|
||||
}
|
||||
|
||||
var buf []byte
|
||||
if appErr != nil {
|
||||
switch verr := appErr.(type) {
|
||||
case *errors.Error:
|
||||
|
@ -294,7 +669,7 @@ func (h *httpServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
|||
}
|
||||
|
||||
if err != nil && handler.sopts.Logger.V(logger.ErrorLevel) {
|
||||
handler.sopts.Logger.Errorf(handler.sopts.Context, "handler err: %v", err)
|
||||
handler.sopts.Logger.Error(handler.sopts.Context, "handler error", err)
|
||||
return
|
||||
}
|
||||
|
||||
|
@ -304,6 +679,6 @@ func (h *httpServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
|||
w.WriteHeader(scode)
|
||||
|
||||
if _, cerr := w.Write(buf); cerr != nil {
|
||||
handler.sopts.Logger.Errorf(ctx, "write failed: %v", cerr)
|
||||
handler.sopts.Logger.Error(ctx, "respoonse write error", cerr)
|
||||
}
|
||||
}
|
||||
|
|
|
@ -0,0 +1,12 @@
|
|||
package handler
|
||||
|
||||
import (
|
||||
// import required packages
|
||||
_ "go.unistack.org/micro-proto/v3/openapiv3"
|
||||
)
|
||||
|
||||
//go:generate sh -c "curl -L https://github.com/swagger-api/swagger-ui/archive/refs/tags/v4.18.3.zip -o - | bsdtar -C swagger-ui --strip-components=2 -xv swagger-ui-4.18.3/dist && rm swagger-ui/*.map swagger-ui/*-es-*.js swagger-ui/swagger-ui.js swagger-ui/swagger-initializer.js"
|
||||
|
||||
//go:generate sh -c "protoc -I./ -I$(go list -f '{{ .Dir }}' -m go.unistack.org/micro-proto/v3) --go-micro_out='components=micro|http|server',standalone=false,debug=true,paths=source_relative:./ ./meter/meter.proto"
|
||||
|
||||
//go:generate sh -c "protoc -I./ -I$(go list -f '{{ .Dir }}' -m go.unistack.org/micro-proto/v3) --go-micro_out='components=micro|http|server',standalone=false,debug=true,paths=source_relative:./ ./health/health.proto"
|
|
@ -0,0 +1,77 @@
|
|||
//go:build ignore
|
||||
|
||||
package graphql_handler
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
|
||||
"github.com/99designs/gqlgen/graphql"
|
||||
"go.unistack.org/micro/v3/logger"
|
||||
"go.unistack.org/micro/v3/store"
|
||||
)
|
||||
|
||||
var _ graphql.Cache = (*cacheWrapper)(nil)
|
||||
|
||||
type Handler struct {
|
||||
opts Options
|
||||
}
|
||||
|
||||
type Option func(*Options)
|
||||
|
||||
type Options struct {
|
||||
cache *cacheWrapper
|
||||
Path string
|
||||
}
|
||||
|
||||
type cacheWrapper struct {
|
||||
s store.Store
|
||||
l logger.Logger
|
||||
}
|
||||
|
||||
func (c *cacheWrapper) Get(ctx context.Context, key string) (interface{}, bool) {
|
||||
var val interface{}
|
||||
if err := c.s.Read(ctx, key, val); err != nil && err != store.ErrNotFound {
|
||||
c.l.Error(ctx, fmt.Sprintf("cache.Get %s failed", key), err)
|
||||
return nil, false
|
||||
}
|
||||
return val, true
|
||||
}
|
||||
|
||||
func (c *cacheWrapper) Add(ctx context.Context, key string, val interface{}) {
|
||||
if err := c.s.Write(ctx, key, val); err != nil {
|
||||
c.l.Error(ctx, fmt.Sprintf("cache.Add %s failed", key), err)
|
||||
}
|
||||
}
|
||||
|
||||
func Store(s store.Store) Option {
|
||||
return func(o *Options) {
|
||||
if o.cache == nil {
|
||||
o.cache = &cacheWrapper{}
|
||||
}
|
||||
o.cache.s = s
|
||||
}
|
||||
}
|
||||
|
||||
func Logger(l logger.Logger) Option {
|
||||
return func(o *Options) {
|
||||
if o.cache == nil {
|
||||
o.cache = &cacheWrapper{}
|
||||
}
|
||||
o.cache.l = l
|
||||
}
|
||||
}
|
||||
|
||||
func Path(path string) Option {
|
||||
return func(o *Options) {
|
||||
o.Path = path
|
||||
}
|
||||
}
|
||||
|
||||
func NewHandler(opts ...Option) *Handler {
|
||||
options := Options{}
|
||||
for _, o := range opts {
|
||||
o(&options)
|
||||
}
|
||||
return &Handler{opts: options}
|
||||
}
|
|
@ -1,8 +0,0 @@
|
|||
package health
|
||||
|
||||
//go:generate sh -c "protoc -I./ -I$(go list -f '{{ .Dir }}' -m go.unistack.org/micro-proto/v3) --go-micro_out='components=micro|http|server',standalone=false,debug=true,paths=source_relative:./ health.proto"
|
||||
|
||||
import (
|
||||
// import required packages
|
||||
_ "go.unistack.org/micro-proto/v3/openapiv3"
|
||||
)
|
|
@ -1,4 +1,4 @@
|
|||
package health // import "go.unistack.org/micro-server-http/v3/handler/health"
|
||||
package health_handler
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
|
|
@ -1,7 +1,7 @@
|
|||
syntax = "proto3";
|
||||
|
||||
package micro.server.http.v3.handler.health;
|
||||
option go_package = "go.unistack.org/micro-server-http/v3/handler/health;health";
|
||||
option go_package = "go.unistack.org/micro-server-http/v3/handler/health;health_handler";
|
||||
|
||||
import "api/annotations.proto";
|
||||
import "openapiv3/annotations.proto";
|
||||
|
|
|
@ -1,45 +1,26 @@
|
|||
// Code generated by protoc-gen-go-micro. DO NOT EDIT.
|
||||
// versions:
|
||||
// - protoc-gen-go-micro v3.10.2
|
||||
// - protoc v3.21.12
|
||||
// source: health.proto
|
||||
// - protoc-gen-go-micro v3.10.4
|
||||
// - protoc v5.26.1
|
||||
// source: health/health.proto
|
||||
|
||||
package health
|
||||
package health_handler
|
||||
|
||||
import (
|
||||
context "context"
|
||||
codec "go.unistack.org/micro-proto/v3/codec"
|
||||
v3 "go.unistack.org/micro-server-http/v3"
|
||||
client "go.unistack.org/micro/v3/client"
|
||||
)
|
||||
|
||||
var (
|
||||
HealthServiceName = "HealthService"
|
||||
)
|
||||
var (
|
||||
HealthServiceServerEndpoints = []v3.EndpointMetadata{
|
||||
{
|
||||
Name: "HealthService.Live",
|
||||
Path: "/live",
|
||||
Method: "GET",
|
||||
Body: "",
|
||||
Stream: false,
|
||||
},
|
||||
{
|
||||
Name: "HealthService.Ready",
|
||||
Path: "/ready",
|
||||
Method: "GET",
|
||||
Body: "",
|
||||
Stream: false,
|
||||
},
|
||||
{
|
||||
Name: "HealthService.Version",
|
||||
Path: "/version",
|
||||
Method: "GET",
|
||||
Body: "",
|
||||
Stream: false,
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
type HealthServiceClient interface {
|
||||
Live(ctx context.Context, req *codec.Frame, opts ...client.CallOption) (*codec.Frame, error)
|
||||
Ready(ctx context.Context, req *codec.Frame, opts ...client.CallOption) (*codec.Frame, error)
|
||||
Version(ctx context.Context, req *codec.Frame, opts ...client.CallOption) (*codec.Frame, error)
|
||||
}
|
||||
|
||||
type HealthServiceServer interface {
|
||||
Live(ctx context.Context, req *codec.Frame, rsp *codec.Frame) error
|
||||
|
|
|
@ -1,16 +1,108 @@
|
|||
// Code generated by protoc-gen-go-micro. DO NOT EDIT.
|
||||
// protoc-gen-go-micro version: v3.10.2
|
||||
// source: health.proto
|
||||
// protoc-gen-go-micro version: v3.10.4
|
||||
// source: health/health.proto
|
||||
|
||||
package health
|
||||
package health_handler
|
||||
|
||||
import (
|
||||
context "context"
|
||||
v31 "go.unistack.org/micro-client-http/v3"
|
||||
codec "go.unistack.org/micro-proto/v3/codec"
|
||||
v3 "go.unistack.org/micro-server-http/v3"
|
||||
client "go.unistack.org/micro/v3/client"
|
||||
server "go.unistack.org/micro/v3/server"
|
||||
http "net/http"
|
||||
)
|
||||
|
||||
var (
|
||||
HealthServiceServerEndpoints = []v3.EndpointMetadata{
|
||||
{
|
||||
Name: "HealthService.Live",
|
||||
Path: "/live",
|
||||
Method: "GET",
|
||||
Body: "",
|
||||
Stream: false,
|
||||
},
|
||||
{
|
||||
Name: "HealthService.Ready",
|
||||
Path: "/ready",
|
||||
Method: "GET",
|
||||
Body: "",
|
||||
Stream: false,
|
||||
},
|
||||
{
|
||||
Name: "HealthService.Version",
|
||||
Path: "/version",
|
||||
Method: "GET",
|
||||
Body: "",
|
||||
Stream: false,
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
type healthServiceClient struct {
|
||||
c client.Client
|
||||
name string
|
||||
}
|
||||
|
||||
func NewHealthServiceClient(name string, c client.Client) HealthServiceClient {
|
||||
return &healthServiceClient{c: c, name: name}
|
||||
}
|
||||
|
||||
func (c *healthServiceClient) Live(ctx context.Context, req *codec.Frame, opts ...client.CallOption) (*codec.Frame, error) {
|
||||
errmap := make(map[string]interface{}, 1)
|
||||
errmap["default"] = &codec.Frame{}
|
||||
opts = append(opts,
|
||||
v31.ErrorMap(errmap),
|
||||
)
|
||||
opts = append(opts,
|
||||
v31.Method(http.MethodGet),
|
||||
v31.Path("/live"),
|
||||
)
|
||||
rsp := &codec.Frame{}
|
||||
err := c.c.Call(ctx, c.c.NewRequest(c.name, "HealthService.Live", req), rsp, opts...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return rsp, nil
|
||||
}
|
||||
|
||||
func (c *healthServiceClient) Ready(ctx context.Context, req *codec.Frame, opts ...client.CallOption) (*codec.Frame, error) {
|
||||
errmap := make(map[string]interface{}, 1)
|
||||
errmap["default"] = &codec.Frame{}
|
||||
opts = append(opts,
|
||||
v31.ErrorMap(errmap),
|
||||
)
|
||||
opts = append(opts,
|
||||
v31.Method(http.MethodGet),
|
||||
v31.Path("/ready"),
|
||||
)
|
||||
rsp := &codec.Frame{}
|
||||
err := c.c.Call(ctx, c.c.NewRequest(c.name, "HealthService.Ready", req), rsp, opts...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return rsp, nil
|
||||
}
|
||||
|
||||
func (c *healthServiceClient) Version(ctx context.Context, req *codec.Frame, opts ...client.CallOption) (*codec.Frame, error) {
|
||||
errmap := make(map[string]interface{}, 1)
|
||||
errmap["default"] = &codec.Frame{}
|
||||
opts = append(opts,
|
||||
v31.ErrorMap(errmap),
|
||||
)
|
||||
opts = append(opts,
|
||||
v31.Method(http.MethodGet),
|
||||
v31.Path("/version"),
|
||||
)
|
||||
rsp := &codec.Frame{}
|
||||
err := c.c.Call(ctx, c.c.NewRequest(c.name, "HealthService.Version", req), rsp, opts...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return rsp, nil
|
||||
}
|
||||
|
||||
type healthServiceServer struct {
|
||||
HealthServiceServer
|
||||
}
|
||||
|
|
|
@ -1,8 +0,0 @@
|
|||
package meter
|
||||
|
||||
//go:generate sh -c "protoc -I./ -I$(go list -f '{{ .Dir }}' -m go.unistack.org/micro-proto/v3) --go-micro_out='components=micro|http|server',standalone=false,debug=true,paths=source_relative:./ meter.proto"
|
||||
|
||||
import (
|
||||
// import required packages
|
||||
_ "go.unistack.org/micro-proto/v3/openapiv3"
|
||||
)
|
|
@ -1,14 +1,36 @@
|
|||
package meter // import "go.unistack.org/micro-server-http/v3/handler/meter"
|
||||
package meter_handler
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"compress/gzip"
|
||||
"context"
|
||||
"io"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
codecpb "go.unistack.org/micro-proto/v3/codec"
|
||||
"go.unistack.org/micro/v3/errors"
|
||||
"go.unistack.org/micro/v3/logger"
|
||||
"go.unistack.org/micro/v3/metadata"
|
||||
"go.unistack.org/micro/v3/meter"
|
||||
)
|
||||
|
||||
const (
|
||||
contentEncodingHeader = "Content-Encoding"
|
||||
acceptEncodingHeader = "Accept-Encoding"
|
||||
)
|
||||
|
||||
var gzipPool = sync.Pool{
|
||||
New: func() interface{} {
|
||||
return gzip.NewWriter(nil)
|
||||
},
|
||||
}
|
||||
|
||||
var bufPool = sync.Pool{
|
||||
New: func() interface{} {
|
||||
return bytes.NewBuffer(nil)
|
||||
},
|
||||
}
|
||||
|
||||
// guard to fail early
|
||||
var _ MeterServiceServer = &Handler{}
|
||||
|
||||
|
@ -19,9 +41,10 @@ type Handler struct {
|
|||
type Option func(*Options)
|
||||
|
||||
type Options struct {
|
||||
Meter meter.Meter
|
||||
Name string
|
||||
MeterOptions []meter.Option
|
||||
Meter meter.Meter
|
||||
Name string
|
||||
MeterOptions []meter.Option
|
||||
DisableCompress bool
|
||||
}
|
||||
|
||||
func Meter(m meter.Meter) Option {
|
||||
|
@ -36,6 +59,12 @@ func Name(name string) Option {
|
|||
}
|
||||
}
|
||||
|
||||
func DisableCompress(g bool) Option {
|
||||
return func(o *Options) {
|
||||
o.DisableCompress = g
|
||||
}
|
||||
}
|
||||
|
||||
func MeterOptions(opts ...meter.Option) Option {
|
||||
return func(o *Options) {
|
||||
o.MeterOptions = append(o.MeterOptions, opts...)
|
||||
|
@ -43,7 +72,7 @@ func MeterOptions(opts ...meter.Option) Option {
|
|||
}
|
||||
|
||||
func NewOptions(opts ...Option) Options {
|
||||
options := Options{Meter: meter.DefaultMeter}
|
||||
options := Options{Meter: meter.DefaultMeter, DisableCompress: false}
|
||||
for _, o := range opts {
|
||||
o(&options)
|
||||
}
|
||||
|
@ -56,12 +85,48 @@ func NewHandler(opts ...Option) *Handler {
|
|||
}
|
||||
|
||||
func (h *Handler) Metrics(ctx context.Context, req *codecpb.Frame, rsp *codecpb.Frame) error {
|
||||
buf := bytes.NewBuffer(nil)
|
||||
if err := h.opts.Meter.Write(buf, h.opts.MeterOptions...); err != nil {
|
||||
return errors.InternalServerError(h.opts.Name, "%v", err)
|
||||
log, ok := logger.FromContext(ctx)
|
||||
if !ok {
|
||||
log = logger.DefaultLogger
|
||||
}
|
||||
|
||||
buf := bufPool.Get().(*bytes.Buffer)
|
||||
defer bufPool.Put(buf)
|
||||
buf.Reset()
|
||||
|
||||
w := io.Writer(buf)
|
||||
|
||||
if md, ok := metadata.FromOutgoingContext(ctx); gzipAccepted(md) && ok && !h.opts.DisableCompress {
|
||||
omd, _ := metadata.FromOutgoingContext(ctx)
|
||||
omd.Set(contentEncodingHeader, "gzip")
|
||||
gz := gzipPool.Get().(*gzip.Writer)
|
||||
defer gzipPool.Put(gz)
|
||||
|
||||
gz.Reset(w)
|
||||
defer gz.Close()
|
||||
|
||||
w = gz
|
||||
gz.Flush()
|
||||
}
|
||||
|
||||
if err := h.opts.Meter.Write(w, h.opts.MeterOptions...); err != nil {
|
||||
log.Error(ctx, "http/meter write failed", err)
|
||||
return nil
|
||||
}
|
||||
|
||||
rsp.Data = buf.Bytes()
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// gzipAccepted returns whether the client will accept gzip-encoded content.
|
||||
func gzipAccepted(md metadata.Metadata) bool {
|
||||
a, ok := md.Get(acceptEncodingHeader)
|
||||
if !ok {
|
||||
return false
|
||||
}
|
||||
if strings.Contains(a, "gzip") {
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
|
|
@ -1,7 +1,7 @@
|
|||
syntax = "proto3";
|
||||
|
||||
package micro.server.http.v3.handler.meter;
|
||||
option go_package = "go.unistack.org/micro-server-http/v3/handler/meter;meter";
|
||||
option go_package = "go.unistack.org/micro-server-http/v3/handler/meter;meter_handler";
|
||||
|
||||
import "api/annotations.proto";
|
||||
import "openapiv3/annotations.proto";
|
||||
|
|
|
@ -1,31 +1,24 @@
|
|||
// Code generated by protoc-gen-go-micro. DO NOT EDIT.
|
||||
// versions:
|
||||
// - protoc-gen-go-micro v3.10.2
|
||||
// - protoc v3.21.12
|
||||
// source: meter.proto
|
||||
// - protoc-gen-go-micro v3.10.4
|
||||
// - protoc v5.26.1
|
||||
// source: meter/meter.proto
|
||||
|
||||
package meter
|
||||
package meter_handler
|
||||
|
||||
import (
|
||||
context "context"
|
||||
codec "go.unistack.org/micro-proto/v3/codec"
|
||||
v3 "go.unistack.org/micro-server-http/v3"
|
||||
client "go.unistack.org/micro/v3/client"
|
||||
)
|
||||
|
||||
var (
|
||||
MeterServiceName = "MeterService"
|
||||
)
|
||||
var (
|
||||
MeterServiceServerEndpoints = []v3.EndpointMetadata{
|
||||
{
|
||||
Name: "MeterService.Metrics",
|
||||
Path: "/metrics",
|
||||
Method: "GET",
|
||||
Body: "",
|
||||
Stream: false,
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
type MeterServiceClient interface {
|
||||
Metrics(ctx context.Context, req *codec.Frame, opts ...client.CallOption) (*codec.Frame, error)
|
||||
}
|
||||
|
||||
type MeterServiceServer interface {
|
||||
Metrics(ctx context.Context, req *codec.Frame, rsp *codec.Frame) error
|
||||
|
|
|
@ -1,16 +1,58 @@
|
|||
// Code generated by protoc-gen-go-micro. DO NOT EDIT.
|
||||
// protoc-gen-go-micro version: v3.10.2
|
||||
// source: meter.proto
|
||||
// protoc-gen-go-micro version: v3.10.4
|
||||
// source: meter/meter.proto
|
||||
|
||||
package meter
|
||||
package meter_handler
|
||||
|
||||
import (
|
||||
context "context"
|
||||
v31 "go.unistack.org/micro-client-http/v3"
|
||||
codec "go.unistack.org/micro-proto/v3/codec"
|
||||
v3 "go.unistack.org/micro-server-http/v3"
|
||||
client "go.unistack.org/micro/v3/client"
|
||||
server "go.unistack.org/micro/v3/server"
|
||||
http "net/http"
|
||||
)
|
||||
|
||||
var (
|
||||
MeterServiceServerEndpoints = []v3.EndpointMetadata{
|
||||
{
|
||||
Name: "MeterService.Metrics",
|
||||
Path: "/metrics",
|
||||
Method: "GET",
|
||||
Body: "",
|
||||
Stream: false,
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
type meterServiceClient struct {
|
||||
c client.Client
|
||||
name string
|
||||
}
|
||||
|
||||
func NewMeterServiceClient(name string, c client.Client) MeterServiceClient {
|
||||
return &meterServiceClient{c: c, name: name}
|
||||
}
|
||||
|
||||
func (c *meterServiceClient) Metrics(ctx context.Context, req *codec.Frame, opts ...client.CallOption) (*codec.Frame, error) {
|
||||
errmap := make(map[string]interface{}, 1)
|
||||
errmap["default"] = &codec.Frame{}
|
||||
opts = append(opts,
|
||||
v31.ErrorMap(errmap),
|
||||
)
|
||||
opts = append(opts,
|
||||
v31.Method(http.MethodGet),
|
||||
v31.Path("/metrics"),
|
||||
)
|
||||
rsp := &codec.Frame{}
|
||||
err := c.c.Call(ctx, c.c.NewRequest(c.name, "MeterService.Metrics", req), rsp, opts...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return rsp, nil
|
||||
}
|
||||
|
||||
type meterServiceServer struct {
|
||||
MeterServiceServer
|
||||
}
|
||||
|
|
|
@ -0,0 +1,46 @@
|
|||
package pprof_handler
|
||||
|
||||
import (
|
||||
"expvar"
|
||||
"net/http"
|
||||
"net/http/pprof"
|
||||
"path"
|
||||
"strings"
|
||||
)
|
||||
|
||||
func NewHandler(prefixPath string, initFuncs ...func()) http.HandlerFunc {
|
||||
for _, fn := range initFuncs {
|
||||
fn()
|
||||
}
|
||||
|
||||
return func(w http.ResponseWriter, r *http.Request) {
|
||||
switch {
|
||||
case strings.EqualFold(r.RequestURI, prefixPath) && r.RequestURI[len(r.RequestURI)-1] != '/':
|
||||
http.Redirect(w, r, r.RequestURI+"/", http.StatusMovedPermanently)
|
||||
case strings.HasPrefix(r.RequestURI, path.Join(prefixPath, "cmdline")):
|
||||
pprof.Cmdline(w, r)
|
||||
case strings.HasPrefix(r.RequestURI, path.Join(prefixPath, "profile")):
|
||||
pprof.Profile(w, r)
|
||||
case strings.HasPrefix(r.RequestURI, path.Join(prefixPath, "symbol")):
|
||||
pprof.Symbol(w, r)
|
||||
case strings.HasPrefix(r.RequestURI, path.Join(prefixPath, "trace")):
|
||||
pprof.Trace(w, r)
|
||||
case strings.HasPrefix(r.RequestURI, path.Join(prefixPath, "goroutine")):
|
||||
pprof.Handler("goroutine").ServeHTTP(w, r)
|
||||
case strings.HasPrefix(r.RequestURI, path.Join(prefixPath, "threadcreate")):
|
||||
pprof.Handler("threadcreate").ServeHTTP(w, r)
|
||||
case strings.HasPrefix(r.RequestURI, path.Join(prefixPath, "mutex")):
|
||||
pprof.Handler("mutex").ServeHTTP(w, r)
|
||||
case strings.HasPrefix(r.RequestURI, path.Join(prefixPath, "heap")):
|
||||
pprof.Handler("heap").ServeHTTP(w, r)
|
||||
case strings.HasPrefix(r.RequestURI, path.Join(prefixPath, "block")):
|
||||
pprof.Handler("block").ServeHTTP(w, r)
|
||||
case strings.HasPrefix(r.RequestURI, path.Join(prefixPath, "allocs")):
|
||||
pprof.Handler("allocs").ServeHTTP(w, r)
|
||||
case strings.HasPrefix(r.RequestURI, path.Join(prefixPath, "vars")):
|
||||
expvar.Handler().ServeHTTP(w, r)
|
||||
default:
|
||||
pprof.Index(w, r)
|
||||
}
|
||||
}
|
||||
}
|
Binary file not shown.
After Width: | Height: | Size: 665 B |
Binary file not shown.
After Width: | Height: | Size: 628 B |
|
@ -0,0 +1,16 @@
|
|||
html {
|
||||
box-sizing: border-box;
|
||||
overflow: -moz-scrollbars-vertical;
|
||||
overflow-y: scroll;
|
||||
}
|
||||
|
||||
*,
|
||||
*:before,
|
||||
*:after {
|
||||
box-sizing: inherit;
|
||||
}
|
||||
|
||||
body {
|
||||
margin: 0;
|
||||
background: #fafafa;
|
||||
}
|
|
@ -0,0 +1,19 @@
|
|||
<!-- HTML for static distribution bundle build -->
|
||||
<!DOCTYPE html>
|
||||
<html lang="en">
|
||||
<head>
|
||||
<meta charset="UTF-8">
|
||||
<title>Swagger UI</title>
|
||||
<link rel="stylesheet" type="text/css" href="./swagger-ui.css" />
|
||||
<link rel="stylesheet" type="text/css" href="index.css" />
|
||||
<link rel="icon" type="image/png" href="./favicon-32x32.png" sizes="32x32" />
|
||||
<link rel="icon" type="image/png" href="./favicon-16x16.png" sizes="16x16" />
|
||||
</head>
|
||||
|
||||
<body>
|
||||
<div id="swagger-ui"></div>
|
||||
<script src="./swagger-ui-bundle.js" charset="UTF-8"> </script>
|
||||
<script src="./swagger-ui-standalone-preset.js" charset="UTF-8"> </script>
|
||||
<script src="./swagger-initializer.js" charset="UTF-8"> </script>
|
||||
</body>
|
||||
</html>
|
|
@ -0,0 +1,79 @@
|
|||
<!doctype html>
|
||||
<html lang="en-US">
|
||||
<head>
|
||||
<title>Swagger UI: OAuth2 Redirect</title>
|
||||
</head>
|
||||
<body>
|
||||
<script>
|
||||
'use strict';
|
||||
function run () {
|
||||
var oauth2 = window.opener.swaggerUIRedirectOauth2;
|
||||
var sentState = oauth2.state;
|
||||
var redirectUrl = oauth2.redirectUrl;
|
||||
var isValid, qp, arr;
|
||||
|
||||
if (/code|token|error/.test(window.location.hash)) {
|
||||
qp = window.location.hash.substring(1).replace('?', '&');
|
||||
} else {
|
||||
qp = location.search.substring(1);
|
||||
}
|
||||
|
||||
arr = qp.split("&");
|
||||
arr.forEach(function (v,i,_arr) { _arr[i] = '"' + v.replace('=', '":"') + '"';});
|
||||
qp = qp ? JSON.parse('{' + arr.join() + '}',
|
||||
function (key, value) {
|
||||
return key === "" ? value : decodeURIComponent(value);
|
||||
}
|
||||
) : {};
|
||||
|
||||
isValid = qp.state === sentState;
|
||||
|
||||
if ((
|
||||
oauth2.auth.schema.get("flow") === "accessCode" ||
|
||||
oauth2.auth.schema.get("flow") === "authorizationCode" ||
|
||||
oauth2.auth.schema.get("flow") === "authorization_code"
|
||||
) && !oauth2.auth.code) {
|
||||
if (!isValid) {
|
||||
oauth2.errCb({
|
||||
authId: oauth2.auth.name,
|
||||
source: "auth",
|
||||
level: "warning",
|
||||
message: "Authorization may be unsafe, passed state was changed in server. The passed state wasn't returned from auth server."
|
||||
});
|
||||
}
|
||||
|
||||
if (qp.code) {
|
||||
delete oauth2.state;
|
||||
oauth2.auth.code = qp.code;
|
||||
oauth2.callback({auth: oauth2.auth, redirectUrl: redirectUrl});
|
||||
} else {
|
||||
let oauthErrorMsg;
|
||||
if (qp.error) {
|
||||
oauthErrorMsg = "["+qp.error+"]: " +
|
||||
(qp.error_description ? qp.error_description+ ". " : "no accessCode received from the server. ") +
|
||||
(qp.error_uri ? "More info: "+qp.error_uri : "");
|
||||
}
|
||||
|
||||
oauth2.errCb({
|
||||
authId: oauth2.auth.name,
|
||||
source: "auth",
|
||||
level: "error",
|
||||
message: oauthErrorMsg || "[Authorization failed]: no accessCode received from the server."
|
||||
});
|
||||
}
|
||||
} else {
|
||||
oauth2.callback({auth: oauth2.auth, token: qp, isValid: isValid, redirectUrl: redirectUrl});
|
||||
}
|
||||
window.close();
|
||||
}
|
||||
|
||||
if (document.readyState !== 'loading') {
|
||||
run();
|
||||
} else {
|
||||
document.addEventListener('DOMContentLoaded', function () {
|
||||
run();
|
||||
});
|
||||
}
|
||||
</script>
|
||||
</body>
|
||||
</html>
|
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
|
@ -0,0 +1,142 @@
|
|||
package swaggerui_handler
|
||||
|
||||
import (
|
||||
"embed"
|
||||
"html/template"
|
||||
"net/http"
|
||||
"path"
|
||||
"reflect"
|
||||
)
|
||||
|
||||
//go:embed *.js *.css *.html *.png
|
||||
var assets embed.FS
|
||||
|
||||
var (
|
||||
Handler = func(prefix string) http.HandlerFunc {
|
||||
return func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != http.MethodGet || path.Base(r.URL.Path) != "swagger-initializer.js" {
|
||||
http.StripPrefix(prefix, http.FileServer(http.FS(assets))).ServeHTTP(w, r)
|
||||
return
|
||||
}
|
||||
tpl := template.New("swagger-initializer.js").Funcs(TemplateFuncs)
|
||||
ptpl, err := tpl.Parse(Template)
|
||||
if err != nil {
|
||||
w.WriteHeader(http.StatusInternalServerError)
|
||||
_, _ = w.Write([]byte(err.Error()))
|
||||
return
|
||||
}
|
||||
if err := ptpl.Execute(w, Config); err != nil {
|
||||
w.WriteHeader(http.StatusInternalServerError)
|
||||
_, _ = w.Write([]byte(err.Error()))
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
TemplateFuncs = template.FuncMap{
|
||||
"isInt": func(i interface{}) bool {
|
||||
v := reflect.ValueOf(i)
|
||||
switch v.Kind() {
|
||||
case reflect.Int, reflect.Int8, reflect.Int32, reflect.Int64, reflect.Uint, reflect.Uint8, reflect.Uint32, reflect.Uint64, reflect.Float32, reflect.Float64:
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
},
|
||||
"isBool": func(i interface{}) bool {
|
||||
v := reflect.ValueOf(i)
|
||||
switch v.Kind() {
|
||||
case reflect.Bool:
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
},
|
||||
"isString": func(i interface{}) bool {
|
||||
v := reflect.ValueOf(i)
|
||||
switch v.Kind() {
|
||||
case reflect.String:
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
},
|
||||
"isSlice": func(i interface{}) bool {
|
||||
v := reflect.ValueOf(i)
|
||||
switch v.Kind() {
|
||||
case reflect.Slice:
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
},
|
||||
"isMap": func(i interface{}) bool {
|
||||
v := reflect.ValueOf(i)
|
||||
switch v.Kind() {
|
||||
case reflect.Map:
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
},
|
||||
}
|
||||
Template = `
|
||||
window.onload = function() {
|
||||
//<editor-fold desc="Changeable Configuration Block">
|
||||
|
||||
window.ui = SwaggerUIBundle({
|
||||
{{- range $k, $v := . }}
|
||||
{{- if (eq (printf "%s" $v) "") -}}
|
||||
{{- continue -}}
|
||||
{{ end }}
|
||||
{{ $k }}: {{ if isBool $v -}}
|
||||
{{- $v -}},
|
||||
{{- else if isInt $v -}}
|
||||
{{- $v -}},
|
||||
{{- else if isString $v -}}
|
||||
"{{- $v -}}",
|
||||
{{- else if and (isSlice $v) (or (eq (printf "%s" $k) "presets") (eq (printf "%s" $k) "plugins")) -}}
|
||||
[
|
||||
{{- range $v }}
|
||||
{{ . }},
|
||||
{{- end }}
|
||||
],
|
||||
{{- end -}}
|
||||
{{ end }}
|
||||
});
|
||||
|
||||
//</editor-fold>
|
||||
};`
|
||||
Config = map[string]interface{}{
|
||||
"configUrl": "",
|
||||
"dom_id": "#swagger-ui",
|
||||
/*
|
||||
"domNode": "",
|
||||
"spec": "",
|
||||
"urls": []interface{}{
|
||||
map[string]interface{}{
|
||||
"url": "",
|
||||
"name": "",
|
||||
},
|
||||
},
|
||||
},
|
||||
*/
|
||||
"url": "https://petstore.swagger.io/v2/swagger.json",
|
||||
"deepLinking": true,
|
||||
"displayOperationId": false,
|
||||
"defaultModelsExpandDepth": 1,
|
||||
"defaultModelExpandDepth": 1,
|
||||
"displayRequestDuration": true,
|
||||
"filter": true,
|
||||
"operationsSorter": "alpha",
|
||||
"showExtensions": true,
|
||||
"tryItOutEnabled": true,
|
||||
"presets": []string{
|
||||
"SwaggerUIBundle.presets.apis",
|
||||
"SwaggerUIStandalonePreset",
|
||||
},
|
||||
"plugins": []string{
|
||||
"SwaggerUIBundle.plugins.DownloadUrl",
|
||||
},
|
||||
"layout": "StandaloneLayout",
|
||||
}
|
||||
)
|
|
@ -0,0 +1,15 @@
|
|||
package swaggerui_handler
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestTemplate(t *testing.T) {
|
||||
t.Skip()
|
||||
h := http.NewServeMux()
|
||||
h.HandleFunc("/", Handler(""))
|
||||
if err := http.ListenAndServe(":8080", h); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
|
@ -0,0 +1,61 @@
|
|||
package swagger_handler
|
||||
|
||||
import (
|
||||
"io/fs"
|
||||
"net/http"
|
||||
|
||||
yamlcodec "go.unistack.org/micro-codec-yaml/v3"
|
||||
rutil "go.unistack.org/micro/v3/util/reflect"
|
||||
)
|
||||
|
||||
// Handler append to generated swagger data from dst map[string]interface{}
|
||||
var Handler = func(dst map[string]interface{}, fsys fs.FS) http.HandlerFunc {
|
||||
c := yamlcodec.NewCodec()
|
||||
return func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != http.MethodGet {
|
||||
w.WriteHeader(http.StatusNotFound)
|
||||
return
|
||||
}
|
||||
|
||||
path := r.URL.Path
|
||||
if len(path) > 1 && path[0] == '/' {
|
||||
path = path[1:]
|
||||
}
|
||||
|
||||
buf, err := fs.ReadFile(fsys, path)
|
||||
if err != nil {
|
||||
w.WriteHeader(http.StatusInternalServerError)
|
||||
_, _ = w.Write([]byte(err.Error()))
|
||||
return
|
||||
}
|
||||
|
||||
if dst == nil {
|
||||
w.WriteHeader(http.StatusOK)
|
||||
_, _ = w.Write(buf)
|
||||
return
|
||||
}
|
||||
|
||||
var src interface{}
|
||||
|
||||
if err = c.Unmarshal(buf, src); err != nil {
|
||||
w.WriteHeader(http.StatusInternalServerError)
|
||||
_, _ = w.Write([]byte(err.Error()))
|
||||
return
|
||||
}
|
||||
|
||||
if err = rutil.Merge(src, dst); err != nil {
|
||||
w.WriteHeader(http.StatusInternalServerError)
|
||||
_, _ = w.Write([]byte(err.Error()))
|
||||
return
|
||||
}
|
||||
|
||||
if buf, err = c.Marshal(src); err != nil {
|
||||
w.WriteHeader(http.StatusInternalServerError)
|
||||
_, _ = w.Write([]byte(err.Error()))
|
||||
return
|
||||
}
|
||||
|
||||
w.WriteHeader(http.StatusOK)
|
||||
_, _ = w.Write(buf)
|
||||
}
|
||||
}
|
167
http.go
167
http.go
|
@ -23,9 +23,9 @@ import (
|
|||
"golang.org/x/net/netutil"
|
||||
)
|
||||
|
||||
var _ server.Server = &httpServer{}
|
||||
var _ server.Server = (*Server)(nil)
|
||||
|
||||
type httpServer struct {
|
||||
type Server struct {
|
||||
hd server.Handler
|
||||
rsvc *register.Service
|
||||
handlers map[string]server.Handler
|
||||
|
@ -40,7 +40,7 @@ type httpServer struct {
|
|||
init bool
|
||||
}
|
||||
|
||||
func (h *httpServer) newCodec(ct string) (codec.Codec, error) {
|
||||
func (h *Server) newCodec(ct string) (codec.Codec, error) {
|
||||
if idx := strings.IndexRune(ct, ';'); idx >= 0 {
|
||||
ct = ct[:idx]
|
||||
}
|
||||
|
@ -53,14 +53,14 @@ func (h *httpServer) newCodec(ct string) (codec.Codec, error) {
|
|||
return nil, codec.ErrUnknownContentType
|
||||
}
|
||||
|
||||
func (h *httpServer) Options() server.Options {
|
||||
func (h *Server) Options() server.Options {
|
||||
h.Lock()
|
||||
opts := h.opts
|
||||
h.Unlock()
|
||||
return opts
|
||||
}
|
||||
|
||||
func (h *httpServer) Init(opts ...server.Option) error {
|
||||
func (h *Server) Init(opts ...server.Option) error {
|
||||
if len(opts) == 0 && h.init {
|
||||
return nil
|
||||
}
|
||||
|
@ -130,7 +130,7 @@ func (h *httpServer) Init(opts ...server.Option) error {
|
|||
return nil
|
||||
}
|
||||
|
||||
func (h *httpServer) Handle(handler server.Handler) error {
|
||||
func (h *Server) Handle(handler server.Handler) error {
|
||||
// passed unknown handler
|
||||
hdlr, ok := handler.(*httpHandler)
|
||||
if !ok {
|
||||
|
@ -159,7 +159,7 @@ func (h *httpServer) Handle(handler server.Handler) error {
|
|||
return nil
|
||||
}
|
||||
|
||||
func (h *httpServer) NewHandler(handler interface{}, opts ...server.HandlerOption) server.Handler {
|
||||
func (h *Server) NewHandler(handler interface{}, opts ...server.HandlerOption) server.Handler {
|
||||
options := server.NewHandlerOptions(opts...)
|
||||
|
||||
eps := make([]*register.Endpoint, 0, len(options.Metadata))
|
||||
|
@ -191,6 +191,11 @@ func (h *httpServer) NewHandler(handler interface{}, opts ...server.HandlerOptio
|
|||
}
|
||||
*/
|
||||
|
||||
registerCORS := false
|
||||
if v, ok := options.Context.Value(registerCORSHandlerKey{}).(bool); ok && v {
|
||||
registerCORS = true
|
||||
}
|
||||
|
||||
for hn, md := range options.Metadata {
|
||||
var method reflect.Method
|
||||
mname := hn[strings.Index(hn, ".")+1:]
|
||||
|
@ -204,7 +209,7 @@ func (h *httpServer) NewHandler(handler interface{}, opts ...server.HandlerOptio
|
|||
}
|
||||
|
||||
if method.Name == "" && h.opts.Logger.V(logger.ErrorLevel) {
|
||||
h.opts.Logger.Errorf(h.opts.Context, "nil method for %s", mname)
|
||||
h.opts.Logger.Error(h.opts.Context, fmt.Sprintf("nil method for %s", mname))
|
||||
continue
|
||||
}
|
||||
|
||||
|
@ -213,7 +218,7 @@ func (h *httpServer) NewHandler(handler interface{}, opts ...server.HandlerOptio
|
|||
h.opts.Logger.Errorf(h.opts.Context, "%v", err)
|
||||
continue
|
||||
} else if mtype == nil {
|
||||
h.opts.Logger.Errorf(h.opts.Context, "nil mtype for %s", mname)
|
||||
h.opts.Logger.Error(h.opts.Context, fmt.Sprintf("nil mtype for %s", mname))
|
||||
continue
|
||||
}
|
||||
|
||||
|
@ -223,14 +228,23 @@ func (h *httpServer) NewHandler(handler interface{}, opts ...server.HandlerOptio
|
|||
pth := &patHandler{mtype: mtype, name: name, rcvr: rcvr}
|
||||
hdlr.name = name
|
||||
|
||||
if err := hdlr.handlers.Insert([]string{md["Method"]}, md["Path"], pth); err != nil {
|
||||
h.opts.Logger.Errorf(h.opts.Context, "cant add handler for %s %s", md["Method"], md["Path"])
|
||||
methods := []string{md["Method"]}
|
||||
if registerCORS {
|
||||
methods = append(methods, http.MethodOptions)
|
||||
}
|
||||
|
||||
if err := hdlr.handlers.Insert(methods, md["Path"], pth); err != nil {
|
||||
h.opts.Logger.Error(h.opts.Context, fmt.Sprintf("cant add handler for %v %s", methods, md["Path"]))
|
||||
}
|
||||
|
||||
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)
|
||||
methods := []string{http.MethodPost}
|
||||
if registerCORS {
|
||||
methods = append(methods, http.MethodOptions)
|
||||
}
|
||||
|
||||
if err := hdlr.handlers.Insert(methods, "/"+hn, pth); err != nil {
|
||||
h.opts.Logger.Error(h.opts.Context, fmt.Sprintf("cant add rpc handler for http.MethodPost %s /%s", hn, hn))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
@ -254,7 +268,7 @@ func (h *httpServer) NewHandler(handler interface{}, opts ...server.HandlerOptio
|
|||
}
|
||||
|
||||
if method.Name == "" && h.opts.Logger.V(logger.ErrorLevel) {
|
||||
h.opts.Logger.Errorf(h.opts.Context, "nil method for %s", mname)
|
||||
h.opts.Logger.Error(h.opts.Context, fmt.Sprintf("nil method for %s", mname))
|
||||
continue
|
||||
}
|
||||
|
||||
|
@ -263,7 +277,7 @@ func (h *httpServer) NewHandler(handler interface{}, opts ...server.HandlerOptio
|
|||
h.opts.Logger.Errorf(h.opts.Context, "%v", err)
|
||||
continue
|
||||
} else if mtype == nil {
|
||||
h.opts.Logger.Errorf(h.opts.Context, "nil mtype for %s", mname)
|
||||
h.opts.Logger.Error(h.opts.Context, fmt.Sprintf("nil mtype for %s", mname))
|
||||
continue
|
||||
}
|
||||
|
||||
|
@ -273,14 +287,24 @@ func (h *httpServer) NewHandler(handler interface{}, opts ...server.HandlerOptio
|
|||
pth := &patHandler{mtype: mtype, name: name, rcvr: rcvr}
|
||||
hdlr.name = name
|
||||
|
||||
if err := hdlr.handlers.Insert([]string{md.Method}, md.Path, pth); err != nil {
|
||||
h.opts.Logger.Errorf(h.opts.Context, "cant add handler for %s %s", md.Method, md.Path)
|
||||
methods := []string{md.Method}
|
||||
if registerCORS {
|
||||
methods = append(methods, http.MethodOptions)
|
||||
}
|
||||
|
||||
if err := hdlr.handlers.Insert(methods, md.Path, pth); err != nil {
|
||||
h.opts.Logger.Error(h.opts.Context, fmt.Sprintf("cant add handler for %s %s", md.Method, md.Path))
|
||||
}
|
||||
|
||||
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)
|
||||
methods := []string{http.MethodPost}
|
||||
if registerCORS {
|
||||
methods = append(methods, http.MethodOptions)
|
||||
}
|
||||
|
||||
h.opts.Logger.Info(h.opts.Context, fmt.Sprintf("register rpc handler for http.MethodPost %s /%s", hn, hn))
|
||||
if err := hdlr.handlers.Insert(methods, "/"+hn, pth); err != nil {
|
||||
h.opts.Logger.Error(h.opts.Context, fmt.Sprintf("cant add rpc handler for http.MethodPost %s /%s", hn, hn))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
@ -288,11 +312,11 @@ func (h *httpServer) NewHandler(handler interface{}, opts ...server.HandlerOptio
|
|||
return hdlr
|
||||
}
|
||||
|
||||
func (h *httpServer) NewSubscriber(topic string, handler interface{}, opts ...server.SubscriberOption) server.Subscriber {
|
||||
func (h *Server) NewSubscriber(topic string, handler interface{}, opts ...server.SubscriberOption) server.Subscriber {
|
||||
return newSubscriber(topic, handler, opts...)
|
||||
}
|
||||
|
||||
func (h *httpServer) Subscribe(sb server.Subscriber) error {
|
||||
func (h *Server) Subscribe(sb server.Subscriber) error {
|
||||
sub, ok := sb.(*httpSubscriber)
|
||||
if !ok {
|
||||
return fmt.Errorf("invalid subscriber: expected *httpSubscriber")
|
||||
|
@ -317,7 +341,7 @@ func (h *httpServer) Subscribe(sb server.Subscriber) error {
|
|||
return nil
|
||||
}
|
||||
|
||||
func (h *httpServer) Register() error {
|
||||
func (h *Server) Register() error {
|
||||
var eps []*register.Endpoint
|
||||
h.RLock()
|
||||
for _, hdlr := range h.handlers {
|
||||
|
@ -363,7 +387,7 @@ func (h *httpServer) Register() error {
|
|||
|
||||
if !registered {
|
||||
if config.Logger.V(logger.InfoLevel) {
|
||||
config.Logger.Infof(config.Context, "Register [%s] Registering node: %s", config.Register.String(), service.Nodes[0].ID)
|
||||
config.Logger.Info(config.Context, fmt.Sprintf("Register [%s] Registering node: %s", config.Register.String(), service.Nodes[0].ID))
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -378,6 +402,17 @@ func (h *httpServer) Register() error {
|
|||
}
|
||||
|
||||
h.Lock()
|
||||
|
||||
h.registered = true
|
||||
h.rsvc = service
|
||||
h.Unlock()
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (h *Server) subscribe() error {
|
||||
config := h.opts
|
||||
|
||||
for sb := range h.subscribers {
|
||||
handler := h.createSubHandler(sb, config)
|
||||
var opts []broker.SubscribeOption
|
||||
|
@ -401,14 +436,10 @@ func (h *httpServer) Register() error {
|
|||
h.subscribers[sb] = []broker.Subscriber{sub}
|
||||
}
|
||||
|
||||
h.registered = true
|
||||
h.rsvc = service
|
||||
h.Unlock()
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (h *httpServer) Deregister() error {
|
||||
func (h *Server) Deregister() error {
|
||||
h.RLock()
|
||||
config := h.opts
|
||||
h.RUnlock()
|
||||
|
@ -419,7 +450,7 @@ func (h *httpServer) Deregister() error {
|
|||
}
|
||||
|
||||
if config.Logger.V(logger.InfoLevel) {
|
||||
config.Logger.Infof(config.Context, "Deregistering node: %s", service.Nodes[0].ID)
|
||||
config.Logger.Info(config.Context, "Deregistering node: "+service.Nodes[0].ID)
|
||||
}
|
||||
|
||||
if err := server.DefaultDeregisterFunc(service, config); err != nil {
|
||||
|
@ -443,10 +474,10 @@ func (h *httpServer) Deregister() error {
|
|||
}
|
||||
|
||||
for _, sub := range subs {
|
||||
config.Logger.Infof(config.Context, "Unsubscribing from topic: %s", sub.Topic())
|
||||
config.Logger.Info(config.Context, "Unsubscribing from topic: "+sub.Topic())
|
||||
if err := sub.Unsubscribe(subCtx); err != nil {
|
||||
h.Unlock()
|
||||
config.Logger.Errorf(config.Context, "failed to unsubscribe topic: %s, error: %v", sb.Topic(), err)
|
||||
config.Logger.Error(config.Context, fmt.Sprintf("failed to unsubscribe topic: %s, error", sb.Topic()), err)
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
@ -456,7 +487,7 @@ func (h *httpServer) Deregister() error {
|
|||
return nil
|
||||
}
|
||||
|
||||
func (h *httpServer) Start() error {
|
||||
func (h *Server) Start() error {
|
||||
h.RLock()
|
||||
config := h.opts
|
||||
h.RUnlock()
|
||||
|
@ -486,7 +517,7 @@ func (h *httpServer) Start() error {
|
|||
}
|
||||
|
||||
if config.Logger.V(logger.InfoLevel) {
|
||||
config.Logger.Infof(config.Context, "Listening on %s", ts.Addr().String())
|
||||
config.Logger.Info(config.Context, "Listening on "+ts.Addr().String())
|
||||
}
|
||||
|
||||
h.Lock()
|
||||
|
@ -494,7 +525,6 @@ func (h *httpServer) Start() error {
|
|||
h.Unlock()
|
||||
|
||||
var handler http.Handler
|
||||
var srvFunc func(net.Listener) error
|
||||
|
||||
// nolint: nestif
|
||||
if h.opts.Context != nil {
|
||||
|
@ -531,7 +561,7 @@ func (h *httpServer) Start() error {
|
|||
|
||||
if err := config.RegisterCheck(h.opts.Context); err != nil {
|
||||
if config.Logger.V(logger.ErrorLevel) {
|
||||
config.Logger.Errorf(config.Context, "Server %s-%s register check error: %s", config.Name, config.ID, err)
|
||||
config.Logger.Error(config.Context, fmt.Sprintf("Server %s-%s register check error", config.Name, config.ID), err)
|
||||
}
|
||||
} else {
|
||||
if err = h.Register(); err != nil {
|
||||
|
@ -539,8 +569,13 @@ func (h *httpServer) Start() error {
|
|||
}
|
||||
}
|
||||
|
||||
if err := h.subscribe(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
fn := handler
|
||||
|
||||
var hs *http.Server
|
||||
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
|
||||
|
@ -548,25 +583,19 @@ func (h *httpServer) Start() error {
|
|||
fn = mwf[i-1](fn)
|
||||
}
|
||||
}
|
||||
if hs, ok := h.opts.Context.Value(serverKey{}).(*http.Server); ok && hs != nil {
|
||||
var ok bool
|
||||
if hs, ok = h.opts.Context.Value(serverKey{}).(*http.Server); ok && hs != nil {
|
||||
hs.Handler = fn
|
||||
srvFunc = hs.Serve
|
||||
} else {
|
||||
hs = &http.Server{Handler: fn}
|
||||
}
|
||||
}
|
||||
|
||||
if srvFunc != nil {
|
||||
go func() {
|
||||
if cerr := srvFunc(ts); cerr != nil && !errors.Is(cerr, net.ErrClosed) {
|
||||
h.opts.Logger.Error(h.opts.Context, cerr)
|
||||
}
|
||||
}()
|
||||
} else {
|
||||
go func() {
|
||||
if cerr := http.Serve(ts, fn); cerr != nil && !errors.Is(cerr, net.ErrClosed) {
|
||||
h.opts.Logger.Error(h.opts.Context, cerr)
|
||||
}
|
||||
}()
|
||||
}
|
||||
go func() {
|
||||
if cerr := hs.Serve(ts); cerr != nil && !errors.Is(cerr, net.ErrClosed) {
|
||||
h.opts.Logger.Error(h.opts.Context, "serve error", cerr)
|
||||
}
|
||||
}()
|
||||
|
||||
go func() {
|
||||
t := new(time.Ticker)
|
||||
|
@ -592,28 +621,28 @@ func (h *httpServer) Start() error {
|
|||
// nolint: nestif
|
||||
if rerr != nil && registered {
|
||||
if config.Logger.V(logger.ErrorLevel) {
|
||||
config.Logger.Errorf(config.Context, "Server %s-%s register check error: %s, deregister it", config.Name, config.ID, rerr)
|
||||
config.Logger.Error(config.Context, fmt.Sprintf("Server %s-%s register check error, deregister it", config.Name, config.ID), rerr)
|
||||
}
|
||||
// deregister self in case of error
|
||||
if err := h.Deregister(); err != nil {
|
||||
if config.Logger.V(logger.ErrorLevel) {
|
||||
config.Logger.Errorf(config.Context, "Server %s-%s deregister error: %s", config.Name, config.ID, err)
|
||||
config.Logger.Error(config.Context, fmt.Sprintf("Server %s-%s deregister error", config.Name, config.ID), err)
|
||||
}
|
||||
}
|
||||
} else if rerr != nil && !registered {
|
||||
if config.Logger.V(logger.ErrorLevel) {
|
||||
config.Logger.Errorf(config.Context, "Server %s-%s register check error: %s", config.Name, config.ID, rerr)
|
||||
config.Logger.Error(config.Context, fmt.Sprintf("Server %s-%s register check error", config.Name, config.ID), rerr)
|
||||
}
|
||||
continue
|
||||
}
|
||||
if err := h.Register(); err != nil {
|
||||
if config.Logger.V(logger.ErrorLevel) {
|
||||
config.Logger.Errorf(config.Context, "Server %s-%s register error: %s", config.Name, config.ID, err)
|
||||
config.Logger.Error(config.Context, fmt.Sprintf("Server %s-%s register error", config.Name, config.ID), err)
|
||||
}
|
||||
}
|
||||
|
||||
if err := h.Register(); err != nil {
|
||||
config.Logger.Errorf(config.Context, "Server register error: %s", err)
|
||||
config.Logger.Error(config.Context, "Server register error", err)
|
||||
}
|
||||
// wait for exit
|
||||
case ch = <-h.exit:
|
||||
|
@ -623,40 +652,48 @@ func (h *httpServer) Start() error {
|
|||
|
||||
// deregister
|
||||
if err := h.Deregister(); err != nil {
|
||||
config.Logger.Errorf(config.Context, "Server deregister error: %s", err)
|
||||
config.Logger.Error(config.Context, "Server deregister error", err)
|
||||
}
|
||||
|
||||
if err := config.Broker.Disconnect(config.Context); err != nil {
|
||||
config.Logger.Errorf(config.Context, "Broker disconnect error: %s", err)
|
||||
config.Logger.Error(config.Context, "Broker disconnect error", err)
|
||||
}
|
||||
|
||||
ch <- ts.Close()
|
||||
ctx, cancel := context.WithTimeout(context.Background(), h.opts.GracefulTimeout)
|
||||
defer cancel()
|
||||
|
||||
err := hs.Shutdown(ctx)
|
||||
if err != nil {
|
||||
err = hs.Close()
|
||||
}
|
||||
|
||||
ch <- err
|
||||
}()
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (h *httpServer) Stop() error {
|
||||
func (h *Server) Stop() error {
|
||||
ch := make(chan error)
|
||||
h.exit <- ch
|
||||
return <-ch
|
||||
}
|
||||
|
||||
func (h *httpServer) String() string {
|
||||
func (h *Server) String() string {
|
||||
return "http"
|
||||
}
|
||||
|
||||
func (h *httpServer) Name() string {
|
||||
func (h *Server) Name() string {
|
||||
return h.opts.Name
|
||||
}
|
||||
|
||||
func NewServer(opts ...server.Option) *httpServer {
|
||||
func NewServer(opts ...server.Option) *Server {
|
||||
options := server.NewOptions(opts...)
|
||||
eh := DefaultErrorHandler
|
||||
if v, ok := options.Context.Value(errorHandlerKey{}).(errorHandler); ok && v != nil {
|
||||
eh = v
|
||||
}
|
||||
return &httpServer{
|
||||
return &Server{
|
||||
opts: options,
|
||||
exit: make(chan chan error),
|
||||
subscribers: make(map[*httpSubscriber][]broker.Subscriber),
|
||||
|
|
|
@ -11,7 +11,6 @@ type httpMessage struct {
|
|||
header metadata.Metadata
|
||||
topic string
|
||||
contentType string
|
||||
body []byte
|
||||
}
|
||||
|
||||
func (r *httpMessage) Topic() string {
|
||||
|
|
13
options.go
13
options.go
|
@ -69,7 +69,7 @@ func getRspHeader(ctx context.Context) http.Header {
|
|||
|
||||
// GetRspCode used internally by generated http server handler
|
||||
func GetRspCode(ctx context.Context) int {
|
||||
var code int
|
||||
code := int(200)
|
||||
if rsp, ok := ctx.Value(rspCodeKey{}).(*rspCodeVal); ok {
|
||||
code = rsp.code
|
||||
}
|
||||
|
@ -85,8 +85,8 @@ func Middleware(mw ...func(http.Handler) http.Handler) server.Option {
|
|||
|
||||
type serverKey struct{}
|
||||
|
||||
// Server provide ability to pass *http.Server
|
||||
func Server(hs *http.Server) server.Option {
|
||||
// HTTPServer provide ability to pass *http.Server
|
||||
func HTTPServer(hs *http.Server) server.Option {
|
||||
return server.SetOption(serverKey{}, hs)
|
||||
}
|
||||
|
||||
|
@ -133,6 +133,13 @@ func RegisterRPCHandler(b bool) server.Option {
|
|||
return server.SetOption(registerRPCHandlerKey{}, b)
|
||||
}
|
||||
|
||||
type registerCORSHandlerKey struct{}
|
||||
|
||||
// RegisterCORSHandler registers cors endpoints with /ServiceName.ServiceEndpoint method POPTIONSOST
|
||||
func RegisterCORSHandler(b bool) server.HandlerOption {
|
||||
return server.SetHandlerOption(registerCORSHandlerKey{}, b)
|
||||
}
|
||||
|
||||
type handlerEndpointsKey struct{}
|
||||
|
||||
type EndpointMetadata struct {
|
||||
|
|
|
@ -10,6 +10,7 @@ import (
|
|||
"go.unistack.org/micro/v3/broker"
|
||||
"go.unistack.org/micro/v3/codec"
|
||||
"go.unistack.org/micro/v3/metadata"
|
||||
"go.unistack.org/micro/v3/options"
|
||||
"go.unistack.org/micro/v3/register"
|
||||
"go.unistack.org/micro/v3/server"
|
||||
)
|
||||
|
@ -101,7 +102,7 @@ func newSubscriber(topic string, sub interface{}, opts ...server.SubscriberOptio
|
|||
}
|
||||
}
|
||||
|
||||
func (s *httpServer) createSubHandler(sb *httpSubscriber, opts server.Options) broker.Handler {
|
||||
func (s *Server) createSubHandler(sb *httpSubscriber, opts server.Options) broker.Handler {
|
||||
return func(p broker.Event) error {
|
||||
msg := p.Message()
|
||||
ct := msg.Header["Content-Type"]
|
||||
|
@ -111,7 +112,6 @@ func (s *httpServer) createSubHandler(sb *httpSubscriber, opts server.Options) b
|
|||
}
|
||||
|
||||
hdr := metadata.Copy(msg.Header)
|
||||
delete(hdr, "Content-Type")
|
||||
ctx := metadata.NewIncomingContext(context.Background(), hdr)
|
||||
|
||||
results := make(chan error, len(sb.handlers))
|
||||
|
@ -160,9 +160,11 @@ func (s *httpServer) createSubHandler(sb *httpSubscriber, opts server.Options) b
|
|||
return nil
|
||||
}
|
||||
|
||||
for i := len(opts.SubWrappers); i > 0; i-- {
|
||||
fn = opts.SubWrappers[i-1](fn)
|
||||
}
|
||||
opts.Hooks.EachNext(func(hook options.Hook) {
|
||||
if h, ok := hook.(server.HookSubHandler); ok {
|
||||
fn = h(fn)
|
||||
}
|
||||
})
|
||||
|
||||
go func() {
|
||||
results <- fn(ctx, &httpMessage{
|
||||
|
|
Loading…
Reference in New Issue