From 8b08b2d7b2772be4d5e78a28177d5a06f77e9c9a Mon Sep 17 00:00:00 2001 From: iamqizhao Date: Wed, 25 Mar 2015 15:18:07 -0700 Subject: [PATCH 1/2] Interface change to accommodate the support of non-protobuf data formats --- benchmark/grpc_testing/test.pb.go | 36 ++++++++++---------- call.go | 19 +++++------ examples/route_guide/proto/route_guide.pb.go | 29 ++++++++-------- interop/grpc_testing/test.pb.go | 36 ++++++++++---------- rpc_util.go | 34 +++++++++++++++--- rpc_util_test.go | 8 ++--- server.go | 11 +++--- stream.go | 29 ++++++++-------- test/grpc_testing/test.pb.go | 36 ++++++++++---------- 9 files changed, 131 insertions(+), 107 deletions(-) diff --git a/benchmark/grpc_testing/test.pb.go b/benchmark/grpc_testing/test.pb.go index 6246f1c9..d454e578 100644 --- a/benchmark/grpc_testing/test.pb.go +++ b/benchmark/grpc_testing/test.pb.go @@ -388,7 +388,7 @@ func (c *testServiceClient) StreamingOutputCall(ctx context.Context, in *Streami return nil, err } x := &testServiceStreamingOutputCallClient{stream} - if err := x.ClientStream.SendProto(in); err != nil { + if err := x.ClientStream.SendMsg(in); err != nil { return nil, err } if err := x.ClientStream.CloseSend(); err != nil { @@ -408,7 +408,7 @@ type testServiceStreamingOutputCallClient struct { func (x *testServiceStreamingOutputCallClient) Recv() (*StreamingOutputCallResponse, error) { m := new(StreamingOutputCallResponse) - if err := x.ClientStream.RecvProto(m); err != nil { + if err := x.ClientStream.RecvMsg(m); err != nil { return nil, err } return m, nil @@ -434,7 +434,7 @@ type testServiceStreamingInputCallClient struct { } func (x *testServiceStreamingInputCallClient) Send(m *StreamingInputCallRequest) error { - return x.ClientStream.SendProto(m) + return x.ClientStream.SendMsg(m) } func (x *testServiceStreamingInputCallClient) CloseAndRecv() (*StreamingInputCallResponse, error) { @@ -442,7 +442,7 @@ func (x *testServiceStreamingInputCallClient) CloseAndRecv() (*StreamingInputCal return nil, err } m := new(StreamingInputCallResponse) - if err := x.ClientStream.RecvProto(m); err != nil { + if err := x.ClientStream.RecvMsg(m); err != nil { return nil, err } return m, nil @@ -468,12 +468,12 @@ type testServiceFullDuplexCallClient struct { } func (x *testServiceFullDuplexCallClient) Send(m *StreamingOutputCallRequest) error { - return x.ClientStream.SendProto(m) + return x.ClientStream.SendMsg(m) } func (x *testServiceFullDuplexCallClient) Recv() (*StreamingOutputCallResponse, error) { m := new(StreamingOutputCallResponse) - if err := x.ClientStream.RecvProto(m); err != nil { + if err := x.ClientStream.RecvMsg(m); err != nil { return nil, err } return m, nil @@ -499,12 +499,12 @@ type testServiceHalfDuplexCallClient struct { } func (x *testServiceHalfDuplexCallClient) Send(m *StreamingOutputCallRequest) error { - return x.ClientStream.SendProto(m) + return x.ClientStream.SendMsg(m) } func (x *testServiceHalfDuplexCallClient) Recv() (*StreamingOutputCallResponse, error) { m := new(StreamingOutputCallResponse) - if err := x.ClientStream.RecvProto(m); err != nil { + if err := x.ClientStream.RecvMsg(m); err != nil { return nil, err } return m, nil @@ -539,7 +539,7 @@ func RegisterTestServiceServer(s *grpc.Server, srv TestServiceServer) { s.RegisterService(&_TestService_serviceDesc, srv) } -func _TestService_EmptyCall_Handler(srv interface{}, ctx context.Context, buf []byte) (proto.Message, error) { +func _TestService_EmptyCall_Handler(srv interface{}, ctx context.Context, buf []byte) (interface{}, error) { in := new(Empty) if err := proto.Unmarshal(buf, in); err != nil { return nil, err @@ -551,7 +551,7 @@ func _TestService_EmptyCall_Handler(srv interface{}, ctx context.Context, buf [] return out, nil } -func _TestService_UnaryCall_Handler(srv interface{}, ctx context.Context, buf []byte) (proto.Message, error) { +func _TestService_UnaryCall_Handler(srv interface{}, ctx context.Context, buf []byte) (interface{}, error) { in := new(SimpleRequest) if err := proto.Unmarshal(buf, in); err != nil { return nil, err @@ -565,7 +565,7 @@ func _TestService_UnaryCall_Handler(srv interface{}, ctx context.Context, buf [] func _TestService_StreamingOutputCall_Handler(srv interface{}, stream grpc.ServerStream) error { m := new(StreamingOutputCallRequest) - if err := stream.RecvProto(m); err != nil { + if err := stream.RecvMsg(m); err != nil { return err } return srv.(TestServiceServer).StreamingOutputCall(m, &testServiceStreamingOutputCallServer{stream}) @@ -581,7 +581,7 @@ type testServiceStreamingOutputCallServer struct { } func (x *testServiceStreamingOutputCallServer) Send(m *StreamingOutputCallResponse) error { - return x.ServerStream.SendProto(m) + return x.ServerStream.SendMsg(m) } func _TestService_StreamingInputCall_Handler(srv interface{}, stream grpc.ServerStream) error { @@ -599,12 +599,12 @@ type testServiceStreamingInputCallServer struct { } func (x *testServiceStreamingInputCallServer) SendAndClose(m *StreamingInputCallResponse) error { - return x.ServerStream.SendProto(m) + return x.ServerStream.SendMsg(m) } func (x *testServiceStreamingInputCallServer) Recv() (*StreamingInputCallRequest, error) { m := new(StreamingInputCallRequest) - if err := x.ServerStream.RecvProto(m); err != nil { + if err := x.ServerStream.RecvMsg(m); err != nil { return nil, err } return m, nil @@ -625,12 +625,12 @@ type testServiceFullDuplexCallServer struct { } func (x *testServiceFullDuplexCallServer) Send(m *StreamingOutputCallResponse) error { - return x.ServerStream.SendProto(m) + return x.ServerStream.SendMsg(m) } func (x *testServiceFullDuplexCallServer) Recv() (*StreamingOutputCallRequest, error) { m := new(StreamingOutputCallRequest) - if err := x.ServerStream.RecvProto(m); err != nil { + if err := x.ServerStream.RecvMsg(m); err != nil { return nil, err } return m, nil @@ -651,12 +651,12 @@ type testServiceHalfDuplexCallServer struct { } func (x *testServiceHalfDuplexCallServer) Send(m *StreamingOutputCallResponse) error { - return x.ServerStream.SendProto(m) + return x.ServerStream.SendMsg(m) } func (x *testServiceHalfDuplexCallServer) Recv() (*StreamingOutputCallRequest, error) { m := new(StreamingOutputCallRequest) - if err := x.ServerStream.RecvProto(m); err != nil { + if err := x.ServerStream.RecvMsg(m); err != nil { return nil, err } return m, nil diff --git a/call.go b/call.go index 32ca7eb3..aad89fa4 100644 --- a/call.go +++ b/call.go @@ -37,18 +37,17 @@ import ( "io" "net" - "github.com/golang/protobuf/proto" "golang.org/x/net/context" "google.golang.org/grpc/codes" "google.golang.org/grpc/metadata" "google.golang.org/grpc/transport" ) -// recv receives and parses an RPC response. +// recvResponse receives and parses an RPC response. // On error, it returns the error and indicates whether the call should be retried. // // TODO(zhaoq): Check whether the received message sequence is valid. -func recv(t transport.ClientTransport, c *callInfo, stream *transport.Stream, reply proto.Message) error { +func recvResponse(t transport.ClientTransport, c *callInfo, stream *transport.Stream, reply interface{}) error { // Try to acquire header metadata from the server if there is any. var err error c.headerMD, err = stream.Header() @@ -57,7 +56,7 @@ func recv(t transport.ClientTransport, c *callInfo, stream *transport.Stream, re } p := &parser{s: stream} for { - if err = recvProto(p, reply); err != nil { + if err = recv(p, protoCodec{}, reply); err != nil { if err == io.EOF { break } @@ -68,8 +67,8 @@ func recv(t transport.ClientTransport, c *callInfo, stream *transport.Stream, re return nil } -// sendRPC writes out various information of an RPC such as Context and Message. -func sendRPC(ctx context.Context, callHdr *transport.CallHdr, t transport.ClientTransport, args proto.Message, opts *transport.Options) (_ *transport.Stream, err error) { +// sendRequest writes out various information of an RPC such as Context and Message. +func sendRequest(ctx context.Context, callHdr *transport.CallHdr, t transport.ClientTransport, args interface{}, opts *transport.Options) (_ *transport.Stream, err error) { stream, err := t.NewStream(ctx, callHdr) if err != nil { return nil, err @@ -82,7 +81,7 @@ func sendRPC(ctx context.Context, callHdr *transport.CallHdr, t transport.Client } }() // TODO(zhaoq): Support compression. - outBuf, err := encode(args, compressionNone) + outBuf, err := encode(protoCodec{}, args, compressionNone) if err != nil { return nil, transport.StreamErrorf(codes.Internal, "grpc: %v", err) } @@ -103,7 +102,7 @@ type callInfo struct { // Invoke is called by the generated code. It sends the RPC request on the // wire and returns after response is received. -func Invoke(ctx context.Context, method string, args, reply proto.Message, cc *ClientConn, opts ...CallOption) error { +func Invoke(ctx context.Context, method string, args, reply interface{}, cc *ClientConn, opts ...CallOption) error { var c callInfo for _, o := range opts { if err := o.before(&c); err != nil { @@ -149,7 +148,7 @@ func Invoke(ctx context.Context, method string, args, reply proto.Message, cc *C } return toRPCErr(err) } - stream, err = sendRPC(ctx, callHdr, t, args, topts) + stream, err = sendRequest(ctx, callHdr, t, args, topts) if err != nil { if _, ok := err.(transport.ConnectionError); ok { lastErr = err @@ -161,7 +160,7 @@ func Invoke(ctx context.Context, method string, args, reply proto.Message, cc *C return toRPCErr(err) } // Receive the response - lastErr = recv(t, &c, stream, reply) + lastErr = recvResponse(t, &c, stream, reply) if _, ok := lastErr.(transport.ConnectionError); ok { continue } diff --git a/examples/route_guide/proto/route_guide.pb.go b/examples/route_guide/proto/route_guide.pb.go index 71a31818..6c21d0d7 100644 --- a/examples/route_guide/proto/route_guide.pb.go +++ b/examples/route_guide/proto/route_guide.pb.go @@ -17,7 +17,8 @@ It has these top-level messages: */ package proto -import proto1 "github.com/golang/protobuf/proto" +import proto "github.com/golang/protobuf/proto" +import math "math" import ( context "golang.org/x/net/context" @@ -186,7 +187,7 @@ func (c *routeGuideClient) ListFeatures(ctx context.Context, in *Rectangle, opts return nil, err } x := &routeGuideListFeaturesClient{stream} - if err := x.ClientStream.SendProto(in); err != nil { + if err := x.ClientStream.SendMsg(in); err != nil { return nil, err } if err := x.ClientStream.CloseSend(); err != nil { @@ -206,7 +207,7 @@ type routeGuideListFeaturesClient struct { func (x *routeGuideListFeaturesClient) Recv() (*Feature, error) { m := new(Feature) - if err := x.ClientStream.RecvProto(m); err != nil { + if err := x.ClientStream.RecvMsg(m); err != nil { return nil, err } return m, nil @@ -232,7 +233,7 @@ type routeGuideRecordRouteClient struct { } func (x *routeGuideRecordRouteClient) Send(m *Point) error { - return x.ClientStream.SendProto(m) + return x.ClientStream.SendMsg(m) } func (x *routeGuideRecordRouteClient) CloseAndRecv() (*RouteSummary, error) { @@ -240,7 +241,7 @@ func (x *routeGuideRecordRouteClient) CloseAndRecv() (*RouteSummary, error) { return nil, err } m := new(RouteSummary) - if err := x.ClientStream.RecvProto(m); err != nil { + if err := x.ClientStream.RecvMsg(m); err != nil { return nil, err } return m, nil @@ -266,12 +267,12 @@ type routeGuideRouteChatClient struct { } func (x *routeGuideRouteChatClient) Send(m *RouteNote) error { - return x.ClientStream.SendProto(m) + return x.ClientStream.SendMsg(m) } func (x *routeGuideRouteChatClient) Recv() (*RouteNote, error) { m := new(RouteNote) - if err := x.ClientStream.RecvProto(m); err != nil { + if err := x.ClientStream.RecvMsg(m); err != nil { return nil, err } return m, nil @@ -310,7 +311,7 @@ func RegisterRouteGuideServer(s *grpc.Server, srv RouteGuideServer) { s.RegisterService(&_RouteGuide_serviceDesc, srv) } -func _RouteGuide_GetFeature_Handler(srv interface{}, ctx context.Context, buf []byte) (proto1.Message, error) { +func _RouteGuide_GetFeature_Handler(srv interface{}, ctx context.Context, buf []byte) (interface{}, error) { in := new(Point) if err := proto1.Unmarshal(buf, in); err != nil { return nil, err @@ -324,7 +325,7 @@ func _RouteGuide_GetFeature_Handler(srv interface{}, ctx context.Context, buf [] func _RouteGuide_ListFeatures_Handler(srv interface{}, stream grpc.ServerStream) error { m := new(Rectangle) - if err := stream.RecvProto(m); err != nil { + if err := stream.RecvMsg(m); err != nil { return err } return srv.(RouteGuideServer).ListFeatures(m, &routeGuideListFeaturesServer{stream}) @@ -340,7 +341,7 @@ type routeGuideListFeaturesServer struct { } func (x *routeGuideListFeaturesServer) Send(m *Feature) error { - return x.ServerStream.SendProto(m) + return x.ServerStream.SendMsg(m) } func _RouteGuide_RecordRoute_Handler(srv interface{}, stream grpc.ServerStream) error { @@ -358,12 +359,12 @@ type routeGuideRecordRouteServer struct { } func (x *routeGuideRecordRouteServer) SendAndClose(m *RouteSummary) error { - return x.ServerStream.SendProto(m) + return x.ServerStream.SendMsg(m) } func (x *routeGuideRecordRouteServer) Recv() (*Point, error) { m := new(Point) - if err := x.ServerStream.RecvProto(m); err != nil { + if err := x.ServerStream.RecvMsg(m); err != nil { return nil, err } return m, nil @@ -384,12 +385,12 @@ type routeGuideRouteChatServer struct { } func (x *routeGuideRouteChatServer) Send(m *RouteNote) error { - return x.ServerStream.SendProto(m) + return x.ServerStream.SendMsg(m) } func (x *routeGuideRouteChatServer) Recv() (*RouteNote, error) { m := new(RouteNote) - if err := x.ServerStream.RecvProto(m); err != nil { + if err := x.ServerStream.RecvMsg(m); err != nil { return nil, err } return m, nil diff --git a/interop/grpc_testing/test.pb.go b/interop/grpc_testing/test.pb.go index 6246f1c9..d454e578 100755 --- a/interop/grpc_testing/test.pb.go +++ b/interop/grpc_testing/test.pb.go @@ -388,7 +388,7 @@ func (c *testServiceClient) StreamingOutputCall(ctx context.Context, in *Streami return nil, err } x := &testServiceStreamingOutputCallClient{stream} - if err := x.ClientStream.SendProto(in); err != nil { + if err := x.ClientStream.SendMsg(in); err != nil { return nil, err } if err := x.ClientStream.CloseSend(); err != nil { @@ -408,7 +408,7 @@ type testServiceStreamingOutputCallClient struct { func (x *testServiceStreamingOutputCallClient) Recv() (*StreamingOutputCallResponse, error) { m := new(StreamingOutputCallResponse) - if err := x.ClientStream.RecvProto(m); err != nil { + if err := x.ClientStream.RecvMsg(m); err != nil { return nil, err } return m, nil @@ -434,7 +434,7 @@ type testServiceStreamingInputCallClient struct { } func (x *testServiceStreamingInputCallClient) Send(m *StreamingInputCallRequest) error { - return x.ClientStream.SendProto(m) + return x.ClientStream.SendMsg(m) } func (x *testServiceStreamingInputCallClient) CloseAndRecv() (*StreamingInputCallResponse, error) { @@ -442,7 +442,7 @@ func (x *testServiceStreamingInputCallClient) CloseAndRecv() (*StreamingInputCal return nil, err } m := new(StreamingInputCallResponse) - if err := x.ClientStream.RecvProto(m); err != nil { + if err := x.ClientStream.RecvMsg(m); err != nil { return nil, err } return m, nil @@ -468,12 +468,12 @@ type testServiceFullDuplexCallClient struct { } func (x *testServiceFullDuplexCallClient) Send(m *StreamingOutputCallRequest) error { - return x.ClientStream.SendProto(m) + return x.ClientStream.SendMsg(m) } func (x *testServiceFullDuplexCallClient) Recv() (*StreamingOutputCallResponse, error) { m := new(StreamingOutputCallResponse) - if err := x.ClientStream.RecvProto(m); err != nil { + if err := x.ClientStream.RecvMsg(m); err != nil { return nil, err } return m, nil @@ -499,12 +499,12 @@ type testServiceHalfDuplexCallClient struct { } func (x *testServiceHalfDuplexCallClient) Send(m *StreamingOutputCallRequest) error { - return x.ClientStream.SendProto(m) + return x.ClientStream.SendMsg(m) } func (x *testServiceHalfDuplexCallClient) Recv() (*StreamingOutputCallResponse, error) { m := new(StreamingOutputCallResponse) - if err := x.ClientStream.RecvProto(m); err != nil { + if err := x.ClientStream.RecvMsg(m); err != nil { return nil, err } return m, nil @@ -539,7 +539,7 @@ func RegisterTestServiceServer(s *grpc.Server, srv TestServiceServer) { s.RegisterService(&_TestService_serviceDesc, srv) } -func _TestService_EmptyCall_Handler(srv interface{}, ctx context.Context, buf []byte) (proto.Message, error) { +func _TestService_EmptyCall_Handler(srv interface{}, ctx context.Context, buf []byte) (interface{}, error) { in := new(Empty) if err := proto.Unmarshal(buf, in); err != nil { return nil, err @@ -551,7 +551,7 @@ func _TestService_EmptyCall_Handler(srv interface{}, ctx context.Context, buf [] return out, nil } -func _TestService_UnaryCall_Handler(srv interface{}, ctx context.Context, buf []byte) (proto.Message, error) { +func _TestService_UnaryCall_Handler(srv interface{}, ctx context.Context, buf []byte) (interface{}, error) { in := new(SimpleRequest) if err := proto.Unmarshal(buf, in); err != nil { return nil, err @@ -565,7 +565,7 @@ func _TestService_UnaryCall_Handler(srv interface{}, ctx context.Context, buf [] func _TestService_StreamingOutputCall_Handler(srv interface{}, stream grpc.ServerStream) error { m := new(StreamingOutputCallRequest) - if err := stream.RecvProto(m); err != nil { + if err := stream.RecvMsg(m); err != nil { return err } return srv.(TestServiceServer).StreamingOutputCall(m, &testServiceStreamingOutputCallServer{stream}) @@ -581,7 +581,7 @@ type testServiceStreamingOutputCallServer struct { } func (x *testServiceStreamingOutputCallServer) Send(m *StreamingOutputCallResponse) error { - return x.ServerStream.SendProto(m) + return x.ServerStream.SendMsg(m) } func _TestService_StreamingInputCall_Handler(srv interface{}, stream grpc.ServerStream) error { @@ -599,12 +599,12 @@ type testServiceStreamingInputCallServer struct { } func (x *testServiceStreamingInputCallServer) SendAndClose(m *StreamingInputCallResponse) error { - return x.ServerStream.SendProto(m) + return x.ServerStream.SendMsg(m) } func (x *testServiceStreamingInputCallServer) Recv() (*StreamingInputCallRequest, error) { m := new(StreamingInputCallRequest) - if err := x.ServerStream.RecvProto(m); err != nil { + if err := x.ServerStream.RecvMsg(m); err != nil { return nil, err } return m, nil @@ -625,12 +625,12 @@ type testServiceFullDuplexCallServer struct { } func (x *testServiceFullDuplexCallServer) Send(m *StreamingOutputCallResponse) error { - return x.ServerStream.SendProto(m) + return x.ServerStream.SendMsg(m) } func (x *testServiceFullDuplexCallServer) Recv() (*StreamingOutputCallRequest, error) { m := new(StreamingOutputCallRequest) - if err := x.ServerStream.RecvProto(m); err != nil { + if err := x.ServerStream.RecvMsg(m); err != nil { return nil, err } return m, nil @@ -651,12 +651,12 @@ type testServiceHalfDuplexCallServer struct { } func (x *testServiceHalfDuplexCallServer) Send(m *StreamingOutputCallResponse) error { - return x.ServerStream.SendProto(m) + return x.ServerStream.SendMsg(m) } func (x *testServiceHalfDuplexCallServer) Recv() (*StreamingOutputCallRequest, error) { m := new(StreamingOutputCallRequest) - if err := x.ServerStream.RecvProto(m); err != nil { + if err := x.ServerStream.RecvMsg(m); err != nil { return nil, err } return m, nil diff --git a/rpc_util.go b/rpc_util.go index 0107d818..623b85b1 100644 --- a/rpc_util.go +++ b/rpc_util.go @@ -49,6 +49,32 @@ import ( "google.golang.org/grpc/transport" ) +// Codec defines the interface gRPC uses to encode and decode messages. +type Codec interface { + // Marshal returns the wire format of v. + Marshal(v interface{}) ([]byte, error) + // Unmarshal parses the wire format into v. + Unmarshal(data []byte, v interface{}) error + // String returns the name of the Codec implementation. The returned + // string will be used as part of content type in transmission. + String() string +} + +// protoCodec is a Codec implemetation with protobuf. It is the default codec for gRPC. +type protoCodec struct{} + +func (protoCodec) Marshal(v interface{}) ([]byte, error) { + return proto.Marshal(v.(proto.Message)) +} + +func (protoCodec) Unmarshal(data []byte, v interface{}) error { + return proto.Unmarshal(data, v.(proto.Message)) +} + +func (protoCodec) String() string { + return "proto" +} + // CallOption configures a Call before it starts or extracts information from // a Call after it completes. type CallOption interface { @@ -131,7 +157,7 @@ func (p *parser) recvMsg() (pf payloadFormat, msg []byte, err error) { // encode serializes msg and prepends the message header. If msg is nil, it // generates the message header of 0 message length. -func encode(msg proto.Message, pf payloadFormat) ([]byte, error) { +func encode(c Codec, msg interface{}, pf payloadFormat) ([]byte, error) { var buf bytes.Buffer // Write message fixed header. buf.WriteByte(uint8(pf)) @@ -140,7 +166,7 @@ func encode(msg proto.Message, pf payloadFormat) ([]byte, error) { if msg != nil { var err error // TODO(zhaoq): optimize to reduce memory alloc and copying. - b, err = proto.Marshal(msg) + b, err = c.Marshal(msg) if err != nil { return nil, err } @@ -153,14 +179,14 @@ func encode(msg proto.Message, pf payloadFormat) ([]byte, error) { return buf.Bytes(), nil } -func recvProto(p *parser, m proto.Message) error { +func recv(p *parser, c Codec, m interface{}) error { pf, d, err := p.recvMsg() if err != nil { return err } switch pf { case compressionNone: - if err := proto.Unmarshal(d, m); err != nil { + if err := c.Unmarshal(d, m); err != nil { return Errorf(codes.Internal, "grpc: %v", err) } default: diff --git a/rpc_util_test.go b/rpc_util_test.go index 6dd408ac..56cac330 100644 --- a/rpc_util_test.go +++ b/rpc_util_test.go @@ -112,9 +112,9 @@ func TestEncode(t *testing.T) { }{ {nil, compressionNone, []byte{0, 0, 0, 0, 0}, nil}, } { - b, err := encode(test.msg, test.pt) + b, err := encode(protoCodec{}, test.msg, test.pt) if err != test.err || !bytes.Equal(b, test.b) { - t.Fatalf("encode(_, %d) = %v, %v\nwant %v, %v", test.pt, b, err, test.b, test.err) + t.Fatalf("encode(_, _, %d) = %v, %v\nwant %v, %v", test.pt, b, err, test.b, test.err) } } } @@ -176,12 +176,12 @@ func TestBackoff(t *testing.T) { // bytes. func bmEncode(b *testing.B, mSize int) { msg := &perfpb.Buffer{Body: make([]byte, mSize)} - encoded, _ := encode(msg, compressionNone) + encoded, _ := encode(protoCodec{}, msg, compressionNone) encodedSz := int64(len(encoded)) b.ReportAllocs() b.ResetTimer() for i := 0; i < b.N; i++ { - encode(msg, compressionNone) + encode(protoCodec{}, msg, compressionNone) } b.SetBytes(encodedSz) } diff --git a/server.go b/server.go index 6c7a78b2..e50f2b5f 100644 --- a/server.go +++ b/server.go @@ -43,14 +43,13 @@ import ( "strings" "sync" - "github.com/golang/protobuf/proto" "golang.org/x/net/context" "google.golang.org/grpc/codes" "google.golang.org/grpc/metadata" "google.golang.org/grpc/transport" ) -type methodHandler func(srv interface{}, ctx context.Context, buf []byte) (proto.Message, error) +type methodHandler func(srv interface{}, ctx context.Context, buf []byte) (interface{}, error) // MethodDesc represents an RPC service's method specification. type MethodDesc struct { @@ -203,8 +202,8 @@ func (s *Server) Serve(lis net.Listener) error { } } -func (s *Server) sendProto(t transport.ServerTransport, stream *transport.Stream, msg proto.Message, pf payloadFormat, opts *transport.Options) error { - p, err := encode(msg, pf) +func (s *Server) sendResponse(t transport.ServerTransport, stream *transport.Stream, msg interface{}, pf payloadFormat, opts *transport.Options) error { + p, err := encode(protoCodec{}, msg, pf) if err != nil { // This typically indicates a fatal issue (e.g., memory // corruption or hardware faults) the application program @@ -213,7 +212,7 @@ func (s *Server) sendProto(t transport.ServerTransport, stream *transport.Stream // TODO(zhaoq): There exist other options also such as only closing the // faulty stream locally and remotely (Other streams can keep going). Find // the optimal option. - log.Fatalf("grpc: Server failed to encode proto message %v", err) + log.Fatalf("grpc: Server failed to encode response %v", err) } return t.Write(stream, p, opts) } @@ -261,7 +260,7 @@ func (s *Server) processUnaryRPC(t transport.ServerTransport, stream *transport. Last: true, Delay: false, } - if err := s.sendProto(t, stream, reply, compressionNone, opts); err != nil { + if err := s.sendResponse(t, stream, reply, compressionNone, opts); err != nil { if _, ok := err.(transport.ConnectionError); ok { return } diff --git a/stream.go b/stream.go index 5f4227c7..625535a6 100644 --- a/stream.go +++ b/stream.go @@ -38,7 +38,6 @@ import ( "io" "net" - "github.com/golang/protobuf/proto" "golang.org/x/net/context" "google.golang.org/grpc/codes" "google.golang.org/grpc/metadata" @@ -61,17 +60,17 @@ type StreamDesc struct { type Stream interface { // Context returns the context for this stream. Context() context.Context - // SendProto blocks until it sends m, the stream is done or the stream + // SendMsg blocks until it sends m, the stream is done or the stream // breaks. // On error, it aborts the stream and returns an RPC status on client // side. On server side, it simply returns the error to the caller. - // SendProto is called by generated code. - SendProto(m proto.Message) error - // RecvProto blocks until it receives a proto message or the stream is + // SendMsg is called by generated code. + SendMsg(m interface{}) error + // RecvMsg blocks until it receives a proto message or the stream is // done. On client side, it returns io.EOF when the stream is done. On // any other error, it aborts the streama nd returns an RPC status. On // server side, it simply returns the error to the caller. - RecvProto(m proto.Message) error + RecvMsg(m interface{}) error } // ClientStream defines the interface a client stream has to satify. @@ -146,7 +145,7 @@ func (cs *clientStream) Trailer() metadata.MD { return cs.s.Trailer() } -func (cs *clientStream) SendProto(m proto.Message) (err error) { +func (cs *clientStream) SendMsg(m interface{}) (err error) { defer func() { if err == nil || err == io.EOF { return @@ -156,21 +155,21 @@ func (cs *clientStream) SendProto(m proto.Message) (err error) { } err = toRPCErr(err) }() - out, err := encode(m, compressionNone) + out, err := encode(protoCodec{}, m, compressionNone) if err != nil { return transport.StreamErrorf(codes.Internal, "grpc: %v", err) } return cs.t.Write(cs.s, out, &transport.Options{Last: false}) } -func (cs *clientStream) RecvProto(m proto.Message) (err error) { - err = recvProto(cs.p, m) +func (cs *clientStream) RecvMsg(m interface{}) (err error) { + err = recv(cs.p, protoCodec{}, m) if err == nil { if !cs.desc.ClientStreams || cs.desc.ServerStreams { return } // Special handling for client streaming rpc. - err = recvProto(cs.p, m) + err = recv(cs.p, protoCodec{}, m) cs.t.CloseStream(cs.s, err) if err == nil { return toRPCErr(errors.New("grpc: client streaming protocol violation: get , want ")) @@ -245,8 +244,8 @@ func (ss *serverStream) SetTrailer(md metadata.MD) { return } -func (ss *serverStream) SendProto(m proto.Message) error { - out, err := encode(m, compressionNone) +func (ss *serverStream) SendMsg(m interface{}) error { + out, err := encode(protoCodec{}, m, compressionNone) if err != nil { err = transport.StreamErrorf(codes.Internal, "grpc: %v", err) return err @@ -254,6 +253,6 @@ func (ss *serverStream) SendProto(m proto.Message) error { return ss.t.Write(ss.s, out, &transport.Options{Last: false}) } -func (ss *serverStream) RecvProto(m proto.Message) error { - return recvProto(ss.p, m) +func (ss *serverStream) RecvMsg(m interface{}) error { + return recv(ss.p, protoCodec{}, m) } diff --git a/test/grpc_testing/test.pb.go b/test/grpc_testing/test.pb.go index 6246f1c9..d454e578 100644 --- a/test/grpc_testing/test.pb.go +++ b/test/grpc_testing/test.pb.go @@ -388,7 +388,7 @@ func (c *testServiceClient) StreamingOutputCall(ctx context.Context, in *Streami return nil, err } x := &testServiceStreamingOutputCallClient{stream} - if err := x.ClientStream.SendProto(in); err != nil { + if err := x.ClientStream.SendMsg(in); err != nil { return nil, err } if err := x.ClientStream.CloseSend(); err != nil { @@ -408,7 +408,7 @@ type testServiceStreamingOutputCallClient struct { func (x *testServiceStreamingOutputCallClient) Recv() (*StreamingOutputCallResponse, error) { m := new(StreamingOutputCallResponse) - if err := x.ClientStream.RecvProto(m); err != nil { + if err := x.ClientStream.RecvMsg(m); err != nil { return nil, err } return m, nil @@ -434,7 +434,7 @@ type testServiceStreamingInputCallClient struct { } func (x *testServiceStreamingInputCallClient) Send(m *StreamingInputCallRequest) error { - return x.ClientStream.SendProto(m) + return x.ClientStream.SendMsg(m) } func (x *testServiceStreamingInputCallClient) CloseAndRecv() (*StreamingInputCallResponse, error) { @@ -442,7 +442,7 @@ func (x *testServiceStreamingInputCallClient) CloseAndRecv() (*StreamingInputCal return nil, err } m := new(StreamingInputCallResponse) - if err := x.ClientStream.RecvProto(m); err != nil { + if err := x.ClientStream.RecvMsg(m); err != nil { return nil, err } return m, nil @@ -468,12 +468,12 @@ type testServiceFullDuplexCallClient struct { } func (x *testServiceFullDuplexCallClient) Send(m *StreamingOutputCallRequest) error { - return x.ClientStream.SendProto(m) + return x.ClientStream.SendMsg(m) } func (x *testServiceFullDuplexCallClient) Recv() (*StreamingOutputCallResponse, error) { m := new(StreamingOutputCallResponse) - if err := x.ClientStream.RecvProto(m); err != nil { + if err := x.ClientStream.RecvMsg(m); err != nil { return nil, err } return m, nil @@ -499,12 +499,12 @@ type testServiceHalfDuplexCallClient struct { } func (x *testServiceHalfDuplexCallClient) Send(m *StreamingOutputCallRequest) error { - return x.ClientStream.SendProto(m) + return x.ClientStream.SendMsg(m) } func (x *testServiceHalfDuplexCallClient) Recv() (*StreamingOutputCallResponse, error) { m := new(StreamingOutputCallResponse) - if err := x.ClientStream.RecvProto(m); err != nil { + if err := x.ClientStream.RecvMsg(m); err != nil { return nil, err } return m, nil @@ -539,7 +539,7 @@ func RegisterTestServiceServer(s *grpc.Server, srv TestServiceServer) { s.RegisterService(&_TestService_serviceDesc, srv) } -func _TestService_EmptyCall_Handler(srv interface{}, ctx context.Context, buf []byte) (proto.Message, error) { +func _TestService_EmptyCall_Handler(srv interface{}, ctx context.Context, buf []byte) (interface{}, error) { in := new(Empty) if err := proto.Unmarshal(buf, in); err != nil { return nil, err @@ -551,7 +551,7 @@ func _TestService_EmptyCall_Handler(srv interface{}, ctx context.Context, buf [] return out, nil } -func _TestService_UnaryCall_Handler(srv interface{}, ctx context.Context, buf []byte) (proto.Message, error) { +func _TestService_UnaryCall_Handler(srv interface{}, ctx context.Context, buf []byte) (interface{}, error) { in := new(SimpleRequest) if err := proto.Unmarshal(buf, in); err != nil { return nil, err @@ -565,7 +565,7 @@ func _TestService_UnaryCall_Handler(srv interface{}, ctx context.Context, buf [] func _TestService_StreamingOutputCall_Handler(srv interface{}, stream grpc.ServerStream) error { m := new(StreamingOutputCallRequest) - if err := stream.RecvProto(m); err != nil { + if err := stream.RecvMsg(m); err != nil { return err } return srv.(TestServiceServer).StreamingOutputCall(m, &testServiceStreamingOutputCallServer{stream}) @@ -581,7 +581,7 @@ type testServiceStreamingOutputCallServer struct { } func (x *testServiceStreamingOutputCallServer) Send(m *StreamingOutputCallResponse) error { - return x.ServerStream.SendProto(m) + return x.ServerStream.SendMsg(m) } func _TestService_StreamingInputCall_Handler(srv interface{}, stream grpc.ServerStream) error { @@ -599,12 +599,12 @@ type testServiceStreamingInputCallServer struct { } func (x *testServiceStreamingInputCallServer) SendAndClose(m *StreamingInputCallResponse) error { - return x.ServerStream.SendProto(m) + return x.ServerStream.SendMsg(m) } func (x *testServiceStreamingInputCallServer) Recv() (*StreamingInputCallRequest, error) { m := new(StreamingInputCallRequest) - if err := x.ServerStream.RecvProto(m); err != nil { + if err := x.ServerStream.RecvMsg(m); err != nil { return nil, err } return m, nil @@ -625,12 +625,12 @@ type testServiceFullDuplexCallServer struct { } func (x *testServiceFullDuplexCallServer) Send(m *StreamingOutputCallResponse) error { - return x.ServerStream.SendProto(m) + return x.ServerStream.SendMsg(m) } func (x *testServiceFullDuplexCallServer) Recv() (*StreamingOutputCallRequest, error) { m := new(StreamingOutputCallRequest) - if err := x.ServerStream.RecvProto(m); err != nil { + if err := x.ServerStream.RecvMsg(m); err != nil { return nil, err } return m, nil @@ -651,12 +651,12 @@ type testServiceHalfDuplexCallServer struct { } func (x *testServiceHalfDuplexCallServer) Send(m *StreamingOutputCallResponse) error { - return x.ServerStream.SendProto(m) + return x.ServerStream.SendMsg(m) } func (x *testServiceHalfDuplexCallServer) Recv() (*StreamingOutputCallRequest, error) { m := new(StreamingOutputCallRequest) - if err := x.ServerStream.RecvProto(m); err != nil { + if err := x.ServerStream.RecvMsg(m); err != nil { return nil, err } return m, nil From 2d13f822616fc95c8da121eb1fa95dad85c98d99 Mon Sep 17 00:00:00 2001 From: iamqizhao Date: Wed, 25 Mar 2015 15:29:01 -0700 Subject: [PATCH 2/2] fix a typo --- examples/route_guide/proto/route_guide.pb.go | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/examples/route_guide/proto/route_guide.pb.go b/examples/route_guide/proto/route_guide.pb.go index 6c21d0d7..c3d14b2c 100644 --- a/examples/route_guide/proto/route_guide.pb.go +++ b/examples/route_guide/proto/route_guide.pb.go @@ -17,8 +17,7 @@ It has these top-level messages: */ package proto -import proto "github.com/golang/protobuf/proto" -import math "math" +import proto1 "github.com/golang/protobuf/proto" import ( context "golang.org/x/net/context"