opensource/grpc-go
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #136 from iamqizhao/non-proto

Browse Source 
Interface change to accommodate the support of non-protobuf data formats
This commit is contained in:
Qi Zhao
2015-03-25 15:41:37 -07:00
parent 0bcb969b1c 2d13f82261
commit f8cb2ac6cf
9 changed files with 129 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
benchmark/grpc_testing/test.pb.go
View File

@ -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

19
call.go
View File

@ -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
}

26
examples/route_guide/proto/route_guide.pb.go
View File

@ -186,7 +186,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 +206,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 +232,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 +240,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 +266,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 +310,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 +324,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 +340,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 +358,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 +384,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

36
interop/grpc_testing/test.pb.go
View File

@ -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

34
rpc_util.go
View File

@ -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:

8
rpc_util_test.go
View File

@ -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)
}

11
server.go
View File

@ -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
}

29
stream.go
View File

@ -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 <nil>, want <EOF>"))
@ -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)
}

36
test/grpc_testing/test.pb.go
View File

@ -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