Compare commits

..

42 Commits
v3.11.5 ... v3

Author SHA1 Message Date
f071b4b145 add server type in metrics and tracing
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2024-11-10 20:18:20 +03:00
44a2f58a69 fixup endpoint name in tracing and metrics
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2024-11-10 18:30:19 +03:00
13432d94bb update for latest micro logger changes
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2024-10-12 13:25:09 +03:00
8d7d145e16 update returning errors (#190)
fix error handling in handler

Reviewed-on: #190
Co-authored-by: Evstigneev Denis <danteevstigneev@yandex.ru>
Co-committed-by: Evstigneev Denis <danteevstigneev@yandex.ru>
2024-10-10 14:29:01 +03:00
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
24 changed files with 2119 additions and 197 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

21
go.mod
View File

@ -1,16 +1,23 @@
module go.unistack.org/micro-server-http/v3
go 1.18
go 1.22
toolchain go1.23.1
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.13
go.unistack.org/micro-codec-yaml/v3 v3.10.2
go.unistack.org/micro-proto/v3 v3.4.1
go.unistack.org/micro/v3 v3.10.97
golang.org/x/net v0.30.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.26.0 // indirect
google.golang.org/genproto/googleapis/rpc v0.0.0-20241007155032-5fefd90f89a9 // indirect
google.golang.org/grpc v1.67.1 // indirect
google.golang.org/protobuf v1.35.1 // indirect
gopkg.in/yaml.v3 v3.0.1 // indirect
)

1456
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"
)
@ -22,7 +29,7 @@ var (
DefaultErrorHandler = func(ctx context.Context, s server.Handler, w http.ResponseWriter, r *http.Request, err error, status int) {
w.WriteHeader(status)
if _, cerr := w.Write([]byte(err.Error())); cerr != nil {
logger.DefaultLogger.Errorf(ctx, "write failed: %v", cerr)
logger.DefaultLogger.Error(ctx, "write error", cerr)
}
}
DefaultContentType = "application/json"
@ -115,10 +122,16 @@ func (h *Server) HTTPHandlerFunc(handler interface{}) (http.HandlerFunc, error)
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, ",")
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
@ -266,9 +279,11 @@ func (h *Server) HTTPHandlerFunc(handler interface{}) (http.HandlerFunc, error)
}
// 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)
@ -318,7 +333,7 @@ func (h *Server) HTTPHandlerFunc(handler interface{}) (http.HandlerFunc, error)
}
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
}
@ -328,7 +343,7 @@ func (h *Server) HTTPHandlerFunc(handler interface{}) (http.HandlerFunc, error)
w.WriteHeader(scode)
if _, cerr := w.Write(buf); cerr != nil {
handler.sopts.Logger.Errorf(ctx, "write failed: %v", cerr)
handler.sopts.Logger.Error(ctx, "write failed", cerr)
}
}, nil
}
@ -339,8 +354,11 @@ func (h *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
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)
@ -415,21 +433,116 @@ func (h *Server) 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)
endpointName := fmt.Sprintf("%s.%s", hldr.name, hldr.mtype.method.Name)
if !slices.Contains(tracer.DefaultSkipEndpoints, endpointName) {
ctx, sp = h.opts.Tracer.Start(ctx, "rpc-server",
tracer.WithSpanKind(tracer.SpanKindServer),
tracer.WithSpanLabels(
"endpoint", endpointName,
),
)
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, endpointName) {
h.opts.Meter.Counter(semconv.ServerRequestInflight, "endpoint", endpointName, "server", "http").Inc()
defer func() {
n := GetRspCode(ctx)
if n > 399 {
h.opts.Meter.Counter(semconv.ServerRequestTotal, "endpoint", endpointName, "server", "http", "status", "success", "code", strconv.Itoa(n)).Inc()
} else {
h.opts.Meter.Counter(semconv.ServerRequestTotal, "endpoint", endpointName, "server", "http", "status", "failure", "code", strconv.Itoa(n)).Inc()
}
te := time.Since(ts)
h.opts.Meter.Summary(semconv.ServerRequestLatencyMicroseconds, "endpoint", endpointName, "server", "http").Update(te.Seconds())
h.opts.Meter.Histogram(semconv.ServerRequestDurationSeconds, "endpoint", endpointName, "server", "http").Update(te.Seconds())
h.opts.Meter.Counter(semconv.ServerRequestInflight, "endpoint", endpointName, "server", "http").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, "rpc-server",
tracer.WithSpanKind(tracer.SpanKindServer),
tracer.WithSpanLabels(
"endpoint", r.URL.Path,
"server", "http",
),
)
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)
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,
"server", "http",
),
}
if slices.Contains(tracer.DefaultSkipEndpoints, endpointName) {
topts = append(topts, tracer.WithSpanRecord(false))
}
ctx, sp = h.opts.Tracer.Start(ctx, "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, "server", "http").Update(te.Seconds())
h.opts.Meter.Histogram(semconv.ServerRequestDurationSeconds, "endpoint", handler.name, "server", "http").Update(te.Seconds())
h.opts.Meter.Counter(semconv.ServerRequestInflight, "endpoint", handler.name, "server", "http").Dec()
n := GetRspCode(ctx)
if n > 399 {
h.opts.Meter.Counter(semconv.ServerRequestTotal, "endpoint", handler.name, "server", "http", "status", "failure", "code", strconv.Itoa(n)).Inc()
} else {
h.opts.Meter.Counter(semconv.ServerRequestTotal, "endpoint", handler.name, "server", "http", "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)
@ -525,13 +638,18 @@ func (h *Server) 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)
@ -580,17 +698,18 @@ func (h *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
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 {
if err != nil {
if handler.sopts.Logger.V(logger.ErrorLevel) {
handler.sopts.Logger.Error(handler.sopts.Context, "handler error", err)
}
scode = http.StatusInternalServerError
} else 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)
handler.sopts.Logger.Error(ctx, "respoonse write error", cerr)
}
}

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

View File

@ -1,4 +1,4 @@
package health // import "go.unistack.org/micro-server-http/v3/handler/health"
package health_handler
import (
"context"

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

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

View File

@ -1,20 +0,0 @@
window.onload = function() {
//<editor-fold desc="Changeable Configuration Block">
// the following lines will be replaced by docker/configurator, when it runs in a docker-container
window.ui = SwaggerUIBundle({
url: "{{ .SWAGGER }}",
dom_id: '#swagger-ui',
deepLinking: true,
presets: [
SwaggerUIBundle.presets.apis,
SwaggerUIStandalonePreset
],
plugins: [
SwaggerUIBundle.plugins.DownloadUrl
],
layout: "StandaloneLayout"
});
//</editor-fold>
};

View File

@ -1,4 +1,4 @@
package swaggerui // import "go.unistack.org/micro-server-http/v3/handler/swagger-ui"
package swaggerui_handler
import (
"embed"

View File

@ -1,4 +1,4 @@
package swaggerui
package swaggerui_handler
import (
"net/http"

View File

@ -1,4 +1,4 @@
package swagger
package swagger_handler
import (
"io/fs"
@ -29,6 +29,12 @@ var Handler = func(dst map[string]interface{}, fsys fs.FS) http.HandlerFunc {
return
}
if dst == nil {
w.WriteHeader(http.StatusOK)
_, _ = w.Write(buf)
return
}
var src interface{}
if err = c.Unmarshal(buf, src); err != nil {

118
http.go
View File

@ -191,6 +191,11 @@ func (h *Server) NewHandler(handler interface{}, opts ...server.HandlerOption) s
}
*/
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,16 +209,16 @@ func (h *Server) NewHandler(handler interface{}, opts ...server.HandlerOption) s
}
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
}
mtype, err := prepareEndpoint(method)
if err != nil && h.opts.Logger.V(logger.ErrorLevel) {
h.opts.Logger.Errorf(h.opts.Context, "%v", err)
h.opts.Logger.Error(h.opts.Context, "endpoint error", 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 *Server) NewHandler(handler interface{}, opts ...server.HandlerOption) s
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,16 +268,16 @@ func (h *Server) NewHandler(handler interface{}, opts ...server.HandlerOption) s
}
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
}
mtype, err := prepareEndpoint(method)
if err != nil && h.opts.Logger.V(logger.ErrorLevel) {
h.opts.Logger.Errorf(h.opts.Context, "%v", err)
h.opts.Logger.Error(h.opts.Context, "prepare endpoint error", 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 *Server) NewHandler(handler interface{}, opts ...server.HandlerOption) s
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))
}
}
}
@ -363,7 +387,7 @@ func (h *Server) 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))
}
}
@ -426,7 +450,7 @@ func (h *Server) 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 {
@ -450,10 +474,10 @@ func (h *Server) 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
}
}
@ -493,7 +517,7 @@ func (h *Server) 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()
@ -501,7 +525,6 @@ func (h *Server) Start() error {
h.Unlock()
var handler http.Handler
var srvFunc func(net.Listener) error
// nolint: nestif
if h.opts.Context != nil {
@ -538,7 +561,7 @@ func (h *Server) 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 {
@ -552,6 +575,7 @@ func (h *Server) Start() error {
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
@ -559,25 +583,19 @@ func (h *Server) 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)
@ -603,28 +621,28 @@ func (h *Server) 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:
@ -634,14 +652,22 @@ func (h *Server) 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

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

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"
)
@ -111,7 +110,6 @@ func (s *Server) createSubHandler(sb *httpSubscriber, opts server.Options) broke
}
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 *Server) createSubHandler(sb *httpSubscriber, opts server.Options) broke
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 *Server) createSubHandler(sb *httpSubscriber, opts server.Options) broke
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")