Compare commits

...

64 Commits
v3.10.10 ... v3

Author SHA1 Message Date
57a7ee9b50 Merge pull request '#348-v3 add check method in should be skipped' (#189) from kgorbunov/micro-server-http:#348-v3 into v3
Reviewed-on: #189
Reviewed-by: Василий Толстов <v.tolstov@unistack.org>
2024-09-20 16:52:49 +03:00
Gorbunov Kirill Andreevich
e1db2d8c1a #348 Add check should not be trace and meter handler 2024-09-20 16:05:48 +03:00
Gorbunov Kirill Andreevich
d47b434a53 Merge remote-tracking branch 'upstream/v3' into v3 2024-09-20 09:52:29 +03:00
249ef06c28 update to latest micro
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2024-09-17 12:47:19 +03:00
0838d2ab9b skip special endpoints
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2024-05-04 19:38:21 +03:00
c95efdbf53 fixup logs
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2024-04-24 18:25:31 +03:00
c81960af51 add metrics and tracing
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2024-04-23 22:21:27 +03:00
1c36c1685f regen
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2024-04-10 00:15:54 +03:00
e889545210 fixup logger
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2024-04-10 00:06:50 +03:00
526e414d60 flush_v3 (#188)
## Pull Request template
Please, go through these steps before clicking submit on this PR.

1. Give a descriptive title to your PR.
2. Provide a description of your changes.
3. Make sure you have some relevant tests.
4. Put `closes #XXXX` in your comment to auto-close the issue that your PR fixes (if applicable).

**PLEASE REMOVE THIS TEMPLATE BEFORE SUBMITTING**

Co-authored-by: Gorbunov Kirill Andreevich <kgorbunov@mtsbank.ru>
Reviewed-on: #188
Co-authored-by: Кирилл Горбунов <kirya_gorbunov_2015@mail.ru>
Co-committed-by: Кирилл Горбунов <kirya_gorbunov_2015@mail.ru>
2024-03-26 14:52:33 +03:00
db423dff58 fixup cors handling
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2024-03-20 12:27:14 +03:00
542d4cec00 improve cors
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2024-03-18 16:38:18 +03:00
0ecd1da0dc improve cors
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2024-03-18 16:25:52 +03:00
7c669c636d Merge pull request 'v3' (#3) from unistack-org/micro-server-http:v3 into v3
Reviewed-on: kgorbunov/micro-server-http#3
2024-03-18 16:11:24 +03:00
0a8755ecb7 fixup
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2024-03-18 15:57:08 +03:00
9c29d92d7f fixup cors path handler
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2024-03-18 15:45:43 +03:00
b871c1be38 add cors option
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2024-03-18 14:51:40 +03:00
74bb12e75e Merge pull request 'fix nil ptr' (#184) from devstigneev/micro-server-http:issue_179_v3 into v3
Reviewed-on: #184
2024-03-12 22:05:32 +03:00
49a95c183b Merge remote-tracking branch 'main/v3' into issue_179_v3
# Conflicts:
#	http.go
2024-03-12 16:48:51 +03:00
5e6bd93a6b fix nil ptr 2024-03-12 16:44:34 +03:00
9ef26caf40 fixup panic
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2024-03-12 00:50:08 +03:00
b3e58d2cb6 fixup handlers
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2024-03-12 00:03:51 +03:00
b89d9fdc5b fixup headers
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2024-03-11 23:30:37 +03:00
95dcdd6025 gzip for v3 #153 (#183)
Co-authored-by: Кирилл Горбунов <kirya_gorbunov_2015@mail.ru>
Co-committed-by: Кирилл Горбунов <kirya_gorbunov_2015@mail.ru>
2024-03-11 13:21:11 +03:00
abe5be3ddc Merge pull request 'prepare for v3' (#181) from devstigneev/micro-server-http:issue_179_v3 into v3
Reviewed-on: #181
2024-03-09 13:59:21 +03:00
c3e6cdd973 prepare for v3 2024-03-09 10:06:13 +03:00
76dcf3af67 fixup redirect
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2024-02-10 23:46:19 +03:00
3e30960694 handler/pprof: clean prefix
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2024-02-09 00:16:34 +03:00
1643393377 Merge pull request 'handler/pprof: initial import' (#178) from pprof into v3
Reviewed-on: #178
2024-02-08 23:49:17 +03:00
313ae201af handler/pprof: initial import
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2024-02-08 23:48:33 +03:00
069eaf4485 Merge pull request 'copy incoming content-type' (#177) from ct into v3
Reviewed-on: #177
2023-12-20 09:26:28 +03:00
fd670155aa copy incoming content-type
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2023-12-20 09:26:06 +03:00
0e3a199e16 Merge pull request 'fix swagger handler' (#172) from handler-swagger into v3
Reviewed-on: #172
2023-08-14 13:43:50 +03:00
a665b69706 fix swagger handler
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2023-08-14 13:43:19 +03:00
90b7b7ec1c fix swagger handler
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2023-08-14 13:42:51 +03:00
63ae848043 Merge pull request 'add tls headers' (#170) from tls_header into v3
Reviewed-on: #170
2023-08-03 10:43:47 +03:00
4d378bbd51 add tls headers
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2023-08-03 10:43:33 +03:00
60d165f907 add tls headers
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2023-08-03 10:42:54 +03:00
5a37de7d74 Merge pull request 'handler/swagger: initial import' (#169) from swagger into v3
Reviewed-on: #169
2023-06-18 17:29:16 +03:00
c7d8b6a3a4 handler/swagger: initial import
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2023-06-18 17:28:29 +03:00
cd3e2526b4 Merge pull request 'fix query param struct filling' (#167) from matchesfix into v3
Reviewed-on: #167
2023-05-29 12:27:16 +03:00
10ae1741fc fix query param struct filling
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2023-05-29 12:26:56 +03:00
ca6cea861f Merge pull request 'add scheme to metadata' (#165) from scheme-v3 into v3
Reviewed-on: #165
2023-05-19 23:25:59 +03:00
002d5c9c53 add scheme to metadata
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2023-05-19 23:25:22 +03:00
1d25e59860 Merge pull request 'move down path handler after specific handler' (#163) from path-handler-v3 into v3
Reviewed-on: #163
2023-05-19 23:04:16 +03:00
ce26e8bf63 move down path handler after specific handler
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2023-05-19 23:03:04 +03:00
8e09142b90 Merge pull request 'check subscribe errors' (#162) from subscribeerr into v3
Reviewed-on: #162
2023-05-13 16:09:47 +03:00
13bf2bbb1e check subscribe errors
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2023-05-13 16:09:21 +03:00
ba292901d7 fixup build
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2023-05-09 21:14:55 +03:00
6d1b6c7014 allow to expose some method via http.HandlerFunc
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2023-05-09 21:13:11 +03:00
1276f99159 export Server to allow to cast
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2023-05-09 21:12:47 +03:00
ee5cbbbcb6 Merge pull request 'issue-155: add swagger-ui handler' (#157) from swaggerui into v3
Reviewed-on: #157
2023-05-04 06:51:42 +03:00
ce0abd2de3 issue-155: add swagger-ui handler
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2023-05-04 06:48:09 +03:00
647f731f50 Merge pull request 'return method not allowed error' (#152) from methodNotAllowed into v3
Reviewed-on: #152
2023-03-09 08:42:50 +03:00
724ffb8c3a return method not allowed error
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2023-03-09 08:42:04 +03:00
25f306425b
Merge pull request #150 from unistack-org/dependabot/go_modules/go.unistack.org/micro/v3-3.10.14
Bump go.unistack.org/micro/v3 from 3.10.13 to 3.10.14
2023-02-27 21:04:56 +03:00
dependabot[bot]
dad2d8d5bd
Bump go.unistack.org/micro/v3 from 3.10.13 to 3.10.14
Bumps [go.unistack.org/micro/v3](https://github.com/unistack-org/micro) from 3.10.13 to 3.10.14.
- [Release notes](https://github.com/unistack-org/micro/releases)
- [Commits](https://github.com/unistack-org/micro/compare/v3.10.13...v3.10.14)

---
updated-dependencies:
- dependency-name: go.unistack.org/micro/v3
  dependency-type: direct:production
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <support@github.com>
2023-02-27 18:02:27 +00:00
ca148cd8fc
Merge pull request #149 from unistack-org/handlers
import from main repo
2023-02-26 14:54:47 +03:00
cd8b3e37f2 import from main repo
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2023-02-26 14:53:09 +03:00
c908274c8b import from main repo
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2023-02-26 14:51:34 +03:00
fd3793d50f
Merge pull request #148 from unistack-org/dependabot/go_modules/go.unistack.org/micro/v3-3.10.13
Bump go.unistack.org/micro/v3 from 3.10.12 to 3.10.13
2023-02-22 20:16:33 +03:00
dependabot[bot]
e67021895e
Bump go.unistack.org/micro/v3 from 3.10.12 to 3.10.13
Bumps [go.unistack.org/micro/v3](https://github.com/unistack-org/micro) from 3.10.12 to 3.10.13.
- [Release notes](https://github.com/unistack-org/micro/releases)
- [Commits](https://github.com/unistack-org/micro/compare/v3.10.12...v3.10.13)

---
updated-dependencies:
- dependency-name: go.unistack.org/micro/v3
  dependency-type: direct:production
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <support@github.com>
2023-02-22 17:14:50 +00:00
9f4dbbf296
Merge pull request #147 from unistack-org/handlerEndpoints
provide HandlerEndpoints
2023-02-21 23:58:00 +03:00
eea8ae605b provide HandlerEndpoints
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2023-02-21 23:56:07 +03:00
32 changed files with 3155 additions and 132 deletions

24
.gitignore vendored Normal file
View File

@ -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

23
go.mod
View File

@ -1,9 +1,24 @@
module go.unistack.org/micro-server-http/v3
go 1.16
go 1.21
toolchain go1.23.1
require (
go.unistack.org/micro/v3 v3.10.12
golang.org/x/net v0.7.0
gopkg.in/yaml.v3 v3.0.1 // indirect
go.unistack.org/micro-client-http/v3 v3.9.12
go.unistack.org/micro-codec-yaml/v3 v3.10.1
go.unistack.org/micro-proto/v3 v3.4.1
go.unistack.org/micro/v3 v3.10.88
golang.org/x/net v0.29.0
)
require (
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.25.0 // indirect
google.golang.org/genproto/googleapis/rpc v0.0.0-20240903143218-8af14fe29dc1 // indirect
google.golang.org/grpc v1.66.2 // indirect
google.golang.org/protobuf v1.34.2 // indirect
gopkg.in/yaml.v3 v3.0.1 // indirect
sigs.k8s.io/yaml v1.4.0 // indirect
)

1450
go.sum

File diff suppressed because it is too large Load Diff

View File

@ -6,14 +6,21 @@ 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/meter"
"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 +67,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 +367,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)
@ -117,7 +407,7 @@ func (h *httpServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
handler = hdlr
break
} else if err == rhttp.ErrMethodNotAllowed && !h.registerRPC {
h.errorHandler(ctx, nil, w, r, fmt.Errorf("not matching route found"), http.StatusNotFound)
h.errorHandler(ctx, nil, w, r, fmt.Errorf("not matching route found"), http.StatusMethodNotAllowed)
return
}
}
@ -143,16 +433,112 @@ 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)
if !slices.Contains(tracer.DefaultSkipEndpoints, h.hd.Name()) {
ctx, sp = h.opts.Tracer.Start(ctx, h.hd.Name()+" rpc-server",
tracer.WithSpanKind(tracer.SpanKindServer),
tracer.WithSpanLabels(
"endpoint", h.hd.Name(),
),
)
defer func() {
n := GetRspCode(ctx)
if s, _ := sp.Status(); s != tracer.SpanStatusError && n > 399 {
sp.SetStatus(tracer.SpanStatusError, http.StatusText(n))
}
sp.Finish()
}()
}
if !slices.Contains(meter.DefaultSkipEndpoints, h.hd.Name()) {
h.opts.Meter.Counter(semconv.ServerRequestInflight, "endpoint", h.hd.Name()).Inc()
defer func() {
n := GetRspCode(ctx)
if n > 399 {
h.opts.Meter.Counter(semconv.ServerRequestTotal, "endpoint", h.hd.Name(), "status", "success", "code", strconv.Itoa(n)).Inc()
} 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()
}()
}
hdlr.ServeHTTP(w, r.WithContext(ctx))
return
}
} else if !match {
// check for http.HandlerFunc handlers
if !slices.Contains(tracer.DefaultSkipEndpoints, r.URL.Path) {
ctx, sp = h.opts.Tracer.Start(ctx, r.URL.Path+" rpc-server",
tracer.WithSpanKind(tracer.SpanKindServer),
tracer.WithSpanLabels(
"endpoint", r.URL.Path,
),
)
defer func() {
if n := GetRspCode(ctx); n > 399 {
sp.SetStatus(tracer.SpanStatusError, http.StatusText(n))
} else {
sp.SetStatus(tracer.SpanStatusError, http.StatusText(http.StatusNotFound))
}
sp.Finish()
}()
}
if ph, _, err := h.pathHandlers.Search(r.Method, r.URL.Path); err == nil {
ph.(http.HandlerFunc)(w, r.WithContext(ctx))
return
}
h.errorHandler(ctx, nil, w, r, fmt.Errorf("not matching route found"), http.StatusNotFound)
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...)
if !slices.Contains(meter.DefaultSkipEndpoints, handler.name) {
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 {
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()
}
}()
}
defer func() {
n := GetRspCode(ctx)
if n > 399 {
if s, _ := sp.Status(); s != tracer.SpanStatusError {
sp.SetStatus(tracer.SpanStatusError, http.StatusText(n))
}
}
sp.Finish()
}()
// get fields from url values
if len(r.URL.RawQuery) > 0 {
umd, cerr := rflutil.URLMap(r.URL.RawQuery)
@ -165,6 +551,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 +582,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 +634,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 +679,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 +695,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 +705,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)
}
}

12
handler/generate.go Normal file
View File

@ -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"

View File

@ -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}
}

82
handler/health/health.go Normal file
View File

@ -0,0 +1,82 @@
package health_handler
import (
"context"
codecpb "go.unistack.org/micro-proto/v3/codec"
"go.unistack.org/micro/v3/errors"
)
var _ HealthServiceServer = &Handler{}
type Handler struct {
opts Options
}
type CheckFunc func(context.Context) error
type Option func(*Options)
type Options struct {
Version string
Name string
LiveChecks []CheckFunc
ReadyChecks []CheckFunc
}
func LiveChecks(fns ...CheckFunc) Option {
return func(o *Options) {
o.LiveChecks = append(o.LiveChecks, fns...)
}
}
func ReadyChecks(fns ...CheckFunc) Option {
return func(o *Options) {
o.ReadyChecks = append(o.ReadyChecks, fns...)
}
}
func Name(name string) Option {
return func(o *Options) {
o.Name = name
}
}
func Version(version string) Option {
return func(o *Options) {
o.Version = version
}
}
func NewHandler(opts ...Option) *Handler {
options := Options{}
for _, o := range opts {
o(&options)
}
return &Handler{opts: options}
}
func (h *Handler) Live(ctx context.Context, req *codecpb.Frame, rsp *codecpb.Frame) error {
var err error
for _, fn := range h.opts.LiveChecks {
if err = fn(ctx); err != nil {
return errors.ServiceUnavailable(h.opts.Name, "%v", err)
}
}
return nil
}
func (h *Handler) Ready(ctx context.Context, req *codecpb.Frame, rsp *codecpb.Frame) error {
var err error
for _, fn := range h.opts.ReadyChecks {
if err = fn(ctx); err != nil {
return errors.ServiceUnavailable(h.opts.Name, "%v", err)
}
}
return nil
}
func (h *Handler) Version(ctx context.Context, req *codecpb.Frame, rsp *codecpb.Frame) error {
rsp.Data = []byte(h.opts.Version)
return nil
}

View File

@ -0,0 +1,50 @@
syntax = "proto3";
package micro.server.http.v3.handler.health;
option go_package = "go.unistack.org/micro-server-http/v3/handler/health;health_handler";
import "api/annotations.proto";
import "openapiv3/annotations.proto";
import "codec/frame.proto";
service HealthService {
rpc Live(micro.codec.Frame) returns (micro.codec.Frame) {
option (micro.openapiv3.openapiv3_operation) = {
operation_id: "Live";
responses: {
default: {
reference: {
_ref: "micro.codec.Frame";
};
};
};
};
option (micro.api.http) = { get: "/live"; };
};
rpc Ready(micro.codec.Frame) returns (micro.codec.Frame) {
option (micro.openapiv3.openapiv3_operation) = {
operation_id: "Ready";
responses: {
default: {
reference: {
_ref: "micro.codec.Frame";
};
};
};
};
option (micro.api.http) = { get: "/ready"; };
};
rpc Version(micro.codec.Frame) returns (micro.codec.Frame) {
option (micro.openapiv3.openapiv3_operation) = {
operation_id: "Version";
responses: {
default: {
reference: {
_ref: "micro.codec.Frame";
};
};
};
};
option (micro.api.http) = { get: "/version"; };
};
};

View File

@ -0,0 +1,29 @@
// Code generated by protoc-gen-go-micro. DO NOT EDIT.
// versions:
// - protoc-gen-go-micro v3.10.4
// - protoc v5.26.1
// source: health/health.proto
package health_handler
import (
context "context"
codec "go.unistack.org/micro-proto/v3/codec"
client "go.unistack.org/micro/v3/client"
)
var (
HealthServiceName = "HealthService"
)
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
Ready(ctx context.Context, req *codec.Frame, rsp *codec.Frame) error
Version(ctx context.Context, req *codec.Frame, rsp *codec.Frame) error
}

View File

@ -0,0 +1,135 @@
// Code generated by protoc-gen-go-micro. DO NOT EDIT.
// protoc-gen-go-micro version: v3.10.4
// source: health/health.proto
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
}
func (h *healthServiceServer) Live(ctx context.Context, req *codec.Frame, rsp *codec.Frame) error {
return h.HealthServiceServer.Live(ctx, req, rsp)
}
func (h *healthServiceServer) Ready(ctx context.Context, req *codec.Frame, rsp *codec.Frame) error {
return h.HealthServiceServer.Ready(ctx, req, rsp)
}
func (h *healthServiceServer) Version(ctx context.Context, req *codec.Frame, rsp *codec.Frame) error {
return h.HealthServiceServer.Version(ctx, req, rsp)
}
func RegisterHealthServiceServer(s server.Server, sh HealthServiceServer, opts ...server.HandlerOption) error {
type healthService interface {
Live(ctx context.Context, req *codec.Frame, rsp *codec.Frame) error
Ready(ctx context.Context, req *codec.Frame, rsp *codec.Frame) error
Version(ctx context.Context, req *codec.Frame, rsp *codec.Frame) error
}
type HealthService struct {
healthService
}
h := &healthServiceServer{sh}
var nopts []server.HandlerOption
nopts = append(nopts, v3.HandlerEndpoints(HealthServiceServerEndpoints))
return s.Handle(s.NewHandler(&HealthService{h}, append(nopts, opts...)...))
}

132
handler/meter/meter.go Normal file
View File

@ -0,0 +1,132 @@
package meter_handler
import (
"bytes"
"compress/gzip"
"context"
"io"
"strings"
"sync"
codecpb "go.unistack.org/micro-proto/v3/codec"
"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{}
type Handler struct {
opts Options
}
type Option func(*Options)
type Options struct {
Meter meter.Meter
Name string
MeterOptions []meter.Option
DisableCompress bool
}
func Meter(m meter.Meter) Option {
return func(o *Options) {
o.Meter = m
}
}
func Name(name string) Option {
return func(o *Options) {
o.Name = name
}
}
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...)
}
}
func NewOptions(opts ...Option) Options {
options := Options{Meter: meter.DefaultMeter, DisableCompress: false}
for _, o := range opts {
o(&options)
}
return options
}
func NewHandler(opts ...Option) *Handler {
options := NewOptions(opts...)
return &Handler{opts: options}
}
func (h *Handler) Metrics(ctx context.Context, req *codecpb.Frame, rsp *codecpb.Frame) error {
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
}

24
handler/meter/meter.proto Normal file
View File

@ -0,0 +1,24 @@
syntax = "proto3";
package micro.server.http.v3.handler.meter;
option go_package = "go.unistack.org/micro-server-http/v3/handler/meter;meter_handler";
import "api/annotations.proto";
import "openapiv3/annotations.proto";
import "codec/frame.proto";
service MeterService {
rpc Metrics(micro.codec.Frame) returns (micro.codec.Frame) {
option (micro.openapiv3.openapiv3_operation) = {
operation_id: "Metrics";
responses: {
default: {
reference: {
_ref: "micro.codec.Frame";
};
};
};
};
option (micro.api.http) = { get: "/metrics"; };
};
};

View File

@ -0,0 +1,25 @@
// Code generated by protoc-gen-go-micro. DO NOT EDIT.
// versions:
// - protoc-gen-go-micro v3.10.4
// - protoc v5.26.1
// source: meter/meter.proto
package meter_handler
import (
context "context"
codec "go.unistack.org/micro-proto/v3/codec"
client "go.unistack.org/micro/v3/client"
)
var (
MeterServiceName = "MeterService"
)
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
}

View File

@ -0,0 +1,75 @@
// Code generated by protoc-gen-go-micro. DO NOT EDIT.
// protoc-gen-go-micro version: v3.10.4
// source: meter/meter.proto
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
}
func (h *meterServiceServer) Metrics(ctx context.Context, req *codec.Frame, rsp *codec.Frame) error {
return h.MeterServiceServer.Metrics(ctx, req, rsp)
}
func RegisterMeterServiceServer(s server.Server, sh MeterServiceServer, opts ...server.HandlerOption) error {
type meterService interface {
Metrics(ctx context.Context, req *codec.Frame, rsp *codec.Frame) error
}
type MeterService struct {
meterService
}
h := &meterServiceServer{sh}
var nopts []server.HandlerOption
nopts = append(nopts, v3.HandlerEndpoints(MeterServiceServerEndpoints))
return s.Handle(s.NewHandler(&MeterService{h}, append(nopts, opts...)...))
}

46
handler/pprof/pprof.go Normal file
View File

@ -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

View File

@ -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;
}

View File

@ -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>

View File

@ -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

View File

@ -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",
}
)

View File

@ -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)
}
}

View File

@ -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)
}
}

203
http.go
View File

@ -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,83 @@ 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))
}
}
}
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) {
h.opts.Logger.Error(h.opts.Context, fmt.Sprintf("nil method for %s", mname))
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 {
h.opts.Logger.Error(h.opts.Context, fmt.Sprintf("nil mtype for %s", mname))
continue
}
rcvr := reflect.ValueOf(handler)
name := reflect.Indirect(rcvr).Type().Name()
pth := &patHandler{mtype: mtype, name: name, rcvr: rcvr}
hdlr.name = name
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 {
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))
}
}
}
@ -238,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")
@ -267,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 {
@ -313,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))
}
}
@ -328,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
@ -351,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()
@ -369,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 {
@ -393,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
}
}
@ -406,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()
@ -436,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()
@ -444,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 {
@ -481,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 {
@ -489,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
@ -498,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)
@ -542,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:
@ -573,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),

View File

@ -11,7 +11,6 @@ type httpMessage struct {
header metadata.Metadata
topic string
contentType string
body []byte
}
func (r *httpMessage) Topic() string {

View File

@ -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,10 +133,25 @@ func RegisterRPCHandler(b bool) server.Option {
return server.SetOption(registerRPCHandlerKey{}, b)
}
type handlerMetadataKey struct{}
type registerCORSHandlerKey struct{}
func HandlerMetadata(md map[string]map[string]map[string]string) server.HandlerOption {
return server.SetHandlerOption(handlerMetadataKey{}, md)
// 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 {
Name string
Path string
Method string
Body string
Stream bool
}
func HandlerEndpoints(md []EndpointMetadata) server.HandlerOption {
return server.SetHandlerOption(handlerEndpointsKey{}, md)
}
type handlerOptions struct {

View File

@ -58,11 +58,7 @@ func (r *rpcRequest) Header() metadata.Metadata {
}
func (r *rpcRequest) Read() ([]byte, error) {
f := &codec.Frame{}
if err := r.codec.ReadBody(r.rw, f); err != nil {
return nil, err
}
return f.Data, nil
return nil, nil
}
func (r *rpcRequest) Stream() bool {

View File

@ -1,15 +1,14 @@
package http
import (
"bytes"
"context"
"fmt"
"reflect"
"strings"
"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 +100,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 +110,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))
@ -132,13 +130,7 @@ func (s *httpServer) createSubHandler(sb *httpSubscriber, opts server.Options) b
req = req.Elem()
}
buf := bytes.NewBuffer(msg.Body)
if err := cf.ReadHeader(buf, &codec.Message{}, codec.Event); err != nil {
return err
}
if err := cf.ReadBody(buf, req.Interface()); err != nil {
if err := cf.Unmarshal(msg.Body, req.Interface()); err != nil {
return err
}
@ -160,9 +152,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{

View File

@ -8,8 +8,52 @@ import (
"testing"
"go.unistack.org/micro/v3/metadata"
"go.unistack.org/micro/v3/options"
"go.unistack.org/micro/v3/server"
)
func Test_Hook(t *testing.T) {
opts := server.Options{}
var fn server.HandlerFunc = func(fctx context.Context, req server.Request, rsp interface{}) (err error) {
// fmt.Println("1")
return nil
}
var fn2 server.HandlerWrapper = func(next server.HandlerFunc) server.HandlerFunc {
return func(ctx context.Context, req server.Request, rsp interface{}) error {
// fmt.Println("2")
return next(ctx, req, rsp)
}
}
var fn3 server.HandlerWrapper = func(next server.HandlerFunc) server.HandlerFunc {
return func(ctx context.Context, req server.Request, rsp interface{}) error {
// fmt.Println("3")
return next(ctx, req, rsp)
}
}
var fn4 server.HandlerWrapper = func(next server.HandlerFunc) server.HandlerFunc {
return func(ctx context.Context, req server.Request, rsp interface{}) error {
// fmt.Println("4")
return next(ctx, req, rsp)
}
}
opts.Hooks = append(opts.Hooks, fn2, fn3, fn4)
opts.Hooks.EachNext(func(hook options.Hook) {
if h, ok := hook.(server.HandlerWrapper); ok {
// fmt.Printf("h %#+v\n", h)
fn = h(fn)
}
})
err := fn(nil, nil, nil)
if err != nil {
t.Fatal(err)
}
}
func TestFillrequest(t *testing.T) {
md := metadata.New(1)
md.Set("ClientID", "xxx")