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 +69 -31
4 changed files with 69 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

39
.gitignore vendored Normal file
View File

@ -0,0 +1,39 @@
# Develop tools
/.vscode/
/.idea/
.idea
.vscode
# Binaries for programs and plugins
*.exe
*.exe~
*.dll
*.so
*.dylib
# 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
# vim temp files
*~
*.swp
*.swo

45
http.go
View File

@ -6,7 +6,7 @@ import (
"bytes" "bytes"
"context" "context"
"fmt" "fmt"
"io/ioutil" "io"
"net" "net"
"net/http" "net/http"
"net/url" "net/url"
@ -46,7 +46,6 @@ type httpClient struct {
httpcli *http.Client httpcli *http.Client
opts client.Options opts client.Options
sync.RWMutex 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) { 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) {
@ -124,7 +123,7 @@ func newRequest(ctx context.Context, log logger.Logger, addr string, req client.
u, err = u.Parse(path) u, err = u.Parse(path)
if err != nil { if err != nil {
return nil, errors.BadRequest("go.micro.client", err.Error()) return nil, errors.BadRequest("go.micro.client", "%+v", err)
} }
var nmsg interface{} var nmsg interface{}
@ -135,12 +134,12 @@ func newRequest(ctx context.Context, log logger.Logger, addr string, req client.
} }
if err != nil { 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)) u, err = url.Parse(fmt.Sprintf("%s://%s%s", scheme, host, path))
if err != nil { 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 var cookies []*http.Cookie
@ -183,11 +182,11 @@ func newRequest(ctx context.Context, log logger.Logger, addr string, req client.
for k, required := range vm { for k, required := range vm {
v, err = rutil.StructFieldByPath(msg, k) v, err = rutil.StructFieldByPath(msg, k)
if err != nil { 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 rutil.IsZero(v) {
if required == "true" { 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 continue
} }
@ -203,12 +202,12 @@ func newRequest(ctx context.Context, log logger.Logger, addr string, req client.
b, err := cf.Marshal(nmsg) b, err := cf.Marshal(nmsg)
if err != nil { 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 var hreq *http.Request
if len(b) > 0 { 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)) hreq.ContentLength = int64(len(b))
header.Set("Content-Length", fmt.Sprintf("%d", hreq.ContentLength)) header.Set("Content-Length", fmt.Sprintf("%d", hreq.ContentLength))
} else { } else {
@ -216,7 +215,7 @@ func newRequest(ctx context.Context, log logger.Logger, addr string, req client.
} }
if err != nil { if err != nil {
return nil, errors.BadRequest("go.micro.client", err.Error()) return nil, errors.BadRequest("go.micro.client", "%+v", err)
} }
hreq.Header = header hreq.Header = header
@ -239,7 +238,7 @@ func (c *httpClient) call(ctx context.Context, addr string, req client.Request,
cf, err := c.newCodec(ct) cf, err := c.newCodec(ct)
if err != nil { 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) hreq, err := newRequest(ctx, c.opts.Logger, addr, req, ct, cf, req.Body(), opts)
if err != nil { if err != nil {
@ -252,14 +251,14 @@ func (c *httpClient) call(ctx context.Context, addr string, req client.Request,
switch err := err.(type) { switch err := err.(type) {
case *url.Error: case *url.Error:
if err, ok := err.Err.(net.Error); ok && err.Timeout() { 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: case net.Error:
if err.Timeout() { 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() defer hrsp.Body.Close()
@ -276,12 +275,12 @@ func (c *httpClient) stream(ctx context.Context, addr string, req client.Request
// get codec // get codec
cf, err := c.newCodec(ct) cf, err := c.newCodec(ct)
if err != nil { 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) cc, err := (c.httpcli.Transport).(*http.Transport).DialContext(ctx, "tcp", addr)
if err != nil { 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{ return &httpStream{
@ -324,7 +323,7 @@ func (c *httpClient) Init(opts ...client.Option) error {
c.funcPublish = c.fnPublish c.funcPublish = c.fnPublish
c.funcBatchPublish = c.fnBatchPublish c.funcBatchPublish = c.fnBatchPublish
c.opts.Hooks.EachNext(func(hook options.Hook) { c.opts.Hooks.EachPrev(func(hook options.Hook) {
switch h := hook.(type) { switch h := hook.(type) {
case client.HookCall: case client.HookCall:
c.funcCall = h(c.funcCall) c.funcCall = h(c.funcCall)
@ -430,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 // call backoff first. Someone may want an initial start delay
t, err := callOpts.Backoff(ctx, req, i) t, err := callOpts.Backoff(ctx, req, i)
if err != nil { 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 // only sleep if greater than 0
@ -444,7 +443,7 @@ func (c *httpClient) fnCall(ctx context.Context, req client.Request, rsp interfa
// TODO apply any filtering here // TODO apply any filtering here
routes, err = c.opts.Lookup(ctx, req, callOpts) routes, err = c.opts.Lookup(ctx, req, callOpts)
if err != nil { if err != nil {
return errors.InternalServerError("go.micro.client", err.Error()) return errors.InternalServerError("go.micro.client", "%+v", err)
} }
// balance the list of nodes // balance the list of nodes
@ -589,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 // call backoff first. Someone may want an initial start delay
t, cerr := callOpts.Backoff(ctx, req, i) t, cerr := callOpts.Backoff(ctx, req, i)
if cerr != nil { 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 // only sleep if greater than 0
@ -603,7 +602,7 @@ func (c *httpClient) fnStream(ctx context.Context, req client.Request, opts ...c
// TODO apply any filtering here // TODO apply any filtering here
routes, err = c.opts.Lookup(ctx, req, callOpts) routes, err = c.opts.Lookup(ctx, req, callOpts)
if err != nil { 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 // balance the list of nodes
@ -728,12 +727,12 @@ func (c *httpClient) publish(ctx context.Context, ps []client.Message, opts ...c
// use codec for payload // use codec for payload
cf, err := c.newCodec(p.ContentType()) cf, err := c.newCodec(p.ContentType())
if err != nil { if err != nil {
return errors.InternalServerError("go.micro.client", err.Error()) return errors.InternalServerError("go.micro.client", "%+v", err)
} }
// set the body // set the body
b, err := cf.Marshal(p.Payload()) b, err := cf.Marshal(p.Payload())
if err != nil { if err != nil {
return errors.InternalServerError("go.micro.client", err.Error()) return errors.InternalServerError("go.micro.client", "%+v", err)
} }
body = b 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)) hrsp, err := http.ReadResponse(h.reader, new(http.Request))
if err != nil { if err != nil {
return errors.InternalServerError("go.micro.client", err.Error()) return errors.InternalServerError("go.micro.client", "%+v", err)
} }
defer hrsp.Body.Close() 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) { if log.V(logger.ErrorLevel) {
log.Error(ctx, "failed to read body", err) 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 hrsp.StatusCode < 400 {
if err = cf.Unmarshal(buf, rsp); err != nil { 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 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 { 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 { 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) { if h.opts.Logger.V(logger.ErrorLevel) {
h.opts.Logger.Error(ctx, "failed to read body", err) 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) { 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)) 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) { 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 // succeseful response
if hrsp.StatusCode < 400 { if hrsp.StatusCode < 400 {
if err = cf.Unmarshal(buf, rsp); err != nil { 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 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 { 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 { if err, ok = rerr.(error); !ok {