unistack-org/micro-client-http
4
0
Fork 2
Code Issues 3 Pull Requests Actions Packages Projects Releases 62 Wiki Activity

lintering && fix call hook #122

Merged
vtolstov merged 4 commits from devstigneev/micro-client-http:v3 into v3 2024-12-18 20:43:01 +03:00
Conversation 0 Commits 4 Files Changed 4 +64 -36
4 changed files with 64 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 0677ac1b86 - Show all commits

31
.gitignore vendored
View File

@ -1,16 +1,39 @@
# Develop tools
/.vscode/
/.idea/
.idea
.vscode
# Binaries for programs and plugins
*.exe
*.exe~
*.dll
*.so
*.dylib
.idea
# Test binary, built with `go test -c`
# Folders
_obj
_test
_build
.DS_Store
# Architecture specific extensions/prefixes
*.[568vq]
[568vq].out
*.cgo1.go
*.cgo2.c
_cgo_defun.c
_cgo_gotypes.go
_cgo_export.*
# Test binary, build 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/
# vim temp files
*~
*.swp
*.swo

53
http.go
View File

@ -6,7 +6,7 @@ import (
"bytes"
"context"
"fmt"
"io/ioutil"
"io"
"net"
"net/http"
"net/url"
@ -46,12 +46,11 @@ type httpClient struct {
httpcli *http.Client
opts client.Options
sync.RWMutex
init bool
}
func newRequest(ctx context.Context, log logger.Logger, addr string, req client.Request, ct string, cf codec.Codec, msg interface{}, opts client.CallOptions) (*http.Request, error) {
var tags []string
parameters := map[string]map[string]string{}
var parameters map[string]map[string]string
scheme := "http"
method := http.MethodPost
body := "*" // as like google api http annotation
@ -82,6 +81,9 @@ func newRequest(ctx context.Context, log logger.Logger, addr string, req client.
tags = t
}
if k, ok := opts.Context.Value(headerKey{}).([]string); ok && len(k) > 0 {
if parameters == nil {
parameters = make(map[string]map[string]string)
}
m, ok := parameters["header"]
if !ok {
m = make(map[string]string)
@ -92,6 +94,9 @@ func newRequest(ctx context.Context, log logger.Logger, addr string, req client.
}
}
if k, ok := opts.Context.Value(cookieKey{}).([]string); ok && len(k) > 0 {
if parameters == nil {
parameters = make(map[string]map[string]string)
}
m, ok := parameters["cookie"]
if !ok {
m = make(map[string]string)
@ -118,7 +123,7 @@ func newRequest(ctx context.Context, log logger.Logger, addr string, req client.
u, err = u.Parse(path)
if err != nil {
return nil, errors.BadRequest("go.micro.client", err.Error())
return nil, errors.BadRequest("go.micro.client", "%+v", err)
}
var nmsg interface{}
@ -129,12 +134,12 @@ func newRequest(ctx context.Context, log logger.Logger, addr string, req client.
}
if err != nil {
return nil, errors.BadRequest("go.micro.client", err.Error())
return nil, errors.BadRequest("go.micro.client", "%+v", err)
}
u, err = url.Parse(fmt.Sprintf("%s://%s%s", scheme, host, path))
if err != nil {
return nil, errors.BadRequest("go.micro.client", err.Error())
return nil, errors.BadRequest("go.micro.client", "%+v", err)
}
var cookies []*http.Cookie
@ -177,11 +182,11 @@ func newRequest(ctx context.Context, log logger.Logger, addr string, req client.
for k, required := range vm {
v, err = rutil.StructFieldByPath(msg, k)
if err != nil {
return nil, errors.BadRequest("go.micro.client", err.Error())
return nil, errors.BadRequest("go.micro.client", "%+v", err)
}
if rutil.IsZero(v) {
if required == "true" {
return nil, errors.BadRequest("go.micro.client", fmt.Sprintf("required field %s not set", k))
return nil, errors.BadRequest("go.micro.client", "required field %s not set", k)
}
continue
}
@ -197,12 +202,12 @@ func newRequest(ctx context.Context, log logger.Logger, addr string, req client.
b, err := cf.Marshal(nmsg)
if err != nil {
return nil, errors.BadRequest("go.micro.client", err.Error())
return nil, errors.BadRequest("go.micro.client", "%+v", err)
}
var hreq *http.Request
if len(b) > 0 {
hreq, err = http.NewRequestWithContext(ctx, method, u.String(), ioutil.NopCloser(bytes.NewBuffer(b)))
hreq, err = http.NewRequestWithContext(ctx, method, u.String(), io.NopCloser(bytes.NewBuffer(b)))
hreq.ContentLength = int64(len(b))
header.Set("Content-Length", fmt.Sprintf("%d", hreq.ContentLength))
} else {
@ -210,7 +215,7 @@ func newRequest(ctx context.Context, log logger.Logger, addr string, req client.
}
if err != nil {
return nil, errors.BadRequest("go.micro.client", err.Error())
return nil, errors.BadRequest("go.micro.client", "%+v", err)
}
hreq.Header = header
@ -233,7 +238,7 @@ func (c *httpClient) call(ctx context.Context, addr string, req client.Request,
cf, err := c.newCodec(ct)
if err != nil {
return errors.BadRequest("go.micro.client", err.Error())
return errors.BadRequest("go.micro.client", "%+v", err)
}
hreq, err := newRequest(ctx, c.opts.Logger, addr, req, ct, cf, req.Body(), opts)
if err != nil {
@ -246,14 +251,14 @@ func (c *httpClient) call(ctx context.Context, addr string, req client.Request,
switch err := err.(type) {
case *url.Error:
if err, ok := err.Err.(net.Error); ok && err.Timeout() {
return errors.Timeout("go.micro.client", err.Error())
return errors.Timeout("go.micro.client", "%+v", err)
}
case net.Error:
if err.Timeout() {
return errors.Timeout("go.micro.client", err.Error())
return errors.Timeout("go.micro.client", "%+v", err)
}
}
return errors.InternalServerError("go.micro.client", err.Error())
return errors.InternalServerError("go.micro.client", "%+v", err)
}
defer hrsp.Body.Close()
@ -270,12 +275,12 @@ func (c *httpClient) stream(ctx context.Context, addr string, req client.Request
// get codec
cf, err := c.newCodec(ct)
if err != nil {
return nil, errors.BadRequest("go.micro.client", err.Error())
return nil, errors.BadRequest("go.micro.client", "%+v", err)
}
cc, err := (c.httpcli.Transport).(*http.Transport).DialContext(ctx, "tcp", addr)
if err != nil {
return nil, errors.InternalServerError("go.micro.client", fmt.Sprintf("Error dialing: %v", err))
return nil, errors.InternalServerError("go.micro.client", "Error dialing: %v", err)
}
return &httpStream{
@ -318,7 +323,7 @@ func (c *httpClient) Init(opts ...client.Option) error {
c.funcPublish = c.fnPublish
c.funcBatchPublish = c.fnBatchPublish
c.opts.Hooks.EachNext(func(hook options.Hook) {
c.opts.Hooks.EachPrev(func(hook options.Hook) {
switch h := hook.(type) {
case client.HookCall:
c.funcCall = h(c.funcCall)
@ -424,7 +429,7 @@ func (c *httpClient) fnCall(ctx context.Context, req client.Request, rsp interfa
// call backoff first. Someone may want an initial start delay
t, err := callOpts.Backoff(ctx, req, i)
if err != nil {
return errors.InternalServerError("go.micro.client", err.Error())
return errors.InternalServerError("go.micro.client", "%+v", err)
}
// only sleep if greater than 0
@ -438,7 +443,7 @@ func (c *httpClient) fnCall(ctx context.Context, req client.Request, rsp interfa
// TODO apply any filtering here
routes, err = c.opts.Lookup(ctx, req, callOpts)
if err != nil {
return errors.InternalServerError("go.micro.client", err.Error())
return errors.InternalServerError("go.micro.client", "%+v", err)
}
// balance the list of nodes
@ -583,7 +588,7 @@ func (c *httpClient) fnStream(ctx context.Context, req client.Request, opts ...c
// call backoff first. Someone may want an initial start delay
t, cerr := callOpts.Backoff(ctx, req, i)
if cerr != nil {
return nil, errors.InternalServerError("go.micro.client", cerr.Error())
return nil, errors.InternalServerError("go.micro.client", "%+v", cerr)
}
// only sleep if greater than 0
@ -597,7 +602,7 @@ func (c *httpClient) fnStream(ctx context.Context, req client.Request, opts ...c
// TODO apply any filtering here
routes, err = c.opts.Lookup(ctx, req, callOpts)
if err != nil {
return nil, errors.InternalServerError("go.micro.client", err.Error())
return nil, errors.InternalServerError("go.micro.client", "%+v", err)
}
// balance the list of nodes
@ -722,12 +727,12 @@ func (c *httpClient) publish(ctx context.Context, ps []client.Message, opts ...c
// use codec for payload
cf, err := c.newCodec(p.ContentType())
if err != nil {
return errors.InternalServerError("go.micro.client", err.Error())
return errors.InternalServerError("go.micro.client", "%+v", err)
}
// set the body
b, err := cf.Marshal(p.Payload())
if err != nil {
return errors.InternalServerError("go.micro.client", err.Error())
return errors.InternalServerError("go.micro.client", "%+v", err)
}
body = b
}

8
stream.go
View File

@ -90,7 +90,7 @@ func (h *httpStream) Recv(msg interface{}) error {
hrsp, err := http.ReadResponse(h.reader, new(http.Request))
if err != nil {
return errors.InternalServerError("go.micro.client", err.Error())
return errors.InternalServerError("go.micro.client", "%+v", err)
}
defer hrsp.Body.Close()
@ -136,7 +136,7 @@ func (h *httpStream) parseRsp(ctx context.Context, log logger.Logger, hrsp *http
if log.V(logger.ErrorLevel) {
log.Error(ctx, "failed to read body", err)
}
return errors.InternalServerError("go.micro.client", string(buf))
return errors.InternalServerError("go.micro.client", "%s", buf)
}
}
@ -146,7 +146,7 @@ func (h *httpStream) parseRsp(ctx context.Context, log logger.Logger, hrsp *http
if hrsp.StatusCode < 400 {
if err = cf.Unmarshal(buf, rsp); err != nil {
return errors.InternalServerError("go.micro.client", err.Error())
return errors.InternalServerError("go.micro.client", "%+v", err)
}
return nil
}
@ -163,7 +163,7 @@ func (h *httpStream) parseRsp(ctx context.Context, log logger.Logger, hrsp *http
}
if cerr := cf.Unmarshal(buf, rerr); cerr != nil {
return errors.InternalServerError("go.micro.client", cerr.Error())
return errors.InternalServerError("go.micro.client", "%+v", cerr)
}
if err, ok = rerr.(error); !ok {

8
util.go
View File

@ -334,7 +334,7 @@ func (h *httpClient) parseRsp(ctx context.Context, hrsp *http.Response, rsp inte
if h.opts.Logger.V(logger.ErrorLevel) {
h.opts.Logger.Error(ctx, "failed to read body", err)
}
return errors.InternalServerError("go.micro.client", string(buf))
return errors.InternalServerError("go.micro.client", "%s", buf)
}
}
@ -343,7 +343,7 @@ func (h *httpClient) parseRsp(ctx context.Context, hrsp *http.Response, rsp inte
if h.opts.Logger.V(logger.DebugLevel) {
h.opts.Logger.Debug(ctx, fmt.Sprintf("response with %v unknown content-type %s %s", hrsp.Header, ct, buf))
}
return errors.InternalServerError("go.micro.client", cerr.Error())
return errors.InternalServerError("go.micro.client", "%+v", cerr)
}
if h.opts.Logger.V(logger.DebugLevel) {
@ -353,7 +353,7 @@ func (h *httpClient) parseRsp(ctx context.Context, hrsp *http.Response, rsp inte
// succeseful response
if hrsp.StatusCode < 400 {
if err = cf.Unmarshal(buf, rsp); err != nil {
return errors.InternalServerError("go.micro.client", err.Error())
return errors.InternalServerError("go.micro.client", "%+v", err)
}
return nil
}
@ -373,7 +373,7 @@ func (h *httpClient) parseRsp(ctx context.Context, hrsp *http.Response, rsp inte
}
if cerr := cf.Unmarshal(buf, rerr); cerr != nil {
return errors.InternalServerError("go.micro.client", cerr.Error())
return errors.InternalServerError("go.micro.client", "%+v", cerr)
}
if err, ok = rerr.(error); !ok {