vendor update for CSI 0.3.0

vendor/google.golang.org/grpc/stream.go

@@ -29,6 +29,7 @@ import (
 	"google.golang.org/grpc/balancer"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/encoding"
+	"google.golang.org/grpc/internal/channelz"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/stats"
 	"google.golang.org/grpc/status"
@@ -36,7 +37,10 @@ import (
 )
 
 // StreamHandler defines the handler called by gRPC server to complete the
-// execution of a streaming RPC.
+// execution of a streaming RPC. If a StreamHandler returns an error, it
+// should be produced by the status package, or else gRPC will use
+// codes.Unknown as the status code and err.Error() as the status message
+// of the RPC.
 type StreamHandler func(srv interface{}, stream ServerStream) error
 
 // StreamDesc represents a streaming RPC service's method specification.
@@ -97,16 +101,33 @@ type ClientStream interface {
 }
 
 // NewStream creates a new Stream for the client side. This is typically
-// called by generated code.
+// called by generated code. ctx is used for the lifetime of the stream.
+//
+// To ensure resources are not leaked due to the stream returned, one of the following
+// actions must be performed:
+//
+//      1. Call Close on the ClientConn.
+//      2. Cancel the context provided.
+//      3. Call RecvMsg until a non-nil error is returned. A protobuf-generated
+//         client-streaming RPC, for instance, might use the helper function
+//         CloseAndRecv (note that CloseSend does not Recv, therefore is not
+//         guaranteed to release all resources).
+//      4. Receive a non-nil, non-io.EOF error from Header or SendMsg.
+//
+// If none of the above happen, a goroutine and a context will be leaked, and grpc
+// will not call the optionally-configured stats handler with a stats.End message.
 func (cc *ClientConn) NewStream(ctx context.Context, desc *StreamDesc, method string, opts ...CallOption) (ClientStream, error) {
+	// allow interceptor to see all applicable call options, which means those
+	// configured as defaults from dial option as well as per-call options
+	opts = combine(cc.dopts.callOptions, opts)
+
 	if cc.dopts.streamInt != nil {
 		return cc.dopts.streamInt(ctx, desc, cc, method, newClientStream, opts...)
 	}
 	return newClientStream(ctx, desc, cc, method, opts...)
 }
 
-// NewClientStream creates a new Stream for the client side. This is typically
-// called by generated code.
+// NewClientStream is a wrapper for ClientConn.NewStream.
 //
 // DEPRECATED: Use ClientConn.NewStream instead.
 func NewClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, opts ...CallOption) (ClientStream, error) {
@@ -114,6 +135,14 @@ func NewClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, meth
 }
 
 func newClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, opts ...CallOption) (_ ClientStream, err error) {
+	if channelz.IsOn() {
+		cc.incrCallsStarted()
+		defer func() {
+			if err != nil {
+				cc.incrCallsFailed()
+			}
+		}()
+	}
 	c := defaultCallInfo()
 	mc := cc.GetMethodConfig(method)
 	if mc.WaitForReady != nil {
@@ -137,7 +166,6 @@ func newClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, meth
 		}
 	}()
 
-	opts = append(cc.dopts.callOptions, opts...)
 	for _, o := range opts {
 		if err := o.before(c); err != nil {
 			return nil, toRPCErr(err)
@@ -202,11 +230,13 @@ func newClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, meth
 	}
 	ctx = newContextWithRPCInfo(ctx, c.failFast)
 	sh := cc.dopts.copts.StatsHandler
+	var beginTime time.Time
 	if sh != nil {
 		ctx = sh.TagRPC(ctx, &stats.RPCTagInfo{FullMethodName: method, FailFast: c.failFast})
+		beginTime = time.Now()
 		begin := &stats.Begin{
 			Client:    true,
-			BeginTime: time.Now(),
+			BeginTime: beginTime,
 			FailFast:  c.failFast,
 		}
 		sh.HandleRPC(ctx, begin)
@@ -214,8 +244,10 @@ func newClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, meth
 			if err != nil {
 				// Only handle end stats if err != nil.
 				end := &stats.End{
-					Client: true,
-					Error:  err,
+					Client:    true,
+					Error:     err,
+					BeginTime: beginTime,
+					EndTime:   time.Now(),
 				}
 				sh.HandleRPC(ctx, end)
 			}
@@ -259,28 +291,29 @@ func newClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, meth
 		break
 	}
 
-	c.stream = s
 	cs := &clientStream{
 		opts:   opts,
 		c:      c,
+		cc:     cc,
 		desc:   desc,
 		codec:  c.codec,
 		cp:     cp,
-		dc:     cc.dopts.dc,
 		comp:   comp,
 		cancel: cancel,
-
-		done: done,
-		t:    t,
-		s:    s,
-		p:    &parser{r: s},
-
-		tracing: EnableTracing,
-		trInfo:  trInfo,
-
-		statsCtx:     ctx,
-		statsHandler: cc.dopts.copts.StatsHandler,
+		attempt: &csAttempt{
+			t:            t,
+			s:            s,
+			p:            &parser{r: s},
+			done:         done,
+			dc:           cc.dopts.dc,
+			ctx:          ctx,
+			trInfo:       trInfo,
+			statsHandler: sh,
+			beginTime:    beginTime,
+		},
 	}
+	cs.c.stream = cs
+	cs.attempt.cs = cs
 	if desc != unaryStreamDesc {
 		// Listen on cc and stream contexts to cleanup when the user closes the
 		// ClientConn or cancels the stream context.  In all other cases, an error
@@ -292,7 +325,7 @@ func newClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, meth
 			case <-cc.ctx.Done():
 				cs.finish(ErrClientConnClosing)
 			case <-ctx.Done():
-				cs.finish(toRPCErr(s.Context().Err()))
+				cs.finish(toRPCErr(ctx.Err()))
 			}
 		}()
 	}
@@ -303,46 +336,57 @@ func newClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, meth
 type clientStream struct {
 	opts []CallOption
 	c    *callInfo
+	cc   *ClientConn
+	desc *StreamDesc
+
+	codec baseCodec
+	cp    Compressor
+	comp  encoding.Compressor
+
+	cancel context.CancelFunc // cancels all attempts
+
+	sentLast bool // sent an end stream
+
+	mu       sync.Mutex // guards finished
+	finished bool       // TODO: replace with atomic cmpxchg or sync.Once?
+
+	attempt *csAttempt // the active client stream attempt
+	// TODO(hedging): hedging will have multiple attempts simultaneously.
+}
+
+// csAttempt implements a single transport stream attempt within a
+// clientStream.
+type csAttempt struct {
+	cs   *clientStream
 	t    transport.ClientTransport
 	s    *transport.Stream
 	p    *parser
-	desc *StreamDesc
+	done func(balancer.DoneInfo)
 
-	codec     baseCodec
-	cp        Compressor
 	dc        Decompressor
-	comp      encoding.Compressor
 	decomp    encoding.Compressor
 	decompSet bool
 
-	// cancel is only called when RecvMsg() returns non-nil error, which means
-	// the stream finishes with error or with io.EOF.
-	cancel context.CancelFunc
+	ctx context.Context // the application's context, wrapped by stats/tracing
 
-	tracing bool // set to EnableTracing when the clientStream is created.
-
-	mu       sync.Mutex
-	done     func(balancer.DoneInfo)
-	sentLast bool // sent an end stream
-	finished bool
-	// trInfo.tr is set when the clientStream is created (if EnableTracing is true),
-	// and is set to nil when the clientStream's finish method is called.
+	mu sync.Mutex // guards trInfo.tr
+	// trInfo.tr is set when created (if EnableTracing is true),
+	// and cleared when the finish method is called.
 	trInfo traceInfo
 
-	// statsCtx keeps the user context for stats handling.
-	// All stats collection should use the statsCtx (instead of the stream context)
-	// so that all the generated stats for a particular RPC can be associated in the processing phase.
-	statsCtx     context.Context
 	statsHandler stats.Handler
+	beginTime    time.Time
 }
 
 func (cs *clientStream) Context() context.Context {
-	return cs.s.Context()
+	// TODO(retry): commit the current attempt (the context has peer-aware data).
+	return cs.attempt.context()
 }
 
 func (cs *clientStream) Header() (metadata.MD, error) {
-	m, err := cs.s.Header()
+	m, err := cs.attempt.header()
 	if err != nil {
+		// TODO(retry): maybe retry on error or commit attempt on success.
 		err = toRPCErr(err)
 		cs.finish(err)
 	}
@@ -350,137 +394,22 @@ func (cs *clientStream) Header() (metadata.MD, error) {
 }
 
 func (cs *clientStream) Trailer() metadata.MD {
-	return cs.s.Trailer()
+	// TODO(retry): on error, maybe retry (trailers-only).
+	return cs.attempt.trailer()
 }
 
 func (cs *clientStream) SendMsg(m interface{}) (err error) {
-	// TODO: Check cs.sentLast and error if we already ended the stream.
-	if cs.tracing {
-		cs.mu.Lock()
-		if cs.trInfo.tr != nil {
-			cs.trInfo.tr.LazyLog(&payload{sent: true, msg: m}, true)
-		}
-		cs.mu.Unlock()
-	}
-	// TODO Investigate how to signal the stats handling party.
-	// generate error stats if err != nil && err != io.EOF?
-	defer func() {
-		// For non-client-streaming RPCs, we return nil instead of EOF on success
-		// because the generated code requires it.  finish is not called; RecvMsg()
-		// will call it with the stream's status independently.
-		if err == io.EOF && !cs.desc.ClientStreams {
-			err = nil
-		}
-		if err != nil && err != io.EOF {
-			// Call finish for errors generated by this SendMsg call.  (Transport
-			// errors are converted to an io.EOF error below; the real error will be
-			// returned from RecvMsg eventually in that case.)
-			cs.finish(err)
-		}
-	}()
-	var outPayload *stats.OutPayload
-	if cs.statsHandler != nil {
-		outPayload = &stats.OutPayload{
-			Client: true,
-		}
-	}
-	hdr, data, err := encode(cs.codec, m, cs.cp, outPayload, cs.comp)
-	if err != nil {
-		return err
-	}
-	if len(data) > *cs.c.maxSendMessageSize {
-		return status.Errorf(codes.ResourceExhausted, "trying to send message larger than max (%d vs. %d)", len(data), *cs.c.maxSendMessageSize)
-	}
-	if !cs.desc.ClientStreams {
-		cs.sentLast = true
-	}
-	err = cs.t.Write(cs.s, hdr, data, &transport.Options{Last: !cs.desc.ClientStreams})
-	if err == nil {
-		if outPayload != nil {
-			outPayload.SentTime = time.Now()
-			cs.statsHandler.HandleRPC(cs.statsCtx, outPayload)
-		}
-		return nil
-	}
-	return io.EOF
+	// TODO(retry): buffer message for replaying if not committed.
+	return cs.attempt.sendMsg(m)
 }
 
 func (cs *clientStream) RecvMsg(m interface{}) (err error) {
-	defer func() {
-		if err != nil || !cs.desc.ServerStreams {
-			// err != nil or non-server-streaming indicates end of stream.
-			cs.finish(err)
-		}
-	}()
-	var inPayload *stats.InPayload
-	if cs.statsHandler != nil {
-		inPayload = &stats.InPayload{
-			Client: true,
-		}
-	}
-	if !cs.decompSet {
-		// Block until we receive headers containing received message encoding.
-		if ct := cs.s.RecvCompress(); ct != "" && ct != encoding.Identity {
-			if cs.dc == nil || cs.dc.Type() != ct {
-				// No configured decompressor, or it does not match the incoming
-				// message encoding; attempt to find a registered compressor that does.
-				cs.dc = nil
-				cs.decomp = encoding.GetCompressor(ct)
-			}
-		} else {
-			// No compression is used; disable our decompressor.
-			cs.dc = nil
-		}
-		// Only initialize this state once per stream.
-		cs.decompSet = true
-	}
-	err = recv(cs.p, cs.codec, cs.s, cs.dc, m, *cs.c.maxReceiveMessageSize, inPayload, cs.decomp)
-	if err != nil {
-		if err == io.EOF {
-			if statusErr := cs.s.Status().Err(); statusErr != nil {
-				return statusErr
-			}
-			return io.EOF // indicates successful end of stream.
-		}
-		return toRPCErr(err)
-	}
-	if cs.tracing {
-		cs.mu.Lock()
-		if cs.trInfo.tr != nil {
-			cs.trInfo.tr.LazyLog(&payload{sent: false, msg: m}, true)
-		}
-		cs.mu.Unlock()
-	}
-	if inPayload != nil {
-		cs.statsHandler.HandleRPC(cs.statsCtx, inPayload)
-	}
-	if cs.desc.ServerStreams {
-		// Subsequent messages should be received by subsequent RecvMsg calls.
-		return nil
-	}
-
-	// Special handling for non-server-stream rpcs.
-	// This recv expects EOF or errors, so we don't collect inPayload.
-	err = recv(cs.p, cs.codec, cs.s, cs.dc, m, *cs.c.maxReceiveMessageSize, nil, cs.decomp)
-	if err == nil {
-		return toRPCErr(errors.New("grpc: client streaming protocol violation: get <nil>, want <EOF>"))
-	}
-	if err == io.EOF {
-		return cs.s.Status().Err() // non-server streaming Recv returns nil on success
-	}
-	return toRPCErr(err)
+	// TODO(retry): maybe retry on error or commit attempt on success.
+	return cs.attempt.recvMsg(m)
 }
 
 func (cs *clientStream) CloseSend() error {
-	if cs.sentLast {
-		return nil
-	}
-	cs.sentLast = true
-	cs.t.Write(cs.s, nil, nil, &transport.Options{Last: true})
-	// We ignore errors from Write and always return nil here.  Any error it
-	// would return would also be returned by a subsequent RecvMsg call, and the
-	// user is supposed to always finish the stream by calling RecvMsg until it
-	// returns err != nil.
+	cs.attempt.closeSend()
 	return nil
 }
 
@@ -490,45 +419,209 @@ func (cs *clientStream) finish(err error) {
 		err = nil
 	}
 	cs.mu.Lock()
-	defer cs.mu.Unlock()
 	if cs.finished {
+		cs.mu.Unlock()
 		return
 	}
 	cs.finished = true
-	cs.t.CloseStream(cs.s, err)
+	cs.mu.Unlock()
+	if channelz.IsOn() {
+		if err != nil {
+			cs.cc.incrCallsFailed()
+		} else {
+			cs.cc.incrCallsSucceeded()
+		}
+	}
+	// TODO(retry): commit current attempt if necessary.
+	cs.attempt.finish(err)
 	for _, o := range cs.opts {
 		o.after(cs.c)
 	}
-	if cs.done != nil {
-		cs.done(balancer.DoneInfo{
-			Err:           err,
-			BytesSent:     true,
-			BytesReceived: cs.s.BytesReceived(),
-		})
-		cs.done = nil
-	}
-	if cs.statsHandler != nil {
-		end := &stats.End{
-			Client:  true,
-			EndTime: time.Now(),
-			Error:   err,
-		}
-		cs.statsHandler.HandleRPC(cs.statsCtx, end)
-	}
 	cs.cancel()
-	if !cs.tracing {
+}
+
+func (a *csAttempt) context() context.Context {
+	return a.s.Context()
+}
+
+func (a *csAttempt) header() (metadata.MD, error) {
+	return a.s.Header()
+}
+
+func (a *csAttempt) trailer() metadata.MD {
+	return a.s.Trailer()
+}
+
+func (a *csAttempt) sendMsg(m interface{}) (err error) {
+	// TODO Investigate how to signal the stats handling party.
+	// generate error stats if err != nil && err != io.EOF?
+	cs := a.cs
+	defer func() {
+		// For non-client-streaming RPCs, we return nil instead of EOF on success
+		// because the generated code requires it.  finish is not called; RecvMsg()
+		// will call it with the stream's status independently.
+		if err == io.EOF && !cs.desc.ClientStreams {
+			err = nil
+		}
+		if err != nil && err != io.EOF {
+			// Call finish on the client stream for errors generated by this SendMsg
+			// call, as these indicate problems created by this client.  (Transport
+			// errors are converted to an io.EOF error below; the real error will be
+			// returned from RecvMsg eventually in that case, or be retried.)
+			cs.finish(err)
+		}
+	}()
+	// TODO: Check cs.sentLast and error if we already ended the stream.
+	if EnableTracing {
+		a.mu.Lock()
+		if a.trInfo.tr != nil {
+			a.trInfo.tr.LazyLog(&payload{sent: true, msg: m}, true)
+		}
+		a.mu.Unlock()
+	}
+	data, err := encode(cs.codec, m)
+	if err != nil {
+		return err
+	}
+	compData, err := compress(data, cs.cp, cs.comp)
+	if err != nil {
+		return err
+	}
+	hdr, payload := msgHeader(data, compData)
+	// TODO(dfawley): should we be checking len(data) instead?
+	if len(payload) > *cs.c.maxSendMessageSize {
+		return status.Errorf(codes.ResourceExhausted, "trying to send message larger than max (%d vs. %d)", len(payload), *cs.c.maxSendMessageSize)
+	}
+
+	if !cs.desc.ClientStreams {
+		cs.sentLast = true
+	}
+	err = a.t.Write(a.s, hdr, payload, &transport.Options{Last: !cs.desc.ClientStreams})
+	if err == nil {
+		if a.statsHandler != nil {
+			a.statsHandler.HandleRPC(a.ctx, outPayload(true, m, data, payload, time.Now()))
+		}
+		if channelz.IsOn() {
+			a.t.IncrMsgSent()
+		}
+		return nil
+	}
+	return io.EOF
+}
+
+func (a *csAttempt) recvMsg(m interface{}) (err error) {
+	cs := a.cs
+	defer func() {
+		if err != nil || !cs.desc.ServerStreams {
+			// err != nil or non-server-streaming indicates end of stream.
+			cs.finish(err)
+		}
+	}()
+	var inPayload *stats.InPayload
+	if a.statsHandler != nil {
+		inPayload = &stats.InPayload{
+			Client: true,
+		}
+	}
+	if !a.decompSet {
+		// Block until we receive headers containing received message encoding.
+		if ct := a.s.RecvCompress(); ct != "" && ct != encoding.Identity {
+			if a.dc == nil || a.dc.Type() != ct {
+				// No configured decompressor, or it does not match the incoming
+				// message encoding; attempt to find a registered compressor that does.
+				a.dc = nil
+				a.decomp = encoding.GetCompressor(ct)
+			}
+		} else {
+			// No compression is used; disable our decompressor.
+			a.dc = nil
+		}
+		// Only initialize this state once per stream.
+		a.decompSet = true
+	}
+	err = recv(a.p, cs.codec, a.s, a.dc, m, *cs.c.maxReceiveMessageSize, inPayload, a.decomp)
+	if err != nil {
+		if err == io.EOF {
+			if statusErr := a.s.Status().Err(); statusErr != nil {
+				return statusErr
+			}
+			return io.EOF // indicates successful end of stream.
+		}
+		return toRPCErr(err)
+	}
+	if EnableTracing {
+		a.mu.Lock()
+		if a.trInfo.tr != nil {
+			a.trInfo.tr.LazyLog(&payload{sent: false, msg: m}, true)
+		}
+		a.mu.Unlock()
+	}
+	if inPayload != nil {
+		a.statsHandler.HandleRPC(a.ctx, inPayload)
+	}
+	if channelz.IsOn() {
+		a.t.IncrMsgRecv()
+	}
+	if cs.desc.ServerStreams {
+		// Subsequent messages should be received by subsequent RecvMsg calls.
+		return nil
+	}
+
+	// Special handling for non-server-stream rpcs.
+	// This recv expects EOF or errors, so we don't collect inPayload.
+	err = recv(a.p, cs.codec, a.s, a.dc, m, *cs.c.maxReceiveMessageSize, nil, a.decomp)
+	if err == nil {
+		return toRPCErr(errors.New("grpc: client streaming protocol violation: get <nil>, want <EOF>"))
+	}
+	if err == io.EOF {
+		return a.s.Status().Err() // non-server streaming Recv returns nil on success
+	}
+	return toRPCErr(err)
+}
+
+func (a *csAttempt) closeSend() {
+	cs := a.cs
+	if cs.sentLast {
 		return
 	}
-	if cs.trInfo.tr != nil {
-		if err == nil {
-			cs.trInfo.tr.LazyPrintf("RPC: [OK]")
-		} else {
-			cs.trInfo.tr.LazyPrintf("RPC: [%v]", err)
-			cs.trInfo.tr.SetError()
-		}
-		cs.trInfo.tr.Finish()
-		cs.trInfo.tr = nil
+	cs.sentLast = true
+	cs.attempt.t.Write(cs.attempt.s, nil, nil, &transport.Options{Last: true})
+	// We ignore errors from Write.  Any error it would return would also be
+	// returned by a subsequent RecvMsg call, and the user is supposed to always
+	// finish the stream by calling RecvMsg until it returns err != nil.
+}
+
+func (a *csAttempt) finish(err error) {
+	a.mu.Lock()
+	a.t.CloseStream(a.s, err)
+
+	if a.done != nil {
+		a.done(balancer.DoneInfo{
+			Err:           err,
+			BytesSent:     true,
+			BytesReceived: a.s.BytesReceived(),
+		})
 	}
+	if a.statsHandler != nil {
+		end := &stats.End{
+			Client:    true,
+			BeginTime: a.beginTime,
+			EndTime:   time.Now(),
+			Error:     err,
+		}
+		a.statsHandler.HandleRPC(a.ctx, end)
+	}
+	if a.trInfo.tr != nil {
+		if err == nil {
+			a.trInfo.tr.LazyPrintf("RPC: [OK]")
+		} else {
+			a.trInfo.tr.LazyPrintf("RPC: [%v]", err)
+			a.trInfo.tr.SetError()
+		}
+		a.trInfo.tr.Finish()
+		a.trInfo.tr = nil
+	}
+	a.mu.Unlock()
 }
 
 // ServerStream defines the interface a server stream has to satisfy.
@@ -552,6 +645,7 @@ type ServerStream interface {
 
 // serverStream implements a server side Stream.
 type serverStream struct {
+	ctx   context.Context
 	t     transport.ServerTransport
 	s     *transport.Stream
 	p     *parser
@@ -572,7 +666,7 @@ type serverStream struct {
 }
 
 func (ss *serverStream) Context() context.Context {
-	return ss.s.Context()
+	return ss.ctx
 }
 
 func (ss *serverStream) SetHeader(md metadata.MD) error {
@@ -591,7 +685,6 @@ func (ss *serverStream) SetTrailer(md metadata.MD) {
 		return
 	}
 	ss.s.SetTrailer(md)
-	return
 }
 
 func (ss *serverStream) SendMsg(m interface{}) (err error) {
@@ -612,24 +705,28 @@ func (ss *serverStream) SendMsg(m interface{}) (err error) {
 			st, _ := status.FromError(toRPCErr(err))
 			ss.t.WriteStatus(ss.s, st)
 		}
+		if channelz.IsOn() && err == nil {
+			ss.t.IncrMsgSent()
+		}
 	}()
-	var outPayload *stats.OutPayload
-	if ss.statsHandler != nil {
-		outPayload = &stats.OutPayload{}
-	}
-	hdr, data, err := encode(ss.codec, m, ss.cp, outPayload, ss.comp)
+	data, err := encode(ss.codec, m)
 	if err != nil {
 		return err
 	}
-	if len(data) > ss.maxSendMessageSize {
-		return status.Errorf(codes.ResourceExhausted, "trying to send message larger than max (%d vs. %d)", len(data), ss.maxSendMessageSize)
+	compData, err := compress(data, ss.cp, ss.comp)
+	if err != nil {
+		return err
 	}
-	if err := ss.t.Write(ss.s, hdr, data, &transport.Options{Last: false}); err != nil {
+	hdr, payload := msgHeader(data, compData)
+	// TODO(dfawley): should we be checking len(data) instead?
+	if len(payload) > ss.maxSendMessageSize {
+		return status.Errorf(codes.ResourceExhausted, "trying to send message larger than max (%d vs. %d)", len(payload), ss.maxSendMessageSize)
+	}
+	if err := ss.t.Write(ss.s, hdr, payload, &transport.Options{Last: false}); err != nil {
 		return toRPCErr(err)
 	}
-	if outPayload != nil {
-		outPayload.SentTime = time.Now()
-		ss.statsHandler.HandleRPC(ss.s.Context(), outPayload)
+	if ss.statsHandler != nil {
+		ss.statsHandler.HandleRPC(ss.s.Context(), outPayload(false, m, data, payload, time.Now()))
 	}
 	return nil
 }
@@ -652,6 +749,9 @@ func (ss *serverStream) RecvMsg(m interface{}) (err error) {
 			st, _ := status.FromError(toRPCErr(err))
 			ss.t.WriteStatus(ss.s, st)
 		}
+		if channelz.IsOn() && err == nil {
+			ss.t.IncrMsgRecv()
+		}
 	}()
 	var inPayload *stats.InPayload
 	if ss.statsHandler != nil {
@@ -675,9 +775,5 @@ func (ss *serverStream) RecvMsg(m interface{}) (err error) {
 // MethodFromServerStream returns the method string for the input stream.
 // The returned string is in the format of "/service/method".
 func MethodFromServerStream(stream ServerStream) (string, bool) {
-	s, ok := transport.StreamFromContext(stream.Context())
-	if !ok {
-		return "", ok
-	}
-	return s.Method(), ok
+	return Method(stream.Context())
 }