Compare commits
11 Commits
Author | SHA1 | Date | |
---|---|---|---|
0eb663fffe | |||
|
fd803b1fe0 | ||
|
4c2178305d | ||
9149aeb3de | |||
dcadd64941 | |||
8207d4154f | |||
d7524cbe01 | |||
0e7b8e73a8 | |||
35146aa717 | |||
3f44a41d30 | |||
090100e522 |
2
go.mod
2
go.mod
@@ -2,4 +2,4 @@ module go.unistack.org/micro-client-http/v3
|
|||||||
|
|
||||||
go 1.18
|
go 1.18
|
||||||
|
|
||||||
require go.unistack.org/micro/v3 v3.10.16
|
require go.unistack.org/micro/v3 v3.10.28
|
||||||
|
10
go.sum
10
go.sum
@@ -1,8 +1,2 @@
|
|||||||
github.com/imdario/mergo v0.3.13/go.mod h1:4lJ1jqUDcsbIECGy0RUJAXNIhg+6ocWgb1ALK2O4oXg=
|
go.unistack.org/micro/v3 v3.10.28 h1:/87lGekrmi0/66pioy+Nh8lVUBBYnVqKoHiNYX5OmMI=
|
||||||
github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ=
|
go.unistack.org/micro/v3 v3.10.28/go.mod h1:eUgtvbtiiz6te93m0ZdmoecbitWwjdBmmr84srmEIKA=
|
||||||
github.com/silas/dag v0.0.0-20211117232152-9d50aa809f35/go.mod h1:7RTUFBdIRC9nZ7/3RyRNH1bdqIShrDejd1YbLwgPS+I=
|
|
||||||
go.unistack.org/micro/v3 v3.10.16 h1:2er/SKKYbV60M+UuJM4eYCF0MZYAIq/yNUrAbTfgq8Q=
|
|
||||||
go.unistack.org/micro/v3 v3.10.16/go.mod h1:uMAc0U/x7dmtICCrblGf0ZLgYegu3VwQAquu+OFCw1Q=
|
|
||||||
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
|
|
||||||
gopkg.in/yaml.v3 v3.0.0/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
|
||||||
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
|
||||||
|
28
http.go
28
http.go
@@ -147,6 +147,11 @@ func newRequest(ctx context.Context, log logger.Logger, addr string, req client.
|
|||||||
if opts.AuthToken != "" {
|
if opts.AuthToken != "" {
|
||||||
header.Set(metadata.HeaderAuthorization, opts.AuthToken)
|
header.Set(metadata.HeaderAuthorization, opts.AuthToken)
|
||||||
}
|
}
|
||||||
|
if opts.RequestMetadata != nil {
|
||||||
|
for k, v := range opts.RequestMetadata {
|
||||||
|
header.Set(k, v)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
if md, ok := metadata.FromOutgoingContext(ctx); ok {
|
if md, ok := metadata.FromOutgoingContext(ctx); ok {
|
||||||
for k, v := range md {
|
for k, v := range md {
|
||||||
@@ -632,6 +637,10 @@ func (h *httpClient) publish(ctx context.Context, ps []client.Message, opts ...c
|
|||||||
if v, ok := os.LookupEnv("MICRO_PROXY"); ok {
|
if v, ok := os.LookupEnv("MICRO_PROXY"); ok {
|
||||||
exchange = v
|
exchange = v
|
||||||
}
|
}
|
||||||
|
// get the exchange
|
||||||
|
if len(options.Exchange) > 0 {
|
||||||
|
exchange = options.Exchange
|
||||||
|
}
|
||||||
|
|
||||||
omd, ok := metadata.FromOutgoingContext(ctx)
|
omd, ok := metadata.FromOutgoingContext(ctx)
|
||||||
if !ok {
|
if !ok {
|
||||||
@@ -643,6 +652,16 @@ func (h *httpClient) publish(ctx context.Context, ps []client.Message, opts ...c
|
|||||||
for _, p := range ps {
|
for _, p := range ps {
|
||||||
md := metadata.Copy(omd)
|
md := metadata.Copy(omd)
|
||||||
md[metadata.HeaderContentType] = p.ContentType()
|
md[metadata.HeaderContentType] = p.ContentType()
|
||||||
|
topic := p.Topic()
|
||||||
|
if len(exchange) > 0 {
|
||||||
|
topic = exchange
|
||||||
|
}
|
||||||
|
md.Set(metadata.HeaderTopic, topic)
|
||||||
|
iter := p.Metadata().Iterator()
|
||||||
|
var k, v string
|
||||||
|
for iter.Next(&k, &v) {
|
||||||
|
md.Set(k, v)
|
||||||
|
}
|
||||||
|
|
||||||
// passed in raw data
|
// passed in raw data
|
||||||
if d, ok := p.Payload().(*codec.Frame); ok {
|
if d, ok := p.Payload().(*codec.Frame); ok {
|
||||||
@@ -661,15 +680,6 @@ func (h *httpClient) publish(ctx context.Context, ps []client.Message, opts ...c
|
|||||||
body = b
|
body = b
|
||||||
}
|
}
|
||||||
|
|
||||||
topic := p.Topic()
|
|
||||||
if len(exchange) > 0 {
|
|
||||||
topic = exchange
|
|
||||||
}
|
|
||||||
|
|
||||||
for k, v := range p.Metadata() {
|
|
||||||
md.Set(k, v)
|
|
||||||
}
|
|
||||||
md.Set(metadata.HeaderTopic, topic)
|
|
||||||
msgs = append(msgs, &broker.Message{Header: md, Body: body})
|
msgs = append(msgs, &broker.Message{Header: md, Body: body})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
60
stream.go
60
stream.go
@@ -2,11 +2,9 @@ package http
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"bufio"
|
"bufio"
|
||||||
"bytes"
|
|
||||||
"context"
|
"context"
|
||||||
"fmt"
|
"fmt"
|
||||||
"io"
|
"io"
|
||||||
"io/ioutil"
|
|
||||||
"net"
|
"net"
|
||||||
"net/http"
|
"net/http"
|
||||||
"sync"
|
"sync"
|
||||||
@@ -121,59 +119,55 @@ func (h *httpStream) Close() error {
|
|||||||
|
|
||||||
func (h *httpStream) parseRsp(ctx context.Context, log logger.Logger, hrsp *http.Response, cf codec.Codec, rsp interface{}, opts client.CallOptions) error {
|
func (h *httpStream) parseRsp(ctx context.Context, log logger.Logger, hrsp *http.Response, cf codec.Codec, rsp interface{}, opts client.CallOptions) error {
|
||||||
var err error
|
var err error
|
||||||
|
var buf []byte
|
||||||
|
|
||||||
|
// fast path return
|
||||||
|
if hrsp.StatusCode == http.StatusNoContent {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
select {
|
select {
|
||||||
case <-ctx.Done():
|
case <-ctx.Done():
|
||||||
err = ctx.Err()
|
err = ctx.Err()
|
||||||
default:
|
default:
|
||||||
// fast path return
|
if hrsp.Body != nil {
|
||||||
if hrsp.StatusCode == http.StatusNoContent {
|
buf, err = io.ReadAll(hrsp.Body)
|
||||||
return nil
|
if err != nil {
|
||||||
|
if log.V(logger.ErrorLevel) {
|
||||||
|
log.Errorf(ctx, "failed to read body: %v", err)
|
||||||
|
}
|
||||||
|
return errors.InternalServerError("go.micro.client", string(buf))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if log.V(logger.DebugLevel) {
|
||||||
|
log.Debugf(ctx, "response %s with %v", buf, hrsp.Header)
|
||||||
}
|
}
|
||||||
|
|
||||||
if hrsp.StatusCode < 400 {
|
if hrsp.StatusCode < 400 {
|
||||||
if log.V(logger.DebugLevel) {
|
if err = cf.Unmarshal(buf, rsp); err != nil {
|
||||||
buf, rerr := io.ReadAll(hrsp.Body)
|
|
||||||
log.Debugf(ctx, "response %s with %v", buf, hrsp.Header)
|
|
||||||
if err != nil {
|
|
||||||
return errors.InternalServerError("go.micro.client", rerr.Error())
|
|
||||||
}
|
|
||||||
hrsp.Body = ioutil.NopCloser(bytes.NewBuffer(buf))
|
|
||||||
}
|
|
||||||
if err = cf.ReadBody(hrsp.Body, rsp); err != nil {
|
|
||||||
return errors.InternalServerError("go.micro.client", err.Error())
|
return errors.InternalServerError("go.micro.client", err.Error())
|
||||||
}
|
}
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
var rerr interface{}
|
||||||
errmap, ok := opts.Context.Value(errorMapKey{}).(map[string]interface{})
|
errmap, ok := opts.Context.Value(errorMapKey{}).(map[string]interface{})
|
||||||
if ok && errmap != nil {
|
if ok && errmap != nil {
|
||||||
if err, ok = errmap[fmt.Sprintf("%d", hrsp.StatusCode)].(error); !ok {
|
if rerr, ok = errmap[fmt.Sprintf("%d", hrsp.StatusCode)].(error); !ok {
|
||||||
err, ok = errmap["default"].(error)
|
rerr, ok = errmap["default"].(error)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if !ok || err == nil {
|
if !ok || rerr == nil {
|
||||||
buf, cerr := io.ReadAll(hrsp.Body)
|
|
||||||
if log.V(logger.DebugLevel) {
|
|
||||||
log.Debugf(ctx, "response %s with %v", buf, hrsp.Header)
|
|
||||||
}
|
|
||||||
if cerr != nil {
|
|
||||||
return errors.InternalServerError("go.micro.client", cerr.Error())
|
|
||||||
}
|
|
||||||
return errors.New("go.micro.client", string(buf), int32(hrsp.StatusCode))
|
return errors.New("go.micro.client", string(buf), int32(hrsp.StatusCode))
|
||||||
}
|
}
|
||||||
|
|
||||||
if log.V(logger.DebugLevel) {
|
if cerr := cf.Unmarshal(buf, rerr); cerr != nil {
|
||||||
buf, rerr := io.ReadAll(hrsp.Body)
|
return errors.InternalServerError("go.micro.client", cerr.Error())
|
||||||
log.Debugf(ctx, "response %s with %v", buf, hrsp.Header)
|
|
||||||
if err != nil {
|
|
||||||
return errors.InternalServerError("go.micro.client", rerr.Error())
|
|
||||||
}
|
|
||||||
hrsp.Body = ioutil.NopCloser(bytes.NewBuffer(buf))
|
|
||||||
}
|
}
|
||||||
|
|
||||||
if cerr := cf.ReadBody(hrsp.Body, err); cerr != nil {
|
if err, ok = rerr.(error); !ok {
|
||||||
err = errors.InternalServerError("go.micro.client", cerr.Error())
|
err = &Error{rerr}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
79
util.go
79
util.go
@@ -1,11 +1,9 @@
|
|||||||
package http
|
package http
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"bytes"
|
|
||||||
"context"
|
"context"
|
||||||
"fmt"
|
"fmt"
|
||||||
"io"
|
"io"
|
||||||
"io/ioutil"
|
|
||||||
"net/http"
|
"net/http"
|
||||||
"net/url"
|
"net/url"
|
||||||
"reflect"
|
"reflect"
|
||||||
@@ -15,6 +13,7 @@ import (
|
|||||||
"go.unistack.org/micro/v3/client"
|
"go.unistack.org/micro/v3/client"
|
||||||
"go.unistack.org/micro/v3/errors"
|
"go.unistack.org/micro/v3/errors"
|
||||||
"go.unistack.org/micro/v3/logger"
|
"go.unistack.org/micro/v3/logger"
|
||||||
|
"go.unistack.org/micro/v3/metadata"
|
||||||
rutil "go.unistack.org/micro/v3/util/reflect"
|
rutil "go.unistack.org/micro/v3/util/reflect"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -218,13 +217,21 @@ func newPathRequest(path string, method string, body string, msg interface{}, ta
|
|||||||
_, _ = b.WriteString(values.Encode())
|
_, _ = b.WriteString(values.Encode())
|
||||||
}
|
}
|
||||||
|
|
||||||
if rutil.IsZero(nmsg) {
|
if rutil.IsZero(nmsg) && !isEmptyStruct(nmsg) {
|
||||||
return b.String(), nil, nil
|
return b.String(), nil, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
return b.String(), nmsg, nil
|
return b.String(), nmsg, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func isEmptyStruct(v interface{}) bool {
|
||||||
|
val := reflect.ValueOf(v)
|
||||||
|
if val.Kind() == reflect.Ptr {
|
||||||
|
val = val.Elem()
|
||||||
|
}
|
||||||
|
return val.Kind() == reflect.Struct && val.NumField() == 0
|
||||||
|
}
|
||||||
|
|
||||||
func newTemplate(path string) ([]string, error) {
|
func newTemplate(path string) ([]string, error) {
|
||||||
if len(path) == 0 || path[0] != '/' {
|
if len(path) == 0 || path[0] != '/' {
|
||||||
return nil, fmt.Errorf("path must starts with /")
|
return nil, fmt.Errorf("path must starts with /")
|
||||||
@@ -247,45 +254,55 @@ func newTemplate(path string) ([]string, error) {
|
|||||||
|
|
||||||
func (h *httpClient) parseRsp(ctx context.Context, hrsp *http.Response, rsp interface{}, opts client.CallOptions) error {
|
func (h *httpClient) parseRsp(ctx context.Context, hrsp *http.Response, rsp interface{}, opts client.CallOptions) error {
|
||||||
var err error
|
var err error
|
||||||
|
var buf []byte
|
||||||
|
|
||||||
|
// fast path return
|
||||||
|
if hrsp.StatusCode == http.StatusNoContent {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
if opts.ResponseMetadata != nil {
|
||||||
|
*opts.ResponseMetadata = metadata.New(len(hrsp.Header))
|
||||||
|
for k, v := range hrsp.Header {
|
||||||
|
opts.ResponseMetadata.Set(k, strings.Join(v, ","))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
select {
|
select {
|
||||||
case <-ctx.Done():
|
case <-ctx.Done():
|
||||||
err = ctx.Err()
|
err = ctx.Err()
|
||||||
default:
|
default:
|
||||||
// fast path return
|
|
||||||
if hrsp.StatusCode == http.StatusNoContent {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
ct := DefaultContentType
|
ct := DefaultContentType
|
||||||
|
|
||||||
if htype := hrsp.Header.Get("Content-Type"); htype != "" {
|
if htype := hrsp.Header.Get("Content-Type"); htype != "" {
|
||||||
ct = htype
|
ct = htype
|
||||||
}
|
}
|
||||||
|
|
||||||
cf, cerr := h.newCodec(ct)
|
if hrsp.Body != nil {
|
||||||
if hrsp.StatusCode >= 400 && cerr != nil {
|
buf, err = io.ReadAll(hrsp.Body)
|
||||||
var buf []byte
|
if err != nil {
|
||||||
if hrsp.Body != nil {
|
if h.opts.Logger.V(logger.ErrorLevel) {
|
||||||
buf, err = io.ReadAll(hrsp.Body)
|
h.opts.Logger.Error(ctx, "failed to read body", err)
|
||||||
if err != nil && h.opts.Logger.V(logger.ErrorLevel) {
|
|
||||||
h.opts.Logger.Errorf(ctx, "failed to read body: %v", err)
|
|
||||||
}
|
}
|
||||||
|
return errors.InternalServerError("go.micro.client", string(buf))
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
cf, cerr := h.newCodec(ct)
|
||||||
|
if cerr != nil {
|
||||||
if h.opts.Logger.V(logger.DebugLevel) {
|
if h.opts.Logger.V(logger.DebugLevel) {
|
||||||
h.opts.Logger.Debugf(ctx, "response %s with %v", buf, hrsp.Header)
|
h.opts.Logger.Debug(ctx, fmt.Sprintf("response with %v unknown content-type %s %s", hrsp.Header, ct, buf))
|
||||||
}
|
|
||||||
// response like text/plain or something else, return original error
|
|
||||||
return errors.New("go.micro.client", string(buf), int32(hrsp.StatusCode))
|
|
||||||
} else if cerr != nil {
|
|
||||||
if h.opts.Logger.V(logger.DebugLevel) {
|
|
||||||
h.opts.Logger.Debugf(ctx, "response with %v unknown content-type", hrsp.Header, ct)
|
|
||||||
}
|
}
|
||||||
return errors.InternalServerError("go.micro.client", cerr.Error())
|
return errors.InternalServerError("go.micro.client", cerr.Error())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if h.opts.Logger.V(logger.DebugLevel) {
|
||||||
|
h.opts.Logger.Debug(ctx, fmt.Sprintf("response %s with %v", buf, hrsp.Header))
|
||||||
|
}
|
||||||
|
|
||||||
// succeseful response
|
// succeseful response
|
||||||
if hrsp.StatusCode < 400 {
|
if hrsp.StatusCode < 400 {
|
||||||
if err = cf.ReadBody(hrsp.Body, 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", err.Error())
|
||||||
}
|
}
|
||||||
return nil
|
return nil
|
||||||
@@ -302,26 +319,10 @@ func (h *httpClient) parseRsp(ctx context.Context, hrsp *http.Response, rsp inte
|
|||||||
}
|
}
|
||||||
|
|
||||||
if !ok || rerr == nil {
|
if !ok || rerr == nil {
|
||||||
buf, rerr := io.ReadAll(hrsp.Body)
|
|
||||||
if h.opts.Logger.V(logger.DebugLevel) {
|
|
||||||
h.opts.Logger.Debugf(ctx, "response %s with %v", buf, hrsp.Header)
|
|
||||||
}
|
|
||||||
if rerr != nil {
|
|
||||||
return errors.InternalServerError("go.micro.client", rerr.Error())
|
|
||||||
}
|
|
||||||
return errors.New("go.micro.client", string(buf), int32(hrsp.StatusCode))
|
return errors.New("go.micro.client", string(buf), int32(hrsp.StatusCode))
|
||||||
}
|
}
|
||||||
|
|
||||||
if h.opts.Logger.V(logger.DebugLevel) {
|
if cerr := cf.Unmarshal(buf, rerr); cerr != nil {
|
||||||
buf, rerr := io.ReadAll(hrsp.Body)
|
|
||||||
h.opts.Logger.Debugf(ctx, "response %s with %v", buf, hrsp.Header)
|
|
||||||
if err != nil {
|
|
||||||
return errors.InternalServerError("go.micro.client", rerr.Error())
|
|
||||||
}
|
|
||||||
hrsp.Body = ioutil.NopCloser(bytes.NewBuffer(buf))
|
|
||||||
}
|
|
||||||
|
|
||||||
if cerr := cf.ReadBody(hrsp.Body, rerr); cerr != nil {
|
|
||||||
return errors.InternalServerError("go.micro.client", cerr.Error())
|
return errors.InternalServerError("go.micro.client", cerr.Error())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
32
util_test.go
32
util_test.go
@@ -59,6 +59,38 @@ func TestNewPathRequest(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestNewPathRequestWithEmptyBody(t *testing.T) {
|
||||||
|
val := struct{}{}
|
||||||
|
cases := []string{
|
||||||
|
"",
|
||||||
|
"*",
|
||||||
|
"{}",
|
||||||
|
"nil",
|
||||||
|
`{"type": "invalid"}`,
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, body := range cases {
|
||||||
|
for _, m := range []string{"POST", "PUT", "PATCH", "GET", "DELETE"} {
|
||||||
|
path, nmsg, err := newPathRequest("/v1/test", m, body, val, []string{"protobuf", "json"}, nil)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if nmsg == nil {
|
||||||
|
t.Fatalf("invalid path: nil nmsg")
|
||||||
|
}
|
||||||
|
|
||||||
|
u, err := url.Parse(path)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
vals := u.Query()
|
||||||
|
if len(vals) != 0 {
|
||||||
|
t.Fatalf("invalid path: %v nmsg: %v", path, nmsg)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestNewPathVarRequest(t *testing.T) {
|
func TestNewPathVarRequest(t *testing.T) {
|
||||||
type Message struct {
|
type Message struct {
|
||||||
Name string `json:"name"`
|
Name string `json:"name"`
|
||||||
|
Reference in New Issue
Block a user