github/ttrpc
4
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

ttrpc: remove use of typeurl

Browse Source 
Rather than employ the typeurl package, we now generate code to
correctly allocate the incoming types from the caller. As a side-effect
of this activity, the services definitions have been split out into a
separate type that handles the full resolution and dispatch of the
method, incuding correctly mapping the RPC status.

This work is a pre-cursor to larger protocol change that will allow us
to handle multiple, concurrent requests.

Signed-off-by: Stephen J Day <stephen.day@docker.com>
This commit is contained in:
Stephen J Day 2017-11-21 18:03:52 -08:00
parent 2d76dba1df
commit 2a81659f49
No known key found for this signature in database
GPG Key ID: 67B3DED84EDC823F
8 changed files with 225 additions and 136 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

41
client.go
View File

@ -4,7 +4,8 @@ import (
"context"
"net"
"github.com/containerd/typeurl"
"github.com/gogo/protobuf/proto"
"github.com/pkg/errors"
)
type Client struct {
@ -17,10 +18,17 @@ func NewClient(conn net.Conn) *Client {
}
}
func (c *Client) Call(ctx context.Context, service, method string, req interface{}) (interface{}, error) {
payload, err := typeurl.MarshalAny(req)
if err != nil {
return nil, err
func (c *Client) Call(ctx context.Context, service, method string, req, resp interface{}) error {
var payload []byte
switch v := req.(type) {
case proto.Message:
var err error
payload, err = proto.Marshal(v)
if err != nil {
return err
}
default:
return errors.Errorf("ttrpc: unknown request type: %T", req)
}
request := Request{
@ -30,22 +38,21 @@ func (c *Client) Call(ctx context.Context, service, method string, req interface
}
if err := c.channel.send(ctx, &request); err != nil {
return nil, err
return err
}
var response Response
if err := c.channel.recv(ctx, &response); err != nil {
return nil, err
return err
}
switch v := resp.(type) {
case proto.Message:
if err := proto.Unmarshal(response.Payload, v); err != nil {
return err
}
default:
return errors.Errorf("ttrpc: unknown response type: %T", resp)
}
// TODO(stevvooe): Reliance on the typeurl isn't great for bootstrapping
// and ease of use. Let's consider a request header frame and body frame as
// a better solution. This will allow the caller to set the exact type.
rpayload, err := typeurl.UnmarshalAny(response.Payload)
if err != nil {
return nil, err
}
return rpayload, nil
return nil
}

36
example/example.pb.go
View File

@ -278,14 +278,22 @@ type ExampleService interface {
Method2(ctx context.Context, req *Method1Request) (*google_protobuf.Empty, error)
}
func RegisterExampleService(srv *github_com_stevvooe_ttrpc.Server, svc ExampleService) error {
return srv.Register("ttrpc.example.v1.Example", map[string]github_com_stevvooe_ttrpc.Handler{
"Method1": github_com_stevvooe_ttrpc.HandlerFunc(func(ctx context.Context, req interface{}) (interface{}, error) {
return svc.Method1(ctx, req.(*Method1Request))
}),
"Method2": github_com_stevvooe_ttrpc.HandlerFunc(func(ctx context.Context, req interface{}) (interface{}, error) {
return svc.Method2(ctx, req.(*Method1Request))
}),
func RegisterExampleService(srv *github_com_stevvooe_ttrpc.Server, svc ExampleService) {
srv.Register("ttrpc.example.v1.Example", map[string]github_com_stevvooe_ttrpc.Method{
"Method1": func(ctx context.Context, unmarshal func(interface{}) error) (interface{}, error) {
var req Method1Request
if err := unmarshal(&req); err != nil {
return nil, err
}
return svc.Method1(ctx, &req)
},
"Method2": func(ctx context.Context, unmarshal func(interface{}) error) (interface{}, error) {
var req Method1Request
if err := unmarshal(&req); err != nil {
return nil, err
}
return svc.Method2(ctx, &req)
},
})
}
@ -300,19 +308,19 @@ func NewExampleClient(client *github_com_stevvooe_ttrpc.Client) ExampleService {
}
func (c *exampleClient) Method1(ctx context.Context, req *Method1Request) (*Method1Response, error) {
resp, err := c.client.Call(ctx, "ttrpc.example.v1.Example", "Method1", req)
if err != nil {
var resp Method1Response
if err := c.client.Call(ctx, "ttrpc.example.v1.Example", "Method1", req, &resp); err != nil {
return nil, err
}
return resp.(*Method1Response), nil
return &resp, nil
}
func (c *exampleClient) Method2(ctx context.Context, req *Method1Request) (*google_protobuf.Empty, error) {
resp, err := c.client.Call(ctx, "ttrpc.example.v1.Example", "Method2", req)
if err != nil {
var resp google_protobuf.Empty
if err := c.client.Call(ctx, "ttrpc.example.v1.Example", "Method2", req, &resp); err != nil {
return nil, err
}
return resp.(*google_protobuf.Empty), nil
return &resp, nil
}
func (m *Method1Request) Unmarshal(dAtA []byte) error {
l := len(dAtA)

13
handlers.go
View File

@ -1,13 +0,0 @@
package ttrpc
import "context"
type Handler interface {
Handle(ctx context.Context, req interface{}) (interface{}, error)
}
type HandlerFunc func(ctx context.Context, req interface{}) (interface{}, error)
func (fn HandlerFunc) Handle(ctx context.Context, req interface{}) (interface{}, error) {
return fn(ctx, req)
}

22
plugin/generator.go
View File

@ -61,16 +61,22 @@ func (p *ttrpcGenerator) genService(fullName string, service *descriptor.Service
p.P()
// registration method
p.P("func Register", serviceName, "(srv *", p.ttrpcPkg.Use(), ".Server, svc ", serviceName, ") error {")
p.P("func Register", serviceName, "(srv *", p.ttrpcPkg.Use(), ".Server, svc ", serviceName, ") {")
p.In()
p.P(`return srv.Register("`, fullName, `", map[string]`, p.ttrpcPkg.Use(), ".Handler{")
p.P(`srv.Register("`, fullName, `", map[string]`, p.ttrpcPkg.Use(), ".Method{")
p.In()
for _, method := range service.Method {
p.P(`"`, method.GetName(), `": `, p.ttrpcPkg.Use(), `.HandlerFunc(func(ctx context.Context, req interface{}) (interface{}, error) {`)
p.P(`"`, method.GetName(), `": `, `func(ctx context.Context, unmarshal func(interface{}) error) (interface{}, error) {`)
p.In()
p.P("return svc.", method.GetName(), "(ctx, req.(*", p.typeName(method.GetInputType()), "))")
p.P("var req ", p.typeName(method.GetInputType()))
p.P(`if err := unmarshal(&req); err != nil {`)
p.In()
p.P(`return nil, err`)
p.Out()
p.P("}),")
p.P(`}`)
p.P("return svc.", method.GetName(), "(ctx, &req)")
p.Out()
p.P("},")
}
p.Out()
p.P("})")
@ -103,13 +109,13 @@ func (p *ttrpcGenerator) genService(fullName string, service *descriptor.Service
"req *", p.typeName(method.GetInputType()), ") ",
"(*", p.typeName(method.GetOutputType()), ", error) {")
p.In()
p.P("resp, err := c.client.Call(ctx, ", `"`+fullName+`", `, `"`+method.GetName()+`"`, ", req)")
p.P("if err != nil {")
p.P("var resp ", p.typeName(method.GetOutputType()))
p.P("if err := c.client.Call(ctx, ", `"`+fullName+`", `, `"`+method.GetName()+`"`, ", req, &resp); err != nil {")
p.In()
p.P("return nil, err")
p.Out()
p.P("}")
p.P("return resp.(*", p.typeName(method.GetOutputType()), "), nil")
p.P("return &resp, nil")
p.Out()
p.P("}")
}

74
server.go
View File

@ -3,29 +3,22 @@ package ttrpc
import (
"context"
"net"
"path"
"github.com/containerd/containerd/errdefs"
"github.com/containerd/containerd/log"
"github.com/containerd/typeurl"
"github.com/pkg/errors"
)
type Server struct {
handlers map[string]map[string]Handler
services *serviceSet
}
func NewServer() *Server {
return &Server{handlers: make(map[string]map[string]Handler)}
return &Server{
services: newServiceSet(),
}
}
func (s *Server) Register(name string, methods map[string]Handler) error {
if _, ok := s.handlers[name]; ok {
return errors.Errorf("duplicate service %v registered", name)
}
s.handlers[name] = methods
return nil
func (s *Server) Register(name string, methods map[string]Method) {
s.services.register(name, methods)
}
func (s *Server) Shutdown(ctx context.Context) error {
@ -70,10 +63,11 @@ func (s *Server) handleConn(conn net.Conn) {
return
}
resp, err := s.dispatch(ctx, &req)
if err != nil {
log.L.WithError(err).Error("failed to dispatch request")
return
p, status := s.services.call(ctx, req.Service, req.Method, req.Payload)
resp := &Response{
Status: status.Proto(),
Payload: p,
}
if err := ch.send(ctx, resp); err != nil {
@ -82,49 +76,3 @@ func (s *Server) handleConn(conn net.Conn) {
}
}
}
func (s *Server) dispatch(ctx context.Context, req *Request) (*Response, error) {
ctx = log.WithLogger(ctx, log.G(ctx).WithField("method", path.Join("/", req.Service, req.Method)))
handler, err := s.resolve(req.Service, req.Method)
if err != nil {
log.L.WithError(err).Error("failed to resolve handler")
return nil, err
}
payload, err := typeurl.UnmarshalAny(req.Payload)
if err != nil {
return nil, err
}
resp, err := handler.Handle(ctx, payload)
if err != nil {
log.L.WithError(err).Error("handler returned an error")
return nil, err
}
apayload, err := typeurl.MarshalAny(resp)
if err != nil {
return nil, err
}
rresp := &Response{
// Status: *st,
Payload: apayload,
}
return rresp, nil
}
func (s *Server) resolve(service, method string) (Handler, error) {
srv, ok := s.handlers[service]
if !ok {
return nil, errors.Wrapf(errdefs.ErrNotFound, "could not resolve service %v", service)
}
handler, ok := srv[method]
if !ok {
return nil, errors.Wrapf(errdefs.ErrNotFound, "could not resolve method %v", method)
}
return handler, nil
}

26
server_test.go
View File

@ -31,12 +31,8 @@ func newTestingClient(client *Client) *testingClient {
}
func (tc *testingClient) Test(ctx context.Context, req *testPayload) (*testPayload, error) {
resp, err := tc.client.Call(ctx, serviceName, "Test", req)
if err != nil {
return nil, err
}
return resp.(*testPayload), nil
var tp testPayload
return &tp, tc.client.Call(ctx, serviceName, "Test", req, &tp)
}
type testPayload struct {
@ -72,17 +68,19 @@ func TestServer(t *testing.T) {
// more mocking of what is generated code. Unlike grpc, we register with a
// closure so that the descriptor is allocated only on registration.
registerTestingService := func(srv *Server, svc testingService) error {
return srv.Register(serviceName, map[string]Handler{
"Test": HandlerFunc(func(ctx context.Context, req interface{}) (interface{}, error) {
return svc.Test(ctx, req.(*testPayload))
}),
registerTestingService := func(srv *Server, svc testingService) {
srv.Register(serviceName, map[string]Method{
"Test": func(ctx context.Context, unmarshal func(interface{}) error) (interface{}, error) {
var req testPayload
if err := unmarshal(&req); err != nil {
return nil, err
}
return svc.Test(ctx, &req)
},
})
}
if err := registerTestingService(server, testImpl); err != nil {
t.Fatal(err)
}
registerTestingService(server, testImpl)
listener, err := net.Listen("tcp", ":0")
if err != nil {

136
services.go Normal file
View File

@ -0,0 +1,136 @@
package ttrpc
import (
"context"
"io"
"os"
"path"
"github.com/containerd/containerd/log"
"github.com/gogo/protobuf/proto"
"github.com/pkg/errors"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
)
type Method func(ctx context.Context, unmarshal func(interface{}) error) (interface{}, error)
type ServiceDesc struct {
Methods map[string]Method
// TODO(stevvooe): Add stream support.
}
type serviceSet struct {
services map[string]ServiceDesc
}
func newServiceSet() *serviceSet {
return &serviceSet{
services: make(map[string]ServiceDesc),
}
}
func (s *serviceSet) register(name string, methods map[string]Method) {
if _, ok := s.services[name]; ok {
panic(errors.Errorf("duplicate service %v registered", name))
}
s.services[name] = ServiceDesc{
Methods: methods,
}
}
func (s *serviceSet) call(ctx context.Context, serviceName, methodName string, p []byte) ([]byte, *status.Status) {
p, err := s.dispatch(ctx, serviceName, methodName, p)
st, ok := status.FromError(err)
if !ok {
st = status.New(convertCode(err), err.Error())
}
return p, st
}
func (s *serviceSet) dispatch(ctx context.Context, serviceName, methodName string, p []byte) ([]byte, error) {
ctx = log.WithLogger(ctx, log.G(ctx).WithField("method", fullPath(serviceName, methodName)))
method, err := s.resolve(serviceName, methodName)
if err != nil {
return nil, err
}
unmarshal := func(obj interface{}) error {
switch v := obj.(type) {
case proto.Message:
if err := proto.Unmarshal(p, v); err != nil {
return status.Errorf(codes.Internal, "ttrpc: error unmarshaling payload: %v", err.Error())
}
default:
return status.Errorf(codes.Internal, "ttrpc: error unsupported request type: %T", v)
}
return nil
}
resp, err := method(ctx, unmarshal)
if err != nil {
return nil, err
}
switch v := resp.(type) {
case proto.Message:
r, err := proto.Marshal(v)
if err != nil {
return nil, status.Errorf(codes.Internal, "ttrpc: error marshaling payload: %v", err.Error())
}
return r, nil
default:
return nil, status.Errorf(codes.Internal, "ttrpc: error unsupported response type: %T", v)
}
}
func (s *serviceSet) resolve(service, method string) (Method, error) {
srv, ok := s.services[service]
if !ok {
return nil, status.Errorf(codes.NotFound, "service %v", service)
}
mthd, ok := srv.Methods[method]
if !ok {
return nil, status.Errorf(codes.NotFound, "method %v", method)
}
return mthd, nil
}
// convertCode maps stdlib go errors into grpc space.
//
// This is ripped from the grpc-go code base.
func convertCode(err error) codes.Code {
switch err {
case nil:
return codes.OK
case io.EOF:
return codes.OutOfRange
case io.ErrClosedPipe, io.ErrNoProgress, io.ErrShortBuffer, io.ErrShortWrite, io.ErrUnexpectedEOF:
return codes.FailedPrecondition
case os.ErrInvalid:
return codes.InvalidArgument
case context.Canceled:
return codes.Canceled
case context.DeadlineExceeded:
return codes.DeadlineExceeded
}
switch {
case os.IsExist(err):
return codes.AlreadyExists
case os.IsNotExist(err):
return codes.NotFound
case os.IsPermission(err):
return codes.PermissionDenied
}
return codes.Unknown
}
func fullPath(service, method string) string {
return "/" + path.Join("/", service, method)
}

13
types.go
View File

@ -3,14 +3,13 @@ package ttrpc
import (
"fmt"
"github.com/containerd/containerd/protobuf/google/rpc"
"github.com/gogo/protobuf/types"
spb "google.golang.org/genproto/googleapis/rpc/status"
)
type Request struct {
Service string `protobuf:"bytes,1,opt,name=service,proto3"`
Method string `protobuf:"bytes,2,opt,name=method,proto3"`
Payload *types.Any `protobuf:"bytes,3,opt,name=payload,proto3"`
Service string `protobuf:"bytes,1,opt,name=service,proto3"`
Method string `protobuf:"bytes,2,opt,name=method,proto3"`
Payload []byte `protobuf:"bytes,3,opt,name=payload,proto3"`
}
func (r *Request) Reset() { *r = Request{} }
@ -18,8 +17,8 @@ func (r *Request) String() string { return fmt.Sprintf("%+#v", r) }
func (r *Request) ProtoMessage() {}
type Response struct {
Status *rpc.Status `protobuf:"bytes,1,opt,name=status,proto3"`
Payload *types.Any `protobuf:"bytes,2,opt,name=payload,proto3"`
Status *spb.Status `protobuf:"bytes,1,opt,name=status,proto3"`
Payload []byte `protobuf:"bytes,2,opt,name=payload,proto3"`
}
func (r *Response) Reset() { *r = Response{} }