unistack-org/micro-client-http
4
0
Fork 2
Code Issues 3 Pull Requests Actions Packages Projects Releases 62 Wiki Activity

add tag support

Browse Source 
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
This commit is contained in:
Василий Толстов 2021-03-23 00:29:27 +03:00
parent 573c1f50e2
commit fd1acd56cb
6 changed files with 168 additions and 57 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

68
http.go
View File

@ -36,9 +36,11 @@ type httpClient struct {
httpcli *http.Client
}
func newRequest(addr string, req client.Request, cf codec.Codec, msg interface{}, opts client.CallOptions) (*http.Request, error) {
func newRequest(addr string, req client.Request, ct string, cf codec.Codec, msg interface{}, opts client.CallOptions) (*http.Request, error) {
hreq := &http.Request{Method: http.MethodPost}
body := "*" // as like google api http annotation
var tags []string
u, err := url.Parse(addr)
if err != nil {
hreq.URL = &url.URL{
@ -59,6 +61,10 @@ func newRequest(addr string, req client.Request, cf codec.Codec, msg interface{}
if b, ok := opts.Context.Value(bodyKey{}).(string); ok {
body = b
}
if t, ok := opts.Context.Value(structTagsKey{}).([]string); ok && len(t) > 0 {
tags = t
}
}
hreq.URL, err = u.Parse(ep)
if err != nil {
@ -66,7 +72,16 @@ func newRequest(addr string, req client.Request, cf codec.Codec, msg interface{}
}
}
path, nmsg, err := newPathRequest(hreq.URL.Path, hreq.Method, body, msg)
if len(tags) == 0 {
switch ct {
default:
tags = append(tags, "json", "protobuf")
case "text/xml":
tags = append(tags, "xml")
}
}
path, nmsg, err := newPathRequest(hreq.URL.Path, hreq.Method, body, msg, tags)
if err != nil {
return nil, errors.BadRequest("go.micro.client", err.Error())
}
@ -100,18 +115,20 @@ func (h *httpClient) call(ctx context.Context, addr string, req client.Request,
}
}
ct := req.ContentType()
// set timeout in nanoseconds
header.Set("Timeout", fmt.Sprintf("%d", opts.RequestTimeout))
// set the content type for the request
header.Set("Content-Type", req.ContentType())
header.Set("Content-Type", ct)
// get codec
cf, err := h.newCodec(req.ContentType())
cf, err := h.newCodec(ct)
if err != nil {
return errors.InternalServerError("go.micro.client", err.Error())
}
hreq, err := newRequest(addr, req, cf, req.Body(), opts)
hreq, err := newRequest(addr, req, ct, cf, req.Body(), opts)
if err != nil {
return err
}
@ -151,10 +168,11 @@ func (h *httpClient) stream(ctx context.Context, addr string, req client.Request
header = make(http.Header, 2)
}
ct := req.ContentType()
// set timeout in nanoseconds
header.Set("Timeout", fmt.Sprintf("%d", opts.RequestTimeout))
// set the content type for the request
header.Set("Content-Type", req.ContentType())
header.Set("Content-Type", ct)
// get codec
cf, err := h.newCodec(req.ContentType())
@ -178,7 +196,8 @@ func (h *httpClient) stream(ctx context.Context, addr string, req client.Request
closed: make(chan bool),
opts: opts,
conn: cc,
codec: cf,
ct: ct,
cf: cf,
header: header,
reader: bufio.NewReader(cc),
request: req,
@ -560,38 +579,3 @@ func NewClient(opts ...client.Option) client.Client {
return c
}
func parseRsp(ctx context.Context, hrsp *http.Response, cf codec.Codec, rsp interface{}, opts client.CallOptions) error {
b, err := ioutil.ReadAll(hrsp.Body)
if err != nil {
return errors.InternalServerError("go.micro.client", err.Error())
}
if hrsp.StatusCode < 400 {
// unmarshal
if err := cf.Unmarshal(b, rsp); err != nil {
return errors.InternalServerError("go.micro.client", err.Error())
}
return nil
}
errmap, ok := opts.Context.Value(errorMapKey{}).(map[string]interface{})
if !ok || errmap == nil {
// user not provide map of errors
// id: req.Service() ??
return errors.New("go.micro.client", string(b), int32(hrsp.StatusCode))
}
if err, ok = errmap[fmt.Sprintf("%d", hrsp.StatusCode)].(error); !ok {
err, ok = errmap["default"].(error)
}
if !ok {
return errors.New("go.micro.client", string(b), int32(hrsp.StatusCode))
}
if cerr := cf.Unmarshal(b, err); cerr != nil {
return errors.InternalServerError("go.micro.client", cerr.Error())
}
return err
}

4
http_test.go
View File

@ -15,7 +15,7 @@ type Request struct {
func TestValidPath(t *testing.T) {
req := &Request{Name: "vtolstov", Field1: "field1", Field2: "field2", Field3: 10}
p, m, err := newPathRequest("/api/v1/{name}/list", "GET", "", req)
p, m, err := newPathRequest("/api/v1/{name}/list", "GET", "", req, nil)
if err != nil {
t.Fatal(err)
}
@ -32,7 +32,7 @@ func TestValidPath(t *testing.T) {
func TestInvalidPath(t *testing.T) {
req := &Request{Name: "vtolstov", Field1: "field1", Field2: "field2", Field3: 10}
p, m, err := newPathRequest("/api/v1/{xname}/list", "GET", "", req)
p, m, err := newPathRequest("/api/v1/{xname}/list", "GET", "", req, nil)
if err == nil {
t.Fatalf("path param must not be filled")
}

6
options.go
View File

@ -95,3 +95,9 @@ type errorMapKey struct{}
func ErrorMap(m map[string]interface{}) client.CallOption {
return client.SetCallOption(errorMapKey{}, m)
}
type structTagsKey struct{}
func StructTags(tags []string) client.CallOption {
return client.SetCallOption(structTagsKey{}, tags)
}

7
stream.go
View File

@ -18,7 +18,8 @@ type httpStream struct {
sync.RWMutex
address string
opts client.CallOptions
codec codec.Codec
ct string
cf codec.Codec
context context.Context
header http.Header
seq uint64
@ -63,7 +64,7 @@ func (h *httpStream) Send(msg interface{}) error {
return errShutdown
}
hreq, err := newRequest(h.address, h.request, h.codec, msg, h.opts)
hreq, err := newRequest(h.address, h.request, h.ct, h.cf, msg, h.opts)
if err != nil {
return err
}
@ -88,7 +89,7 @@ func (h *httpStream) Recv(msg interface{}) error {
}
defer hrsp.Body.Close()
return parseRsp(h.context, hrsp, h.codec, msg, h.opts)
return parseRsp(h.context, hrsp, h.cf, msg, h.opts)
}
func (h *httpStream) Error() error {

83
util.go
View File

@ -1,12 +1,17 @@
package http
import (
"context"
"fmt"
"io/ioutil"
"net/http"
"reflect"
"strings"
"sync"
"github.com/unistack-org/micro/v3/client"
"github.com/unistack-org/micro/v3/codec"
"github.com/unistack-org/micro/v3/errors"
rutil "github.com/unistack-org/micro/v3/util/reflect"
util "github.com/unistack-org/micro/v3/util/router"
)
@ -16,7 +21,7 @@ var (
mu sync.RWMutex
)
func newPathRequest(path string, method string, body string, msg interface{}) (string, interface{}, error) {
func newPathRequest(path string, method string, body string, msg interface{}, tags []string) (string, interface{}, error) {
// parse via https://github.com/googleapis/googleapis/blob/master/google/api/http.proto definition
tpl, err := newTemplate(path)
if err != nil {
@ -52,13 +57,38 @@ func newPathRequest(path string, method string, body string, msg interface{}) (s
continue
}
fld := tmsg.Type().Field(i)
lfield := strings.ToLower(fld.Name)
if _, ok := fieldsmap[lfield]; ok {
fieldsmap[lfield] = fmt.Sprintf("%v", val.Interface())
} else if (body == "*" || body == lfield) && method != http.MethodGet {
t := &tag{}
for _, tn := range tags {
ts, ok := fld.Tag.Lookup(tn)
if !ok {
continue
}
tp := strings.Split(ts, ",")
// special
switch tn {
case "protobuf": // special
t = &tag{key: tn, name: tp[3][5:], opts: append(tp[:3], tp[4:]...)}
default:
t = &tag{key: tn, name: tp[0], opts: tp[1:]}
}
if t.name != "" {
break
}
}
if t.name == "" {
// fallback to lowercase
t.name = strings.ToLower(fld.Name)
}
if _, ok := fieldsmap[t.name]; ok {
fieldsmap[t.name] = fmt.Sprintf("%v", val.Interface())
} else if (body == "*" || body == t.name) && method != http.MethodGet {
tnmsg.Field(i).Set(val)
} else {
values[lfield] = fmt.Sprintf("%v", val.Interface())
values[t.name] = fmt.Sprintf("%v", val.Interface())
}
}
@ -120,3 +150,44 @@ func newTemplate(path string) (util.Template, error) {
return tpl, nil
}
func parseRsp(ctx context.Context, hrsp *http.Response, cf codec.Codec, rsp interface{}, opts client.CallOptions) error {
b, err := ioutil.ReadAll(hrsp.Body)
if err != nil {
return errors.InternalServerError("go.micro.client", err.Error())
}
if hrsp.StatusCode < 400 {
// unmarshal
if err := cf.Unmarshal(b, rsp); err != nil {
return errors.InternalServerError("go.micro.client", err.Error())
}
return nil
}
errmap, ok := opts.Context.Value(errorMapKey{}).(map[string]interface{})
if !ok || errmap == nil {
// user not provide map of errors
// id: req.Service() ??
return errors.New("go.micro.client", string(b), int32(hrsp.StatusCode))
}
if err, ok = errmap[fmt.Sprintf("%d", hrsp.StatusCode)].(error); !ok {
err, ok = errmap["default"].(error)
}
if !ok {
return errors.New("go.micro.client", string(b), int32(hrsp.StatusCode))
}
if cerr := cf.Unmarshal(b, err); cerr != nil {
return errors.InternalServerError("go.micro.client", cerr.Error())
}
return err
}
type tag struct {
key string
name string
opts []string
}

57
util_test.go
View File

@ -5,10 +5,19 @@ import (
"testing"
)
func TestTemplate(t *testing.T) {
tpl, err := newTemplate("/v1/{ClientID}/list")
if err != nil {
t.Fatal(err)
}
_ = tpl
// fmt.Printf("%#+v\n", tpl.Pool)
}
func TestNewPathRequest(t *testing.T) {
type Message struct {
Name string
Val1 string
Name string `json:"name"`
Val1 string `protobuf:"bytes,1,opt,name=val1,proto3" json:"val1"`
Val2 int64
Val3 []string
}
@ -16,7 +25,8 @@ func TestNewPathRequest(t *testing.T) {
omsg := &Message{Name: "test_name", Val1: "test_val1", Val2: 100, Val3: []string{"slice"}}
for _, m := range []string{"POST", "PUT", "PATCH", "GET", "DELETE"} {
path, nmsg, err := newPathRequest("/v1/test", m, "", omsg)
body := ""
path, nmsg, err := newPathRequest("/v1/test", m, body, omsg, []string{"protobuf", "json"})
if err != nil {
t.Fatal(err)
}
@ -26,7 +36,46 @@ func TestNewPathRequest(t *testing.T) {
}
vals := u.Query()
if v, ok := vals["name"]; !ok || v[0] != "test_name" {
t.Fatalf("invlid path: %v nmsg: %v", path, nmsg)
t.Fatalf("invalid path: %v nmsg: %v", path, nmsg)
}
}
}
func TestNewPathVarRequest(t *testing.T) {
type Message struct {
Name string `json:"name"`
Val1 string `protobuf:"bytes,1,opt,name=val1,proto3" json:"val1"`
Val2 int64
Val3 []string
}
omsg := &Message{Name: "test_name", Val1: "test_val1", Val2: 100, Val3: []string{"slice"}}
for _, m := range []string{"POST", "PUT", "PATCH", "GET", "DELETE"} {
body := ""
if m != "GET" {
body = "*"
}
path, nmsg, err := newPathRequest("/v1/test/{val1}", m, body, omsg, []string{"protobuf", "json"})
if err != nil {
t.Fatal(err)
}
u, err := url.Parse(path)
if err != nil {
t.Fatal(err)
}
if m != "GET" {
if _, ok := nmsg.(*Message); !ok {
t.Fatalf("invalid nmsg: %#+v\n", nmsg)
}
if nmsg.(*Message).Name != "test_name" {
t.Fatalf("invalid nmsg: %v nmsg: %v", path, nmsg)
}
} else {
vals := u.Query()
if v, ok := vals["val2"]; !ok || v[0] != "100" {
t.Fatalf("invalid path: %v nmsg: %v", path, nmsg)
}
}
}
}