micro/codec/grpc/grpc.go

150 lines
3.1 KiB
Go
Raw Normal View History

2019-01-02 15:55:06 +03:00
// Package grpc provides a grpc codec
package grpc
import (
"encoding/json"
"errors"
2019-01-13 22:54:07 +03:00
"fmt"
2019-01-02 15:55:06 +03:00
"io"
"strings"
"github.com/golang/protobuf/proto"
"github.com/micro/go-micro/codec"
)
type Codec struct {
Conn io.ReadWriteCloser
ContentType string
}
func (c *Codec) ReadHeader(m *codec.Message, t codec.MessageType) error {
if ct := m.Header["Content-Type"]; len(ct) > 0 {
c.ContentType = ct
}
if ct := m.Header["content-type"]; len(ct) > 0 {
c.ContentType = ct
}
// service method
path := m.Header[":path"]
if len(path) == 0 || path[0] != '/' {
2019-01-24 13:11:02 +03:00
m.Target = m.Header["Micro-Service"]
m.Endpoint = m.Header["Micro-Endpoint"]
2019-01-02 15:55:06 +03:00
} else {
// [ , a.package.Foo, Bar]
parts := strings.Split(path, "/")
if len(parts) != 3 {
return errors.New("Unknown request path")
}
service := strings.Split(parts[1], ".")
2019-01-11 00:25:31 +03:00
m.Endpoint = strings.Join([]string{service[len(service)-1], parts[2]}, ".")
2019-01-02 15:55:06 +03:00
m.Target = strings.Join(service[:len(service)-1], ".")
}
return nil
}
func (c *Codec) ReadBody(b interface{}) error {
// no body
if b == nil {
return nil
}
_, buf, err := decode(c.Conn)
if err != nil {
return err
}
switch c.ContentType {
case "application/grpc+json":
return json.Unmarshal(buf, b)
case "application/grpc+proto", "application/grpc":
return proto.Unmarshal(buf, b.(proto.Message))
}
return errors.New("Unsupported Content-Type")
}
func (c *Codec) Write(m *codec.Message, b interface{}) error {
var buf []byte
var err error
if ct := m.Header["Content-Type"]; len(ct) > 0 {
c.ContentType = ct
}
if ct := m.Header["content-type"]; len(ct) > 0 {
c.ContentType = ct
}
2019-01-13 22:54:07 +03:00
switch m.Type {
case codec.Request:
parts := strings.Split(m.Endpoint, ".")
m.Header[":method"] = "POST"
m.Header[":path"] = fmt.Sprintf("/%s.%s/%s", m.Target, parts[0], parts[1])
m.Header[":proto"] = "HTTP/2.0"
m.Header["te"] = "trailers"
m.Header["user-agent"] = "grpc-go/1.0.0"
m.Header[":authority"] = m.Target
m.Header["content-type"] = c.ContentType
case codec.Response:
m.Header["Trailer"] = "grpc-status" //, grpc-message"
m.Header["content-type"] = c.ContentType
m.Header[":status"] = "200"
2019-01-13 22:54:07 +03:00
m.Header["grpc-status"] = "0"
// m.Header["grpc-message"] = ""
2019-08-25 21:30:22 +03:00
case codec.Error:
m.Header["Trailer"] = "grpc-status, grpc-message"
// micro end of stream
if m.Error == "EOS" {
m.Header["grpc-status"] = "0"
} else {
m.Header["grpc-message"] = m.Error
m.Header["grpc-status"] = "13"
}
return nil
2019-01-13 22:54:07 +03:00
}
2019-01-02 15:55:06 +03:00
2019-01-13 22:54:07 +03:00
// marshal content
2019-01-02 15:55:06 +03:00
switch c.ContentType {
case "application/grpc+json":
buf, err = json.Marshal(b)
case "application/grpc+proto", "application/grpc":
pb, ok := b.(proto.Message)
if ok {
buf, err = proto.Marshal(pb)
}
default:
err = errors.New("Unsupported Content-Type")
}
2019-01-13 22:54:07 +03:00
// check error
2019-01-02 15:55:06 +03:00
if err != nil {
m.Header["grpc-status"] = "8"
m.Header["grpc-message"] = err.Error()
return err
}
2020-01-01 01:58:14 +03:00
if len(buf) == 0 {
return nil
}
2019-01-02 15:55:06 +03:00
return encode(0, buf, c.Conn)
}
func (c *Codec) Close() error {
return c.Conn.Close()
}
func (c *Codec) String() string {
return "grpc"
}
func NewCodec(c io.ReadWriteCloser) codec.Codec {
return &Codec{
Conn: c,
ContentType: "application/grpc",
}
}