Compare commits
12 Commits
Author | SHA1 | Date | |
---|---|---|---|
68b32989fc | |||
21a41a8e03 | |||
9150958044 | |||
|
e358db44ca | ||
5cdd48329e | |||
|
01b5e1db54 | ||
|
900e08458b | ||
|
f4fff1c77a | ||
4bbf97a309 | |||
02f29b0ef3 | |||
|
ba69a7dfcd | ||
|
45aee3c441 |
2
go.mod
2
go.mod
@@ -2,4 +2,4 @@ module github.com/unistack-org/micro-client-http/v3
|
||||
|
||||
go 1.16
|
||||
|
||||
require github.com/unistack-org/micro/v3 v3.3.0
|
||||
require github.com/unistack-org/micro/v3 v3.3.12
|
||||
|
8
go.sum
8
go.sum
@@ -5,11 +5,11 @@ github.com/google/uuid v1.2.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+
|
||||
github.com/imdario/mergo v0.3.12/go.mod h1:jmQim1M+e3UYxmgPu/WyfjB3N3VflVyUjjjwH0dnCYA=
|
||||
github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ=
|
||||
github.com/silas/dag v0.0.0-20210121180416-41cf55125c34/go.mod h1:7RTUFBdIRC9nZ7/3RyRNH1bdqIShrDejd1YbLwgPS+I=
|
||||
github.com/unistack-org/micro/v3 v3.3.0 h1:pEj/8QVFzMlNMEL//q/Te8qgG+XI6LTYIQrb6hMymgk=
|
||||
github.com/unistack-org/micro/v3 v3.3.0/go.mod h1:iJwCWq2PECMxigfqe6TPC5GLWvj6P94Kk+PTVZGL3w8=
|
||||
golang.org/x/net v0.0.0-20210316092652-d523dce5a7f4/go.mod h1:RBQZq4jEuRlivfhVLdyRGr576XBO4/greRjx4P4O3yc=
|
||||
github.com/unistack-org/micro/v3 v3.3.12 h1:3CZsjskLlu2MG3uW5A3e++KDaFdUNlGS/Ck+sGhtHL4=
|
||||
github.com/unistack-org/micro/v3 v3.3.12/go.mod h1:98hNcMXp/WyWJwLwCuwrhN1Jm7aCWaRNsMfRjK8Fq+Y=
|
||||
golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
|
||||
golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20210315160823-c6e025ad8005/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
|
||||
golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
|
||||
golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
|
||||
|
71
http.go
71
http.go
@@ -19,6 +19,7 @@ import (
|
||||
"github.com/unistack-org/micro/v3/errors"
|
||||
"github.com/unistack-org/micro/v3/metadata"
|
||||
"github.com/unistack-org/micro/v3/router"
|
||||
rutil "github.com/unistack-org/micro/v3/util/reflect"
|
||||
)
|
||||
|
||||
var (
|
||||
@@ -98,13 +99,24 @@ func newRequest(addr string, req client.Request, ct string, cf codec.Codec, msg
|
||||
return nil, errors.BadRequest("go.micro.client", err.Error())
|
||||
}
|
||||
|
||||
// marshal request is struct not empty
|
||||
if nmsg != nil {
|
||||
var b []byte
|
||||
|
||||
if nmsg != nil {
|
||||
if ct == "application/x-www-form-urlencoded" {
|
||||
data, err := rutil.StructURLValues(nmsg, "", tags)
|
||||
if err != nil {
|
||||
return nil, errors.BadRequest("go.micro.client", err.Error())
|
||||
}
|
||||
b = []byte(data.Encode())
|
||||
} else {
|
||||
b, err = cf.Marshal(nmsg)
|
||||
if err != nil {
|
||||
return nil, errors.BadRequest("go.micro.client", err.Error())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if len(b) > 0 {
|
||||
hreq.Body = ioutil.NopCloser(bytes.NewBuffer(b))
|
||||
hreq.ContentLength = int64(len(b))
|
||||
}
|
||||
@@ -114,21 +126,32 @@ func newRequest(addr string, req client.Request, ct string, cf codec.Codec, msg
|
||||
|
||||
func (h *httpClient) call(ctx context.Context, addr string, req client.Request, rsp interface{}, opts client.CallOptions) error {
|
||||
header := make(http.Header, 2)
|
||||
if md, ok := metadata.FromContext(ctx); ok {
|
||||
if md, ok := metadata.FromOutgoingContext(ctx); ok {
|
||||
for k, v := range md {
|
||||
header.Set(k, v)
|
||||
}
|
||||
}
|
||||
|
||||
ct := req.ContentType()
|
||||
if len(opts.ContentType) > 0 {
|
||||
ct = opts.ContentType
|
||||
}
|
||||
|
||||
// set timeout in nanoseconds
|
||||
header.Set("Timeout", fmt.Sprintf("%d", opts.RequestTimeout))
|
||||
// set the content type for the request
|
||||
header.Set("Content-Type", ct)
|
||||
|
||||
var cf codec.Codec
|
||||
var err error
|
||||
// get codec
|
||||
cf, err := h.newCodec(ct)
|
||||
switch ct {
|
||||
case "application/x-www-form-urlencoded":
|
||||
cf, err = h.newCodec(DefaultContentType)
|
||||
default:
|
||||
cf, err = h.newCodec(ct)
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
return errors.InternalServerError("go.micro.client", err.Error())
|
||||
}
|
||||
@@ -164,7 +187,7 @@ func (h *httpClient) call(ctx context.Context, addr string, req client.Request,
|
||||
func (h *httpClient) stream(ctx context.Context, addr string, req client.Request, opts client.CallOptions) (client.Stream, error) {
|
||||
var header http.Header
|
||||
|
||||
if md, ok := metadata.FromContext(ctx); ok {
|
||||
if md, ok := metadata.FromOutgoingContext(ctx); ok {
|
||||
header = make(http.Header, len(md)+2)
|
||||
for k, v := range md {
|
||||
header.Set(k, v)
|
||||
@@ -174,6 +197,10 @@ func (h *httpClient) stream(ctx context.Context, addr string, req client.Request
|
||||
}
|
||||
|
||||
ct := req.ContentType()
|
||||
if len(opts.ContentType) > 0 {
|
||||
ct = opts.ContentType
|
||||
}
|
||||
|
||||
// set timeout in nanoseconds
|
||||
header.Set("Timeout", fmt.Sprintf("%d", opts.RequestTimeout))
|
||||
// set the content type for the request
|
||||
@@ -190,7 +217,7 @@ func (h *httpClient) stream(ctx context.Context, addr string, req client.Request
|
||||
if err == nil && u.Scheme != "" && u.Host != "" {
|
||||
dialAddr = u.Host
|
||||
}
|
||||
cc, err := h.dialer.DialContext(ctx, "tcp", addr)
|
||||
cc, err := (h.httpcli.Transport).(*http.Transport).DialContext(ctx, "tcp", addr)
|
||||
if err != nil {
|
||||
return nil, errors.InternalServerError("go.micro.client", fmt.Sprintf("Error dialing: %v", err))
|
||||
}
|
||||
@@ -255,8 +282,8 @@ func (h *httpClient) NewMessage(topic string, msg interface{}, opts ...client.Me
|
||||
return newHTTPMessage(topic, msg, h.opts.ContentType, opts...)
|
||||
}
|
||||
|
||||
func (h *httpClient) NewRequest(service, method string, req interface{}, reqOpts ...client.RequestOption) client.Request {
|
||||
return newHTTPRequest(service, method, req, h.opts.ContentType, reqOpts...)
|
||||
func (h *httpClient) NewRequest(service, method string, req interface{}, opts ...client.RequestOption) client.Request {
|
||||
return newHTTPRequest(service, method, req, h.opts.ContentType, opts...)
|
||||
}
|
||||
|
||||
func (h *httpClient) Call(ctx context.Context, req client.Request, rsp interface{}, opts ...client.CallOption) error {
|
||||
@@ -525,7 +552,7 @@ func (h *httpClient) Stream(ctx context.Context, req client.Request, opts ...cli
|
||||
func (h *httpClient) Publish(ctx context.Context, p client.Message, opts ...client.PublishOption) error {
|
||||
options := client.NewPublishOptions(opts...)
|
||||
|
||||
md, ok := metadata.FromContext(ctx)
|
||||
md, ok := metadata.FromOutgoingContext(ctx)
|
||||
if !ok {
|
||||
md = metadata.New(2)
|
||||
}
|
||||
@@ -587,17 +614,31 @@ func NewClient(opts ...client.Option) client.Client {
|
||||
opts: options,
|
||||
}
|
||||
|
||||
dialer, ok := options.Context.Value(httpDialerKey{}).(*net.Dialer)
|
||||
if !ok {
|
||||
dialer = &net.Dialer{
|
||||
Timeout: 30 * time.Second,
|
||||
KeepAlive: 30 * time.Second,
|
||||
}
|
||||
}
|
||||
|
||||
if httpcli, ok := options.Context.Value(httpClientKey{}).(*http.Client); ok {
|
||||
rc.httpcli = httpcli
|
||||
} else {
|
||||
rc.httpcli = http.DefaultClient
|
||||
// TODO customTransport := http.DefaultTransport.(*http.Transport).Clone()
|
||||
tr := &http.Transport{
|
||||
Proxy: http.ProxyFromEnvironment,
|
||||
DialContext: dialer.DialContext,
|
||||
ForceAttemptHTTP2: true,
|
||||
MaxConnsPerHost: 100,
|
||||
MaxIdleConns: 20,
|
||||
IdleConnTimeout: 60 * time.Second,
|
||||
TLSHandshakeTimeout: 10 * time.Second,
|
||||
ExpectContinueTimeout: 1 * time.Second,
|
||||
TLSClientConfig: options.TLSConfig,
|
||||
}
|
||||
if dialer, ok := options.Context.Value(httpDialerKey{}).(*net.Dialer); ok {
|
||||
rc.dialer = dialer
|
||||
} else {
|
||||
rc.dialer = &net.Dialer{}
|
||||
rc.httpcli = &http.Client{Transport: tr}
|
||||
}
|
||||
|
||||
c := client.Client(rc)
|
||||
|
||||
// wrap in reverse
|
||||
|
12
options.go
12
options.go
@@ -1,7 +1,6 @@
|
||||
package http
|
||||
|
||||
import (
|
||||
"crypto/tls"
|
||||
"net"
|
||||
"net/http"
|
||||
|
||||
@@ -33,28 +32,21 @@ type tlsAuth struct{}
|
||||
type maxRecvMsgSizeKey struct{}
|
||||
type maxSendMsgSizeKey struct{}
|
||||
|
||||
// maximum streams on a connectioin
|
||||
// PoolMaxStreams maximum streams on a connectioin
|
||||
func PoolMaxStreams(n int) client.Option {
|
||||
return client.SetOption(poolMaxStreams{}, n)
|
||||
}
|
||||
|
||||
// maximum idle conns of a pool
|
||||
// PoolMaxIdle maximum idle conns of a pool
|
||||
func PoolMaxIdle(d int) client.Option {
|
||||
return client.SetOption(poolMaxIdle{}, d)
|
||||
}
|
||||
|
||||
// AuthTLS should be used to setup a secure authentication using TLS
|
||||
func AuthTLS(t *tls.Config) client.Option {
|
||||
return client.SetOption(tlsAuth{}, t)
|
||||
}
|
||||
|
||||
//
|
||||
// MaxRecvMsgSize set the maximum size of message that client can receive.
|
||||
func MaxRecvMsgSize(s int) client.Option {
|
||||
return client.SetOption(maxRecvMsgSizeKey{}, s)
|
||||
}
|
||||
|
||||
//
|
||||
// MaxSendMsgSize set the maximum size of message that client can send.
|
||||
func MaxSendMsgSize(s int) client.Option {
|
||||
return client.SetOption(maxSendMsgSizeKey{}, s)
|
||||
|
18
util.go
18
util.go
@@ -87,6 +87,10 @@ func newPathRequest(path string, method string, body string, msg interface{}, ta
|
||||
t.name = strings.ToLower(fld.Name)
|
||||
}
|
||||
|
||||
if !val.IsValid() || val.IsZero() {
|
||||
continue
|
||||
}
|
||||
|
||||
if _, ok := fieldsmap[t.name]; ok {
|
||||
fieldsmap[t.name] = fmt.Sprintf("%v", val.Interface())
|
||||
} else if (body == "*" || body == t.name) && method != http.MethodGet {
|
||||
@@ -156,16 +160,24 @@ func newTemplate(path string) (util.Template, error) {
|
||||
}
|
||||
|
||||
func parseRsp(ctx context.Context, hrsp *http.Response, cf codec.Codec, rsp interface{}, opts client.CallOptions) error {
|
||||
// fast path return
|
||||
if hrsp.StatusCode == http.StatusNoContent {
|
||||
return nil
|
||||
}
|
||||
|
||||
b, err := ioutil.ReadAll(hrsp.Body)
|
||||
if err != nil {
|
||||
return errors.InternalServerError("go.micro.client", err.Error())
|
||||
}
|
||||
|
||||
if hrsp.StatusCode < 400 {
|
||||
// unmarshal only if body not nil
|
||||
if len(b) > 0 {
|
||||
// unmarshal
|
||||
if err := cf.Unmarshal(b, rsp); err != nil {
|
||||
return errors.InternalServerError("go.micro.client", err.Error())
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -183,8 +195,12 @@ func parseRsp(ctx context.Context, hrsp *http.Response, cf codec.Codec, rsp inte
|
||||
return errors.New("go.micro.client", string(b), int32(hrsp.StatusCode))
|
||||
}
|
||||
|
||||
if len(b) > 0 {
|
||||
if cerr := cf.Unmarshal(b, err); cerr != nil {
|
||||
return errors.InternalServerError("go.micro.client", cerr.Error())
|
||||
err = errors.InternalServerError("go.micro.client", cerr.Error())
|
||||
}
|
||||
} else {
|
||||
err = errors.New("go.micro.client", string(b), int32(hrsp.StatusCode))
|
||||
}
|
||||
|
||||
return err
|
||||
|
Reference in New Issue
Block a user