Compare commits
2 Commits
Author | SHA1 | Date | |
---|---|---|---|
0b21dd6660 | |||
18eb0d9e5c |
@@ -5,14 +5,14 @@ This plugin is a http client for micro.
|
|||||||
## Overview
|
## Overview
|
||||||
|
|
||||||
The http client wraps `net/http` to provide a robust micro client with service discovery, load balancing and streaming.
|
The http client wraps `net/http` to provide a robust micro client with service discovery, load balancing and streaming.
|
||||||
It complies with the [micro.Client](https://godoc.org/go.unistack.org/micro-client-http/v3#Client) interface.
|
It complies with the [micro.Client](https://godoc.org/go.unistack.org/micro-client-http/v4#Client) interface.
|
||||||
|
|
||||||
## Usage
|
## Usage
|
||||||
|
|
||||||
### Use directly
|
### Use directly
|
||||||
|
|
||||||
```go
|
```go
|
||||||
import "go.unistack.org/micro-client-http/v3"
|
import "go.unistack.org/micro-client-http/v4"
|
||||||
|
|
||||||
service := micro.NewService(
|
service := micro.NewService(
|
||||||
micro.Name("my.service"),
|
micro.Name("my.service"),
|
||||||
|
6
go.mod
6
go.mod
@@ -1,5 +1,5 @@
|
|||||||
module go.unistack.org/micro-client-http/v3
|
module go.unistack.org/micro-client-http/v4
|
||||||
|
|
||||||
go 1.18
|
go 1.19
|
||||||
|
|
||||||
require go.unistack.org/micro/v3 v3.10.28
|
require go.unistack.org/micro/v4 v4.0.1
|
||||||
|
4
go.sum
4
go.sum
@@ -1,2 +1,2 @@
|
|||||||
go.unistack.org/micro/v3 v3.10.28 h1:/87lGekrmi0/66pioy+Nh8lVUBBYnVqKoHiNYX5OmMI=
|
go.unistack.org/micro/v4 v4.0.1 h1:xo1IxbVfgh8i0eY0VeYa3cbb13u5n/Mxnp3FOgWD4Jo=
|
||||||
go.unistack.org/micro/v3 v3.10.28/go.mod h1:eUgtvbtiiz6te93m0ZdmoecbitWwjdBmmr84srmEIKA=
|
go.unistack.org/micro/v4 v4.0.1/go.mod h1:p/J5UcSJjfHsWGT31uKoghQ5rUQZzQJBAFy+Z4+ZVMs=
|
||||||
|
48
http.go
48
http.go
@@ -1,5 +1,5 @@
|
|||||||
// Package http provides a http client
|
// Package http provides a http client
|
||||||
package http // import "go.unistack.org/micro-client-http/v3"
|
package http // import "go.unistack.org/micro-client-http/v4"
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"bufio"
|
"bufio"
|
||||||
@@ -15,14 +15,14 @@ import (
|
|||||||
"sync"
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"go.unistack.org/micro/v3/broker"
|
"go.unistack.org/micro/v4/broker"
|
||||||
"go.unistack.org/micro/v3/client"
|
"go.unistack.org/micro/v4/client"
|
||||||
"go.unistack.org/micro/v3/codec"
|
"go.unistack.org/micro/v4/codec"
|
||||||
"go.unistack.org/micro/v3/errors"
|
"go.unistack.org/micro/v4/errors"
|
||||||
"go.unistack.org/micro/v3/logger"
|
"go.unistack.org/micro/v4/logger"
|
||||||
"go.unistack.org/micro/v3/metadata"
|
"go.unistack.org/micro/v4/metadata"
|
||||||
"go.unistack.org/micro/v3/selector"
|
"go.unistack.org/micro/v4/selector"
|
||||||
rutil "go.unistack.org/micro/v3/util/reflect"
|
rutil "go.unistack.org/micro/v4/util/reflect"
|
||||||
)
|
)
|
||||||
|
|
||||||
var DefaultContentType = "application/json"
|
var DefaultContentType = "application/json"
|
||||||
@@ -147,11 +147,6 @@ 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 {
|
||||||
@@ -217,7 +212,7 @@ func newRequest(ctx context.Context, log logger.Logger, addr string, req client.
|
|||||||
}
|
}
|
||||||
|
|
||||||
if log.V(logger.DebugLevel) {
|
if log.V(logger.DebugLevel) {
|
||||||
log.Debug(ctx, fmt.Sprintf("request %s to %s with headers %v body %s", method, u.String(), hreq.Header, b))
|
log.Debugf(ctx, "request %s to %s with headers %v body %s", method, u.String(), hreq.Header, b)
|
||||||
}
|
}
|
||||||
|
|
||||||
return hreq, nil
|
return hreq, nil
|
||||||
@@ -637,10 +632,6 @@ 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 {
|
||||||
@@ -652,16 +643,6 @@ 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 {
|
||||||
@@ -680,6 +661,15 @@ 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})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@@ -1,8 +1,8 @@
|
|||||||
package http
|
package http
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"go.unistack.org/micro/v3/client"
|
"go.unistack.org/micro/v4/client"
|
||||||
"go.unistack.org/micro/v3/metadata"
|
"go.unistack.org/micro/v4/metadata"
|
||||||
)
|
)
|
||||||
|
|
||||||
type httpMessage struct {
|
type httpMessage struct {
|
||||||
|
@@ -4,8 +4,8 @@ import (
|
|||||||
"net"
|
"net"
|
||||||
"net/http"
|
"net/http"
|
||||||
|
|
||||||
"go.unistack.org/micro/v3/client"
|
"go.unistack.org/micro/v4/client"
|
||||||
"go.unistack.org/micro/v3/metadata"
|
"go.unistack.org/micro/v4/metadata"
|
||||||
)
|
)
|
||||||
|
|
||||||
var (
|
var (
|
||||||
|
@@ -1,8 +1,8 @@
|
|||||||
package http
|
package http
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"go.unistack.org/micro/v3/client"
|
"go.unistack.org/micro/v4/client"
|
||||||
"go.unistack.org/micro/v3/codec"
|
"go.unistack.org/micro/v4/codec"
|
||||||
)
|
)
|
||||||
|
|
||||||
type httpRequest struct {
|
type httpRequest struct {
|
||||||
|
@@ -9,10 +9,10 @@ import (
|
|||||||
"net/http"
|
"net/http"
|
||||||
"sync"
|
"sync"
|
||||||
|
|
||||||
"go.unistack.org/micro/v3/client"
|
"go.unistack.org/micro/v4/client"
|
||||||
"go.unistack.org/micro/v3/codec"
|
"go.unistack.org/micro/v4/codec"
|
||||||
"go.unistack.org/micro/v3/errors"
|
"go.unistack.org/micro/v4/errors"
|
||||||
"go.unistack.org/micro/v3/logger"
|
"go.unistack.org/micro/v4/logger"
|
||||||
)
|
)
|
||||||
|
|
||||||
// Implements the streamer interface
|
// Implements the streamer interface
|
||||||
|
32
util.go
32
util.go
@@ -10,11 +10,10 @@ import (
|
|||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
|
|
||||||
"go.unistack.org/micro/v3/client"
|
"go.unistack.org/micro/v4/client"
|
||||||
"go.unistack.org/micro/v3/errors"
|
"go.unistack.org/micro/v4/errors"
|
||||||
"go.unistack.org/micro/v3/logger"
|
"go.unistack.org/micro/v4/logger"
|
||||||
"go.unistack.org/micro/v3/metadata"
|
rutil "go.unistack.org/micro/v4/util/reflect"
|
||||||
rutil "go.unistack.org/micro/v3/util/reflect"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
var (
|
var (
|
||||||
@@ -217,21 +216,13 @@ func newPathRequest(path string, method string, body string, msg interface{}, ta
|
|||||||
_, _ = b.WriteString(values.Encode())
|
_, _ = b.WriteString(values.Encode())
|
||||||
}
|
}
|
||||||
|
|
||||||
if rutil.IsZero(nmsg) && !isEmptyStruct(nmsg) {
|
if rutil.IsZero(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 /")
|
||||||
@@ -261,13 +252,6 @@ func (h *httpClient) parseRsp(ctx context.Context, hrsp *http.Response, rsp inte
|
|||||||
return nil
|
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()
|
||||||
@@ -282,7 +266,7 @@ func (h *httpClient) parseRsp(ctx context.Context, hrsp *http.Response, rsp inte
|
|||||||
buf, err = io.ReadAll(hrsp.Body)
|
buf, err = io.ReadAll(hrsp.Body)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
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.Errorf(ctx, "failed to read body: %v", err)
|
||||||
}
|
}
|
||||||
return errors.InternalServerError("go.micro.client", string(buf))
|
return errors.InternalServerError("go.micro.client", string(buf))
|
||||||
}
|
}
|
||||||
@@ -291,13 +275,13 @@ func (h *httpClient) parseRsp(ctx context.Context, hrsp *http.Response, rsp inte
|
|||||||
cf, cerr := h.newCodec(ct)
|
cf, cerr := h.newCodec(ct)
|
||||||
if cerr != nil {
|
if cerr != nil {
|
||||||
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.Debugf(ctx, "response with %v unknown content-type %s", hrsp.Header, ct, buf)
|
||||||
}
|
}
|
||||||
return errors.InternalServerError("go.micro.client", cerr.Error())
|
return errors.InternalServerError("go.micro.client", cerr.Error())
|
||||||
}
|
}
|
||||||
|
|
||||||
if h.opts.Logger.V(logger.DebugLevel) {
|
if h.opts.Logger.V(logger.DebugLevel) {
|
||||||
h.opts.Logger.Debug(ctx, fmt.Sprintf("response %s with %v", buf, hrsp.Header))
|
h.opts.Logger.Debugf(ctx, "response %s with %v", buf, hrsp.Header)
|
||||||
}
|
}
|
||||||
|
|
||||||
// succeseful response
|
// succeseful response
|
||||||
|
32
util_test.go
32
util_test.go
@@ -59,38 +59,6 @@ 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