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

renaming

Browse Source
This commit is contained in:
Menghan Li
2016-11-04 16:02:19 -07:00
parent aa5b5c7e2a
commit a4edc462a2
4 changed files with 36 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
call.go
View File

@ -80,7 +80,7 @@ func recvResponse(ctx context.Context, dopts dialOptions, t transport.ClientTran
} }
} }
if inPayload != nil && err == io.EOF && stream.StatusCode() == codes.OK { if inPayload != nil && err == io.EOF && stream.StatusCode() == codes.OK {
// TODO in the current implementation, inTrailer may be handled before inStats in some cases. // TODO in the current implementation, inTrailer may be handled before inPayload in some cases.
// Fix the order if necessary. // Fix the order if necessary.
stats.Handle(ctx, inPayload) stats.Handle(ctx, inPayload)
} }

30
rpc_util.go
View File

@ -257,7 +257,7 @@ func (p *parser) recvMsg(maxMsgSize int) (pf payloadFormat, msg []byte, err erro
// encode serializes msg and prepends the message header. If msg is nil, it // encode serializes msg and prepends the message header. If msg is nil, it
// generates the message header of 0 message length. // generates the message header of 0 message length.
func encode(c Codec, msg interface{}, cp Compressor, cbuf *bytes.Buffer, outStats *stats.OutPayload) ([]byte, error) { func encode(c Codec, msg interface{}, cp Compressor, cbuf *bytes.Buffer, outPayload *stats.OutPayload) ([]byte, error) {
var ( var (
b []byte b []byte
length uint length uint
@ -269,10 +269,10 @@ func encode(c Codec, msg interface{}, cp Compressor, cbuf *bytes.Buffer, outStat
if err != nil { if err != nil {
return nil, err return nil, err
} }
if outStats != nil { if outPayload != nil {
outStats.Payload = msg outPayload.Payload = msg
outStats.Data = b outPayload.Data = b
outStats.Length = len(b) outPayload.Length = len(b)
} }
if cp != nil { if cp != nil {
if err := cp.Do(cbuf, b); err != nil { if err := cp.Do(cbuf, b); err != nil {
@ -304,8 +304,8 @@ func encode(c Codec, msg interface{}, cp Compressor, cbuf *bytes.Buffer, outStat
// Copy encoded msg to buf // Copy encoded msg to buf
copy(buf[5:], b) copy(buf[5:], b)
if outStats != nil { if outPayload != nil {
outStats.WireLength = len(buf) outPayload.WireLength = len(buf)
} }
return buf, nil return buf, nil
@ -324,14 +324,14 @@ func checkRecvPayload(pf payloadFormat, recvCompress string, dc Decompressor) er
return nil return nil
} }
func recv(p *parser, c Codec, s *transport.Stream, dc Decompressor, m interface{}, maxMsgSize int, inStats *stats.InPayload) error { func recv(p *parser, c Codec, s *transport.Stream, dc Decompressor, m interface{}, maxMsgSize int, inPayload *stats.InPayload) error {
pf, d, err := p.recvMsg(maxMsgSize) pf, d, err := p.recvMsg(maxMsgSize)
if err != nil { if err != nil {
return err return err
} }
if inStats != nil { if inPayload != nil {
inStats.RecvTime = time.Now() inPayload.RecvTime = time.Now()
inStats.WireLength = len(d) inPayload.WireLength = len(d)
} }
if err := checkRecvPayload(pf, s.RecvCompress(), dc); err != nil { if err := checkRecvPayload(pf, s.RecvCompress(), dc); err != nil {
return err return err
@ -350,10 +350,10 @@ func recv(p *parser, c Codec, s *transport.Stream, dc Decompressor, m interface{
if err := c.Unmarshal(d, m); err != nil { if err := c.Unmarshal(d, m); err != nil {
return Errorf(codes.Internal, "grpc: failed to unmarshal the received message %v", err) return Errorf(codes.Internal, "grpc: failed to unmarshal the received message %v", err)
} }
if inStats != nil { if inPayload != nil {
inStats.Payload = m inPayload.Payload = m
inStats.Data = d inPayload.Data = d
inStats.Length = len(d) inPayload.Length = len(d)
} }
return nil return nil
} }

18
server.go
View File

@ -620,9 +620,9 @@ func (s *Server) processUnaryRPC(t transport.ServerTransport, stream *transport.
p := &parser{r: stream} p := &parser{r: stream}
for { for {
pf, req, err := p.recvMsg(s.opts.maxMsgSize) pf, req, err := p.recvMsg(s.opts.maxMsgSize)
var inStats *stats.InPayload var inPayload *stats.InPayload
if stats.On() { if stats.On() {
inStats = &stats.InPayload{ inPayload = &stats.InPayload{
RecvTime: time.Now(), RecvTime: time.Now(),
} }
@ -670,8 +670,8 @@ func (s *Server) processUnaryRPC(t transport.ServerTransport, stream *transport.
statusCode := codes.OK statusCode := codes.OK
statusDesc := "" statusDesc := ""
df := func(v interface{}) error { df := func(v interface{}) error {
if inStats != nil { if inPayload != nil {
inStats.WireLength = len(req) inPayload.WireLength = len(req)
} }
if pf == compressionMade { if pf == compressionMade {
var err error var err error
@ -692,11 +692,11 @@ func (s *Server) processUnaryRPC(t transport.ServerTransport, stream *transport.
if err := s.opts.codec.Unmarshal(req, v); err != nil { if err := s.opts.codec.Unmarshal(req, v); err != nil {
return err return err
} }
if inStats != nil { if inPayload != nil {
inStats.Payload = v inPayload.Payload = v
inStats.Data = req inPayload.Data = req
inStats.Length = len(req) inPayload.Length = len(req)
stats.Handle(stream.Context(), inStats) stats.Handle(stream.Context(), inPayload)
} }
if trInfo != nil { if trInfo != nil {
trInfo.tr.LazyLog(&payload{sent: false, msg: v}, true) trInfo.tr.LazyLog(&payload{sent: false, msg: v}, true)

22
stream.go
View File

@ -382,13 +382,13 @@ func (cs *clientStream) RecvMsg(m interface{}) (err error) {
stats.Handle(cs.userCtx, end) stats.Handle(cs.userCtx, end)
} }
}() }()
var inStats *stats.InPayload var inPayload *stats.InPayload
if stats.On() { if stats.On() {
inStats = &stats.InPayload{ inPayload = &stats.InPayload{
Client: true, Client: true,
} }
} }
err = recv(cs.p, cs.codec, cs.s, cs.dc, m, math.MaxInt32, inStats) err = recv(cs.p, cs.codec, cs.s, cs.dc, m, math.MaxInt32, inPayload)
defer func() { defer func() {
// err != nil indicates the termination of the stream. // err != nil indicates the termination of the stream.
if err != nil { if err != nil {
@ -403,14 +403,14 @@ func (cs *clientStream) RecvMsg(m interface{}) (err error) {
} }
cs.mu.Unlock() cs.mu.Unlock()
} }
if inStats != nil { if inPayload != nil {
stats.Handle(cs.userCtx, inStats) stats.Handle(cs.userCtx, inPayload)
} }
if !cs.desc.ClientStreams || cs.desc.ServerStreams { if !cs.desc.ClientStreams || cs.desc.ServerStreams {
return return
} }
// Special handling for client streaming rpc. // Special handling for client streaming rpc.
// This recv expects EOF or errors, so we don't collect inStats. // This recv expects EOF or errors, so we don't collect inPayload.
err = recv(cs.p, cs.codec, cs.s, cs.dc, m, math.MaxInt32, nil) err = recv(cs.p, cs.codec, cs.s, cs.dc, m, math.MaxInt32, nil)
cs.closeTransportStream(err) cs.closeTransportStream(err)
if err == nil { if err == nil {
@ -606,11 +606,11 @@ func (ss *serverStream) RecvMsg(m interface{}) (err error) {
ss.mu.Unlock() ss.mu.Unlock()
} }
}() }()
var inStats *stats.InPayload var inPayload *stats.InPayload
if stats.On() { if stats.On() {
inStats = &stats.InPayload{} inPayload = &stats.InPayload{}
} }
if err := recv(ss.p, ss.codec, ss.s, ss.dc, m, ss.maxMsgSize, inStats); err != nil { if err := recv(ss.p, ss.codec, ss.s, ss.dc, m, ss.maxMsgSize, inPayload); err != nil {
if err == io.EOF { if err == io.EOF {
return err return err
} }
@ -619,8 +619,8 @@ func (ss *serverStream) RecvMsg(m interface{}) (err error) {
} }
return toRPCErr(err) return toRPCErr(err)
} }
if inStats != nil { if inPayload != nil {
stats.Handle(ss.s.Context(), inStats) stats.Handle(ss.s.Context(), inPayload)
} }
return nil return nil
} }