unistack-org/micro
4
0
Fork 2
Code Issues 18 Pull Requests Actions Packages Projects Releases 324 Wiki Activity

major codec upgrade

Browse Source 
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
This commit is contained in:
Василий Толстов
2020-11-23 16:18:47 +03:00
parent daffa9e548
commit c9049c3845
30 changed files with 196 additions and 1004 deletions
Download Patch File Download Diff File Expand all files Collapse all files

495
api/handler/rpc/rpc.go
View File

@@ -1,495 +0,0 @@
// Package rpc is a go-micro rpc handler.
package rpc
import (
"encoding/json"
"io"
"net/http"
"strconv"
"strings"
jsonpatch "github.com/evanphx/json-patch/v5"
"github.com/oxtoacart/bpool"
jsonrpc "github.com/unistack-org/micro-codec-jsonrpc"
protorpc "github.com/unistack-org/micro-codec-protorpc"
"github.com/unistack-org/micro/v3/api"
"github.com/unistack-org/micro/v3/api/handler"
"github.com/unistack-org/micro/v3/api/internal/proto"
"github.com/unistack-org/micro/v3/client"
"github.com/unistack-org/micro/v3/codec"
"github.com/unistack-org/micro/v3/errors"
"github.com/unistack-org/micro/v3/logger"
"github.com/unistack-org/micro/v3/metadata"
"github.com/unistack-org/micro/v3/util/ctx"
"github.com/unistack-org/micro/v3/util/qson"
"github.com/unistack-org/micro/v3/util/router"
)
const (
Handler = "rpc"
)
var (
// supported json codecs
jsonCodecs = []string{
"application/grpc+json",
"application/json",
"application/json-rpc",
}
// support proto codecs
protoCodecs = []string{
"application/grpc",
"application/grpc+proto",
"application/proto",
"application/protobuf",
"application/proto-rpc",
"application/octet-stream",
}
bufferPool = bpool.NewSizedBufferPool(1024, 8)
)
type rpcHandler struct {
opts handler.Options
s *api.Service
}
type buffer struct {
io.ReadCloser
}
func (b *buffer) Write(_ []byte) (int, error) {
return 0, nil
}
func (h *rpcHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
bsize := handler.DefaultMaxRecvSize
if h.opts.MaxRecvSize > 0 {
bsize = h.opts.MaxRecvSize
}
r.Body = http.MaxBytesReader(w, r.Body, bsize)
defer r.Body.Close()
var service *api.Service
if h.s != nil {
// we were given the service
service = h.s
} else if h.opts.Router != nil {
// try get service from router
s, err := h.opts.Router.Route(r)
if err != nil {
writeError(w, r, errors.InternalServerError("go.micro.api", err.Error()))
return
}
service = s
} else {
// we have no way of routing the request
writeError(w, r, errors.InternalServerError("go.micro.api", "no route found"))
return
}
ct := r.Header.Get("Content-Type")
// Strip charset from Content-Type (like `application/json; charset=UTF-8`)
if idx := strings.IndexRune(ct, ';'); idx >= 0 {
ct = ct[:idx]
}
// micro client
c := h.opts.Client
// create context
cx := ctx.FromRequest(r)
// set merged context to request
*r = *r.Clone(cx)
// if stream we currently only support json
if isStream(r, service) {
serveWebsocket(cx, w, r, service, c)
return
}
// create custom router
callOpt := client.WithRouter(router.New(service.Services))
// walk the standard call path
// get payload
br, err := requestPayload(r)
if err != nil {
writeError(w, r, err)
return
}
var rsp []byte
switch {
// proto codecs
case hasCodec(ct, protoCodecs):
request := &proto.Message{}
// if the extracted payload isn't empty lets use it
if len(br) > 0 {
request = proto.NewMessage(br)
}
// create request/response
response := &proto.Message{}
req := c.NewRequest(
service.Name,
service.Endpoint.Name,
request,
client.WithContentType(ct),
)
// make the call
if err := c.Call(cx, req, response, callOpt); err != nil {
writeError(w, r, err)
return
}
// marshall response
rsp, err = response.Marshal()
if err != nil {
writeError(w, r, err)
return
}
default:
// if json codec is not present set to json
if !hasCodec(ct, jsonCodecs) {
ct = "application/json"
}
// default to trying json
var request json.RawMessage
// if the extracted payload isn't empty lets use it
if len(br) > 0 {
request = json.RawMessage(br)
}
// create request/response
var response json.RawMessage
req := c.NewRequest(
service.Name,
service.Endpoint.Name,
&request,
client.WithContentType(ct),
)
// make the call
if err := c.Call(cx, req, &response, callOpt); err != nil {
writeError(w, r, err)
return
}
// marshall response
rsp, err = response.MarshalJSON()
if err != nil {
writeError(w, r, err)
return
}
}
// write the response
writeResponse(w, r, rsp)
}
func (rh *rpcHandler) String() string {
return "rpc"
}
func hasCodec(ct string, codecs []string) bool {
for _, codec := range codecs {
if ct == codec {
return true
}
}
return false
}
// requestPayload takes a *http.Request.
// If the request is a GET the query string parameters are extracted and marshaled to JSON and the raw bytes are returned.
// If the request method is a POST the request body is read and returned
func requestPayload(r *http.Request) ([]byte, error) {
var err error
// we have to decode json-rpc and proto-rpc because we suck
// well actually because there's no proxy codec right now
ct := r.Header.Get("Content-Type")
switch {
case strings.Contains(ct, "application/json-rpc"):
msg := codec.Message{
Type: codec.Request,
Header: metadata.New(0),
}
c := jsonrpc.NewCodec(&buffer{r.Body})
if err = c.ReadHeader(&msg, codec.Request); err != nil {
return nil, err
}
var raw json.RawMessage
if err = c.ReadBody(&raw); err != nil {
return nil, err
}
return ([]byte)(raw), nil
case strings.Contains(ct, "application/proto-rpc"), strings.Contains(ct, "application/octet-stream"):
msg := codec.Message{
Type: codec.Request,
Header: metadata.New(0),
}
c := protorpc.NewCodec(&buffer{r.Body})
if err = c.ReadHeader(&msg, codec.Request); err != nil {
return nil, err
}
var raw proto.Message
if err = c.ReadBody(&raw); err != nil {
return nil, err
}
return raw.Marshal()
case strings.Contains(ct, "application/www-x-form-urlencoded"):
if err = r.ParseForm(); err != nil {
return nil, err
}
// generate a new set of values from the form
vals := make(map[string]string, len(r.Form))
for k, v := range r.Form {
vals[k] = strings.Join(v, ",")
}
// marshal
return json.Marshal(vals)
// TODO: application/grpc
}
// otherwise as per usual
ctx := r.Context()
// dont user metadata.FromContext as it mangles names
md, ok := metadata.FromContext(ctx)
if !ok {
md = metadata.New(0)
}
// allocate maximum
matches := make(map[string]interface{}, len(md))
bodydst := ""
// get fields from url path
for k, v := range md {
k = strings.ToLower(k)
// filter own keys
if strings.HasPrefix(k, "x-api-field-") {
matches[strings.TrimPrefix(k, "x-api-field-")] = v
delete(md, k)
} else if k == "x-api-body" {
bodydst = v
delete(md, k)
}
}
// map of all fields
req := make(map[string]interface{}, len(md))
// get fields from url values
if len(r.URL.RawQuery) > 0 {
umd := make(map[string]interface{})
err = qson.Unmarshal(&umd, r.URL.RawQuery)
if err != nil {
return nil, err
}
for k, v := range umd {
matches[k] = v
}
}
// restore context without fields
*r = *r.Clone(metadata.NewContext(ctx, md))
for k, v := range matches {
ps := strings.Split(k, ".")
if len(ps) == 1 {
req[k] = v
continue
}
em := make(map[string]interface{})
em[ps[len(ps)-1]] = v
for i := len(ps) - 2; i > 0; i-- {
nm := make(map[string]interface{})
nm[ps[i]] = em
em = nm
}
if vm, ok := req[ps[0]]; ok {
// nested map
nm := vm.(map[string]interface{})
for vk, vv := range em {
nm[vk] = vv
}
req[ps[0]] = nm
} else {
req[ps[0]] = em
}
}
pathbuf := []byte("{}")
if len(req) > 0 {
pathbuf, err = json.Marshal(req)
if err != nil {
return nil, err
}
}
urlbuf := []byte("{}")
out, err := jsonpatch.MergeMergePatches(urlbuf, pathbuf)
if err != nil {
return nil, err
}
switch r.Method {
case "GET":
// empty response
if strings.Contains(ct, "application/json") && string(out) == "{}" {
return out, nil
} else if string(out) == "{}" && !strings.Contains(ct, "application/json") {
return []byte{}, nil
}
return out, nil
case "PATCH", "POST", "PUT", "DELETE":
bodybuf := []byte("{}")
buf := bufferPool.Get()
defer bufferPool.Put(buf)
if _, err := buf.ReadFrom(r.Body); err != nil {
return nil, err
}
if b := buf.Bytes(); len(b) > 0 {
bodybuf = b
}
if bodydst == "" || bodydst == "*" {
if out, err = jsonpatch.MergeMergePatches(out, bodybuf); err == nil {
return out, nil
}
}
var jsonbody map[string]interface{}
if json.Valid(bodybuf) {
if err = json.Unmarshal(bodybuf, &jsonbody); err != nil {
return nil, err
}
}
dstmap := make(map[string]interface{})
ps := strings.Split(bodydst, ".")
if len(ps) == 1 {
if jsonbody != nil {
dstmap[ps[0]] = jsonbody
} else {
// old unexpected behaviour
dstmap[ps[0]] = bodybuf
}
} else {
em := make(map[string]interface{})
if jsonbody != nil {
em[ps[len(ps)-1]] = jsonbody
} else {
// old unexpected behaviour
em[ps[len(ps)-1]] = bodybuf
}
for i := len(ps) - 2; i > 0; i-- {
nm := make(map[string]interface{})
nm[ps[i]] = em
em = nm
}
dstmap[ps[0]] = em
}
bodyout, err := json.Marshal(dstmap)
if err != nil {
return nil, err
}
if out, err = jsonpatch.MergeMergePatches(out, bodyout); err == nil {
return out, nil
}
//fallback to previous unknown behaviour
return bodybuf, nil
}
return []byte{}, nil
}
func writeError(w http.ResponseWriter, r *http.Request, err error) {
ce := errors.Parse(err.Error())
switch ce.Code {
case 0:
// assuming it's totally screwed
ce.Code = 500
ce.Id = "go.micro.api"
ce.Status = http.StatusText(500)
ce.Detail = "error during request: " + ce.Detail
w.WriteHeader(500)
default:
w.WriteHeader(int(ce.Code))
}
// response content type
w.Header().Set("Content-Type", "application/json")
// Set trailers
if strings.Contains(r.Header.Get("Content-Type"), "application/grpc") {
w.Header().Set("Trailer", "grpc-status")
w.Header().Set("Trailer", "grpc-message")
w.Header().Set("grpc-status", "13")
w.Header().Set("grpc-message", ce.Detail)
}
_, werr := w.Write([]byte(ce.Error()))
if werr != nil {
if logger.V(logger.ErrorLevel) {
logger.Error(werr.Error())
}
}
}
func writeResponse(w http.ResponseWriter, r *http.Request, rsp []byte) {
w.Header().Set("Content-Type", r.Header.Get("Content-Type"))
w.Header().Set("Content-Length", strconv.Itoa(len(rsp)))
// Set trailers
if strings.Contains(r.Header.Get("Content-Type"), "application/grpc") {
w.Header().Set("Trailer", "grpc-status")
w.Header().Set("Trailer", "grpc-message")
w.Header().Set("grpc-status", "0")
w.Header().Set("grpc-message", "")
}
// write 204 status if rsp is nil
if len(rsp) == 0 {
w.WriteHeader(http.StatusNoContent)
}
// write response
_, err := w.Write(rsp)
if err != nil {
if logger.V(logger.ErrorLevel) {
logger.Error(err.Error())
}
}
}
func NewHandler(opts ...handler.Option) handler.Handler {
options := handler.NewOptions(opts...)
return &rpcHandler{
opts: options,
}
}
func WithService(s *api.Service, opts ...handler.Option) handler.Handler {
options := handler.NewOptions(opts...)
return &rpcHandler{
opts: options,
s: s,
}
}

112
api/handler/rpc/rpc_test.go
View File

@@ -1,112 +0,0 @@
package rpc
import (
"bytes"
"net/http"
"testing"
go_api "github.com/unistack-org/micro/v3/api/proto"
jsonpb "google.golang.org/protobuf/encoding/protojson"
"google.golang.org/protobuf/proto"
)
func TestRequestPayloadFromRequest(t *testing.T) {
// our test event so that we can validate serialising / deserializing of true protos works
protoEvent := go_api.Event{
Name: "Test",
}
protoBytes, err := proto.Marshal(&protoEvent)
if err != nil {
t.Fatal("Failed to marshal proto", err)
}
jsonBytes, err := jsonpb.Marshal(&protoEvent)
if err != nil {
t.Fatal("Failed to marshal proto to JSON ", err)
}
jsonUrlBytes := []byte(`{"key1":"val1","key2":"val2","name":"Test"}`)
t.Run("extracting a json from a POST request with url params", func(t *testing.T) {
r, err := http.NewRequest("POST", "http://localhost/my/path?key1=val1&key2=val2", bytes.NewReader(jsonBytes))
if err != nil {
t.Fatalf("Failed to created http.Request: %v", err)
}
extByte, err := requestPayload(r)
if err != nil {
t.Fatalf("Failed to extract payload from request: %v", err)
}
if string(extByte) != string(jsonUrlBytes) {
t.Fatalf("Expected %v and %v to match", string(extByte), jsonUrlBytes)
}
})
t.Run("extracting a proto from a POST request", func(t *testing.T) {
r, err := http.NewRequest("POST", "http://localhost/my/path", bytes.NewReader(protoBytes))
if err != nil {
t.Fatalf("Failed to created http.Request: %v", err)
}
extByte, err := requestPayload(r)
if err != nil {
t.Fatalf("Failed to extract payload from request: %v", err)
}
if string(extByte) != string(protoBytes) {
t.Fatalf("Expected %v and %v to match", string(extByte), string(protoBytes))
}
})
t.Run("extracting JSON from a POST request", func(t *testing.T) {
r, err := http.NewRequest("POST", "http://localhost/my/path", bytes.NewReader(jsonBytes))
if err != nil {
t.Fatalf("Failed to created http.Request: %v", err)
}
extByte, err := requestPayload(r)
if err != nil {
t.Fatalf("Failed to extract payload from request: %v", err)
}
if string(extByte) != string(jsonBytes) {
t.Fatalf("Expected %v and %v to match", string(extByte), string(jsonBytes))
}
})
t.Run("extracting params from a GET request", func(t *testing.T) {
r, err := http.NewRequest("GET", "http://localhost/my/path", nil)
if err != nil {
t.Fatalf("Failed to created http.Request: %v", err)
}
q := r.URL.Query()
q.Add("name", "Test")
r.URL.RawQuery = q.Encode()
extByte, err := requestPayload(r)
if err != nil {
t.Fatalf("Failed to extract payload from request: %v", err)
}
if string(extByte) != string(jsonBytes) {
t.Fatalf("Expected %v and %v to match", string(extByte), string(jsonBytes))
}
})
t.Run("GET request with no params", func(t *testing.T) {
r, err := http.NewRequest("GET", "http://localhost/my/path", nil)
if err != nil {
t.Fatalf("Failed to created http.Request: %v", err)
}
extByte, err := requestPayload(r)
if err != nil {
t.Fatalf("Failed to extract payload from request: %v", err)
}
if string(extByte) != "" {
t.Fatalf("Expected %v and %v to match", string(extByte), "")
}
})
}

263
api/handler/rpc/stream.go
View File

@@ -1,263 +0,0 @@
package rpc
import (
"bytes"
"context"
"encoding/json"
"io"
"net/http"
"strings"
"time"
"github.com/gobwas/httphead"
"github.com/gobwas/ws"
"github.com/gobwas/ws/wsutil"
raw "github.com/unistack-org/micro-codec-bytes"
"github.com/unistack-org/micro/v3/api"
"github.com/unistack-org/micro/v3/client"
"github.com/unistack-org/micro/v3/logger"
"github.com/unistack-org/micro/v3/util/router"
)
// serveWebsocket will stream rpc back over websockets assuming json
func serveWebsocket(ctx context.Context, w http.ResponseWriter, r *http.Request, service *api.Service, c client.Client) {
var op ws.OpCode
ct := r.Header.Get("Content-Type")
// Strip charset from Content-Type (like `application/json; charset=UTF-8`)
if idx := strings.IndexRune(ct, ';'); idx >= 0 {
ct = ct[:idx]
}
// check proto from request
switch ct {
case "application/json":
op = ws.OpText
default:
op = ws.OpBinary
}
hdr := make(http.Header)
if proto, ok := r.Header["Sec-WebSocket-Protocol"]; ok {
for _, p := range proto {
switch p {
case "binary":
hdr["Sec-WebSocket-Protocol"] = []string{"binary"}
op = ws.OpBinary
default:
op = ws.OpBinary
}
}
}
payload, err := requestPayload(r)
if err != nil {
if logger.V(logger.ErrorLevel) {
logger.Error(err.Error())
}
return
}
upgrader := ws.HTTPUpgrader{Timeout: 5 * time.Second,
Protocol: func(proto string) bool {
if strings.Contains(proto, "binary") {
return true
}
// fallback to support all protocols now
return true
},
Extension: func(httphead.Option) bool {
// disable extensions for compatibility
return false
},
Header: hdr,
}
conn, rw, _, err := upgrader.Upgrade(r, w)
if err != nil {
if logger.V(logger.ErrorLevel) {
logger.Error(err.Error())
}
return
}
defer func() {
if err := conn.Close(); err != nil {
if logger.V(logger.ErrorLevel) {
logger.Error(err.Error())
}
return
}
}()
var request interface{}
if !bytes.Equal(payload, []byte(`{}`)) {
switch ct {
case "application/json", "":
m := json.RawMessage(payload)
request = &m
default:
request = &raw.Frame{Data: payload}
}
}
// we always need to set content type for message
if ct == "" {
ct = "application/json"
}
req := c.NewRequest(
service.Name,
service.Endpoint.Name,
request,
client.WithContentType(ct),
client.StreamingRequest(),
)
// create custom router
callOpt := client.WithRouter(router.New(service.Services))
// create a new stream
stream, err := c.Stream(ctx, req, callOpt)
if err != nil {
if logger.V(logger.ErrorLevel) {
logger.Error(err.Error())
}
return
}
if request != nil {
if err = stream.Send(request); err != nil {
if logger.V(logger.ErrorLevel) {
logger.Error(err.Error())
}
return
}
}
go writeLoop(rw, stream)
rsp := stream.Response()
// receive from stream and send to client
for {
select {
case <-ctx.Done():
return
case <-stream.Context().Done():
return
default:
// read backend response body
buf, err := rsp.Read()
if err != nil {
// wants to avoid import grpc/status.Status
if strings.Contains(err.Error(), "context canceled") {
return
}
if logger.V(logger.ErrorLevel) {
logger.Error(err.Error())
}
return
}
// write the response
if err := wsutil.WriteServerMessage(rw, op, buf); err != nil {
if logger.V(logger.ErrorLevel) {
logger.Error(err.Error())
}
return
}
if err = rw.Flush(); err != nil {
if logger.V(logger.ErrorLevel) {
logger.Error(err.Error())
}
return
}
}
}
}
// writeLoop
func writeLoop(rw io.ReadWriter, stream client.Stream) {
// close stream when done
defer stream.Close()
for {
select {
case <-stream.Context().Done():
return
default:
buf, op, err := wsutil.ReadClientData(rw)
if err != nil {
if wserr, ok := err.(wsutil.ClosedError); ok {
switch wserr.Code {
case ws.StatusGoingAway:
// this happens when user leave the page
return
case ws.StatusNormalClosure, ws.StatusNoStatusRcvd:
// this happens when user close ws connection, or we don't get any status
return
}
}
if logger.V(logger.ErrorLevel) {
logger.Error(err.Error())
}
return
}
switch op {
default:
// not relevant
continue
case ws.OpText, ws.OpBinary:
break
}
// send to backend
// default to trying json
// if the extracted payload isn't empty lets use it
request := &raw.Frame{Data: buf}
if err := stream.Send(request); err != nil {
if logger.V(logger.ErrorLevel) {
logger.Error(err.Error())
}
return
}
}
}
}
func isStream(r *http.Request, srv *api.Service) bool {
// check if it's a web socket
if !isWebSocket(r) {
return false
}
// check if the endpoint supports streaming
for _, service := range srv.Services {
for _, ep := range service.Endpoints {
// skip if it doesn't match the name
if ep.Name != srv.Endpoint.Name {
continue
}
// matched if the name
if v := ep.Metadata["stream"]; v == "true" {
return true
}
}
}
return false
}
func isWebSocket(r *http.Request) bool {
contains := func(key, val string) bool {
vv := strings.Split(r.Header.Get(key), ",")
for _, v := range vv {
if val == strings.ToLower(strings.TrimSpace(v)) {
return true
}
}
return false
}
if contains("Connection", "upgrade") && contains("Upgrade", "websocket") {
return true
}
return false
}