Compare commits
10 Commits
Author | SHA1 | Date | |
---|---|---|---|
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
|
go 1.16
|
||||||
|
|
||||||
require github.com/unistack-org/micro/v3 v3.3.0
|
require github.com/unistack-org/micro/v3 v3.3.11
|
||||||
|
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/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/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/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.11 h1:Jr0gAw5lLqgddiHKQeWUOUeP6ZqgRhz52EA9zJ5MJ3U=
|
||||||
github.com/unistack-org/micro/v3 v3.3.0/go.mod h1:iJwCWq2PECMxigfqe6TPC5GLWvj6P94Kk+PTVZGL3w8=
|
github.com/unistack-org/micro/v3 v3.3.11/go.mod h1:PPCt675o3HPcODFbJ4iRWPmQFAk1WQ+asQSOb/syq6U=
|
||||||
golang.org/x/net v0.0.0-20210316092652-d523dce5a7f4/go.mod h1:RBQZq4jEuRlivfhVLdyRGr576XBO4/greRjx4P4O3yc=
|
golang.org/x/net v0.0.0-20210331212208-0fccb6fa2b5c/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
|
||||||
golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
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/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/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
|
||||||
golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
|
golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
|
||||||
|
72
http.go
72
http.go
@@ -19,6 +19,7 @@ import (
|
|||||||
"github.com/unistack-org/micro/v3/errors"
|
"github.com/unistack-org/micro/v3/errors"
|
||||||
"github.com/unistack-org/micro/v3/metadata"
|
"github.com/unistack-org/micro/v3/metadata"
|
||||||
"github.com/unistack-org/micro/v3/router"
|
"github.com/unistack-org/micro/v3/router"
|
||||||
|
rutil "github.com/unistack-org/micro/v3/util/reflect"
|
||||||
)
|
)
|
||||||
|
|
||||||
var (
|
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())
|
return nil, errors.BadRequest("go.micro.client", err.Error())
|
||||||
}
|
}
|
||||||
|
|
||||||
// marshal request is struct not empty
|
var b []byte
|
||||||
|
|
||||||
if nmsg != nil {
|
if nmsg != nil {
|
||||||
var b []byte
|
if ct == "application/x-www-form-urlencoded" {
|
||||||
b, err = cf.Marshal(nmsg)
|
data, err := rutil.StructURLValues(nmsg, "", tags)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, errors.BadRequest("go.micro.client", err.Error())
|
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.Body = ioutil.NopCloser(bytes.NewBuffer(b))
|
||||||
hreq.ContentLength = int64(len(b))
|
hreq.ContentLength = int64(len(b))
|
||||||
}
|
}
|
||||||
@@ -114,7 +126,7 @@ 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 {
|
func (h *httpClient) call(ctx context.Context, addr string, req client.Request, rsp interface{}, opts client.CallOptions) error {
|
||||||
header := make(http.Header, 2)
|
header := make(http.Header, 2)
|
||||||
if md, ok := metadata.FromContext(ctx); ok {
|
if md, ok := metadata.FromOutgoingContext(ctx); ok {
|
||||||
for k, v := range md {
|
for k, v := range md {
|
||||||
header.Set(k, v)
|
header.Set(k, v)
|
||||||
}
|
}
|
||||||
@@ -127,8 +139,16 @@ func (h *httpClient) call(ctx context.Context, addr string, req client.Request,
|
|||||||
// set the content type for the request
|
// set the content type for the request
|
||||||
header.Set("Content-Type", ct)
|
header.Set("Content-Type", ct)
|
||||||
|
|
||||||
|
var cf codec.Codec
|
||||||
|
var err error
|
||||||
// get codec
|
// 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 {
|
if err != nil {
|
||||||
return errors.InternalServerError("go.micro.client", err.Error())
|
return errors.InternalServerError("go.micro.client", err.Error())
|
||||||
}
|
}
|
||||||
@@ -164,7 +184,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) {
|
func (h *httpClient) stream(ctx context.Context, addr string, req client.Request, opts client.CallOptions) (client.Stream, error) {
|
||||||
var header http.Header
|
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)
|
header = make(http.Header, len(md)+2)
|
||||||
for k, v := range md {
|
for k, v := range md {
|
||||||
header.Set(k, v)
|
header.Set(k, v)
|
||||||
@@ -190,7 +210,7 @@ func (h *httpClient) stream(ctx context.Context, addr string, req client.Request
|
|||||||
if err == nil && u.Scheme != "" && u.Host != "" {
|
if err == nil && u.Scheme != "" && u.Host != "" {
|
||||||
dialAddr = 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 {
|
if err != nil {
|
||||||
return nil, errors.InternalServerError("go.micro.client", fmt.Sprintf("Error dialing: %v", err))
|
return nil, errors.InternalServerError("go.micro.client", fmt.Sprintf("Error dialing: %v", err))
|
||||||
}
|
}
|
||||||
@@ -255,8 +275,8 @@ func (h *httpClient) NewMessage(topic string, msg interface{}, opts ...client.Me
|
|||||||
return newHTTPMessage(topic, msg, h.opts.ContentType, opts...)
|
return newHTTPMessage(topic, msg, h.opts.ContentType, opts...)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (h *httpClient) NewRequest(service, method string, req interface{}, reqOpts ...client.RequestOption) client.Request {
|
func (h *httpClient) NewRequest(service, method string, req interface{}, opts ...client.RequestOption) client.Request {
|
||||||
return newHTTPRequest(service, method, req, h.opts.ContentType, reqOpts...)
|
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 {
|
func (h *httpClient) Call(ctx context.Context, req client.Request, rsp interface{}, opts ...client.CallOption) error {
|
||||||
@@ -525,7 +545,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 {
|
func (h *httpClient) Publish(ctx context.Context, p client.Message, opts ...client.PublishOption) error {
|
||||||
options := client.NewPublishOptions(opts...)
|
options := client.NewPublishOptions(opts...)
|
||||||
|
|
||||||
md, ok := metadata.FromContext(ctx)
|
md, ok := metadata.FromOutgoingContext(ctx)
|
||||||
if !ok {
|
if !ok {
|
||||||
md = metadata.New(2)
|
md = metadata.New(2)
|
||||||
}
|
}
|
||||||
@@ -587,17 +607,31 @@ func NewClient(opts ...client.Option) client.Client {
|
|||||||
opts: options,
|
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 {
|
if httpcli, ok := options.Context.Value(httpClientKey{}).(*http.Client); ok {
|
||||||
rc.httpcli = httpcli
|
rc.httpcli = httpcli
|
||||||
} else {
|
} 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,
|
||||||
|
}
|
||||||
|
rc.httpcli = &http.Client{Transport: tr}
|
||||||
}
|
}
|
||||||
if dialer, ok := options.Context.Value(httpDialerKey{}).(*net.Dialer); ok {
|
|
||||||
rc.dialer = dialer
|
|
||||||
} else {
|
|
||||||
rc.dialer = &net.Dialer{}
|
|
||||||
}
|
|
||||||
|
|
||||||
c := client.Client(rc)
|
c := client.Client(rc)
|
||||||
|
|
||||||
// wrap in reverse
|
// wrap in reverse
|
||||||
|
12
options.go
12
options.go
@@ -1,7 +1,6 @@
|
|||||||
package http
|
package http
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"crypto/tls"
|
|
||||||
"net"
|
"net"
|
||||||
"net/http"
|
"net/http"
|
||||||
|
|
||||||
@@ -33,28 +32,21 @@ type tlsAuth struct{}
|
|||||||
type maxRecvMsgSizeKey struct{}
|
type maxRecvMsgSizeKey struct{}
|
||||||
type maxSendMsgSizeKey struct{}
|
type maxSendMsgSizeKey struct{}
|
||||||
|
|
||||||
// maximum streams on a connectioin
|
// PoolMaxStreams maximum streams on a connectioin
|
||||||
func PoolMaxStreams(n int) client.Option {
|
func PoolMaxStreams(n int) client.Option {
|
||||||
return client.SetOption(poolMaxStreams{}, n)
|
return client.SetOption(poolMaxStreams{}, n)
|
||||||
}
|
}
|
||||||
|
|
||||||
// maximum idle conns of a pool
|
// PoolMaxIdle maximum idle conns of a pool
|
||||||
func PoolMaxIdle(d int) client.Option {
|
func PoolMaxIdle(d int) client.Option {
|
||||||
return client.SetOption(poolMaxIdle{}, d)
|
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.
|
// MaxRecvMsgSize set the maximum size of message that client can receive.
|
||||||
func MaxRecvMsgSize(s int) client.Option {
|
func MaxRecvMsgSize(s int) client.Option {
|
||||||
return client.SetOption(maxRecvMsgSizeKey{}, s)
|
return client.SetOption(maxRecvMsgSizeKey{}, s)
|
||||||
}
|
}
|
||||||
|
|
||||||
//
|
|
||||||
// MaxSendMsgSize set the maximum size of message that client can send.
|
// MaxSendMsgSize set the maximum size of message that client can send.
|
||||||
func MaxSendMsgSize(s int) client.Option {
|
func MaxSendMsgSize(s int) client.Option {
|
||||||
return client.SetOption(maxSendMsgSizeKey{}, s)
|
return client.SetOption(maxSendMsgSizeKey{}, s)
|
||||||
|
4
util.go
4
util.go
@@ -87,6 +87,10 @@ func newPathRequest(path string, method string, body string, msg interface{}, ta
|
|||||||
t.name = strings.ToLower(fld.Name)
|
t.name = strings.ToLower(fld.Name)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if !val.IsValid() || val.IsZero() {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
if _, ok := fieldsmap[t.name]; ok {
|
if _, ok := fieldsmap[t.name]; ok {
|
||||||
fieldsmap[t.name] = fmt.Sprintf("%v", val.Interface())
|
fieldsmap[t.name] = fmt.Sprintf("%v", val.Interface())
|
||||||
} else if (body == "*" || body == t.name) && method != http.MethodGet {
|
} else if (body == "*" || body == t.name) && method != http.MethodGet {
|
||||||
|
Reference in New Issue
Block a user