Update containerd.

Signed-off-by: Lantao Liu <lantaol@google.com>

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

@@ -19,7 +19,6 @@
 package grpc
 
 import (
-	"bytes"
 	"errors"
 	"io"
 	"sync"
@@ -29,8 +28,8 @@ import (
 	"golang.org/x/net/trace"
 	"google.golang.org/grpc/balancer"
 	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/encoding"
 	"google.golang.org/grpc/metadata"
-	"google.golang.org/grpc/peer"
 	"google.golang.org/grpc/stats"
 	"google.golang.org/grpc/status"
 	"google.golang.org/grpc/transport"
@@ -51,6 +50,8 @@ type StreamDesc struct {
 }
 
 // Stream defines the common interface a client or server stream has to satisfy.
+//
+// All errors returned from Stream are compatible with the status package.
 type Stream interface {
 	// Context returns the context for this stream.
 	Context() context.Context
@@ -89,41 +90,52 @@ type ClientStream interface {
 	// Stream.SendMsg() may return a non-nil error when something wrong happens sending
 	// the request. The returned error indicates the status of this sending, not the final
 	// status of the RPC.
-	// Always call Stream.RecvMsg() to get the final status if you care about the status of
-	// the RPC.
+	//
+	// Always call Stream.RecvMsg() to drain the stream and get the final
+	// status, otherwise there could be leaked resources.
 	Stream
 }
 
-// NewClientStream creates a new Stream for the client side. This is called
-// by generated code.
-func NewClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, opts ...CallOption) (_ ClientStream, err error) {
+// NewStream creates a new Stream for the client side. This is typically
+// called by generated code.
+func (cc *ClientConn) NewStream(ctx context.Context, desc *StreamDesc, method string, opts ...CallOption) (ClientStream, error) {
 	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.
+//
+// DEPRECATED: Use ClientConn.NewStream instead.
+func NewClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, opts ...CallOption) (ClientStream, error) {
+	return cc.NewStream(ctx, desc, method, opts...)
+}
+
 func newClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, opts ...CallOption) (_ ClientStream, err error) {
-	var (
-		t      transport.ClientTransport
-		s      *transport.Stream
-		done   func(balancer.DoneInfo)
-		cancel context.CancelFunc
-	)
 	c := defaultCallInfo()
 	mc := cc.GetMethodConfig(method)
 	if mc.WaitForReady != nil {
 		c.failFast = !*mc.WaitForReady
 	}
 
-	if mc.Timeout != nil {
+	// Possible context leak:
+	// The cancel function for the child context we create will only be called
+	// when RecvMsg returns a non-nil error, if the ClientConn is closed, or if
+	// an error is generated by SendMsg.
+	// https://github.com/grpc/grpc-go/issues/1818.
+	var cancel context.CancelFunc
+	if mc.Timeout != nil && *mc.Timeout >= 0 {
 		ctx, cancel = context.WithTimeout(ctx, *mc.Timeout)
-		defer func() {
-			if err != nil {
-				cancel()
-			}
-		}()
+	} else {
+		ctx, cancel = context.WithCancel(ctx)
 	}
+	defer func() {
+		if err != nil {
+			cancel()
+		}
+	}()
 
 	opts = append(cc.dopts.callOptions, opts...)
 	for _, o := range opts {
@@ -133,6 +145,9 @@ func newClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, meth
 	}
 	c.maxSendMessageSize = getMaxSize(mc.MaxReqSize, c.maxSendMessageSize, defaultClientMaxSendMessageSize)
 	c.maxReceiveMessageSize = getMaxSize(mc.MaxRespSize, c.maxReceiveMessageSize, defaultClientMaxReceiveMessageSize)
+	if err := setCallInfoCodec(c); err != nil {
+		return nil, err
+	}
 
 	callHdr := &transport.CallHdr{
 		Host:   cc.authority,
@@ -141,10 +156,27 @@ func newClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, meth
 		// so we don't flush the header.
 		// If it's client streaming, the user may never send a request or send it any
 		// time soon, so we ask the transport to flush the header.
-		Flush: desc.ClientStreams,
+		Flush:          desc.ClientStreams,
+		ContentSubtype: c.contentSubtype,
 	}
-	if cc.dopts.cp != nil {
+
+	// Set our outgoing compression according to the UseCompressor CallOption, if
+	// set.  In that case, also find the compressor from the encoding package.
+	// Otherwise, use the compressor configured by the WithCompressor DialOption,
+	// if set.
+	var cp Compressor
+	var comp encoding.Compressor
+	if ct := c.compressorType; ct != "" {
+		callHdr.SendCompress = ct
+		if ct != encoding.Identity {
+			comp = encoding.GetCompressor(ct)
+			if comp == nil {
+				return nil, status.Errorf(codes.Internal, "grpc: Compressor is not installed for requested grpc-encoding %q", ct)
+			}
+		}
+	} else if cc.dopts.cp != nil {
 		callHdr.SendCompress = cc.dopts.cp.Type()
+		cp = cc.dopts.cp
 	}
 	if c.creds != nil {
 		callHdr.Creds = c.creds
@@ -189,53 +221,53 @@ func newClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, meth
 			}
 		}()
 	}
+
+	var (
+		t    transport.ClientTransport
+		s    *transport.Stream
+		done func(balancer.DoneInfo)
+	)
 	for {
+		// Check to make sure the context has expired.  This will prevent us from
+		// looping forever if an error occurs for wait-for-ready RPCs where no data
+		// is sent on the wire.
+		select {
+		case <-ctx.Done():
+			return nil, toRPCErr(ctx.Err())
+		default:
+		}
+
 		t, done, err = cc.getTransport(ctx, c.failFast)
 		if err != nil {
-			// TODO(zhaoq): Probably revisit the error handling.
-			if _, ok := status.FromError(err); ok {
-				return nil, err
-			}
-			if err == errConnClosing || err == errConnUnavailable {
-				if c.failFast {
-					return nil, Errorf(codes.Unavailable, "%v", err)
-				}
-				continue
-			}
-			// All the other errors are treated as Internal errors.
-			return nil, Errorf(codes.Internal, "%v", err)
+			return nil, err
 		}
 
 		s, err = t.NewStream(ctx, callHdr)
 		if err != nil {
-			if _, ok := err.(transport.ConnectionError); ok && done != nil {
-				// If error is connection error, transport was sending data on wire,
-				// and we are not sure if anything has been sent on wire.
-				// If error is not connection error, we are sure nothing has been sent.
-				updateRPCInfoInContext(ctx, rpcInfo{bytesSent: true, bytesReceived: false})
-			}
 			if done != nil {
 				done(balancer.DoneInfo{Err: err})
 				done = nil
 			}
-			if _, ok := err.(transport.ConnectionError); (ok || err == transport.ErrStreamDrain) && !c.failFast {
+			// In the event of any error from NewStream, we never attempted to write
+			// anything to the wire, so we can retry indefinitely for non-fail-fast
+			// RPCs.
+			if !c.failFast {
 				continue
 			}
 			return nil, toRPCErr(err)
 		}
 		break
 	}
-	// Set callInfo.peer object from stream's context.
-	if peer, ok := peer.FromContext(s.Context()); ok {
-		c.peer = peer
-	}
+
+	c.stream = s
 	cs := &clientStream{
 		opts:   opts,
 		c:      c,
 		desc:   desc,
-		codec:  cc.dopts.codec,
-		cp:     cc.dopts.cp,
+		codec:  c.codec,
+		cp:     cp,
 		dc:     cc.dopts.dc,
+		comp:   comp,
 		cancel: cancel,
 
 		done: done,
@@ -249,50 +281,49 @@ func newClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, meth
 		statsCtx:     ctx,
 		statsHandler: cc.dopts.copts.StatsHandler,
 	}
-	// Listen on ctx.Done() to detect cancellation and s.Done() to detect normal termination
-	// when there is no pending I/O operations on this stream.
-	go func() {
-		select {
-		case <-t.Error():
-			// Incur transport error, simply exit.
-		case <-cc.ctx.Done():
-			cs.finish(ErrClientConnClosing)
-			cs.closeTransportStream(ErrClientConnClosing)
-		case <-s.Done():
-			// TODO: The trace of the RPC is terminated here when there is no pending
-			// I/O, which is probably not the optimal solution.
-			cs.finish(s.Status().Err())
-			cs.closeTransportStream(nil)
-		case <-s.GoAway():
-			cs.finish(errConnDrain)
-			cs.closeTransportStream(errConnDrain)
-		case <-s.Context().Done():
-			err := s.Context().Err()
-			cs.finish(err)
-			cs.closeTransportStream(transport.ContextErr(err))
-		}
-	}()
+	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
+		// should already be injected into the recv buffer by the transport, which
+		// the client will eventually receive, and then we will cancel the stream's
+		// context in clientStream.finish.
+		go func() {
+			select {
+			case <-cc.ctx.Done():
+				cs.finish(ErrClientConnClosing)
+			case <-ctx.Done():
+				cs.finish(toRPCErr(s.Context().Err()))
+			}
+		}()
+	}
 	return cs, nil
 }
 
 // clientStream implements a client side Stream.
 type clientStream struct {
-	opts   []CallOption
-	c      *callInfo
-	t      transport.ClientTransport
-	s      *transport.Stream
-	p      *parser
-	desc   *StreamDesc
-	codec  Codec
-	cp     Compressor
-	dc     Decompressor
+	opts []CallOption
+	c    *callInfo
+	t    transport.ClientTransport
+	s    *transport.Stream
+	p    *parser
+	desc *StreamDesc
+
+	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
 
 	tracing bool // set to EnableTracing when the clientStream is created.
 
 	mu       sync.Mutex
 	done     func(balancer.DoneInfo)
-	closed   bool
+	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.
@@ -312,9 +343,8 @@ func (cs *clientStream) Context() context.Context {
 func (cs *clientStream) Header() (metadata.MD, error) {
 	m, err := cs.s.Header()
 	if err != nil {
-		if _, ok := err.(transport.ConnectionError); !ok {
-			cs.closeTransportStream(err)
-		}
+		err = toRPCErr(err)
+		cs.finish(err)
 	}
 	return m, err
 }
@@ -324,6 +354,7 @@ func (cs *clientStream) Trailer() metadata.MD {
 }
 
 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 {
@@ -334,26 +365,18 @@ func (cs *clientStream) SendMsg(m interface{}) (err error) {
 	// TODO Investigate how to signal the stats handling party.
 	// generate error stats if err != nil && err != io.EOF?
 	defer func() {
-		if err != nil {
+		// 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)
 		}
-		if err == nil {
-			return
-		}
-		if err == io.EOF {
-			// Specialize the process for server streaming. SendMsg is only called
-			// once when creating the stream object. io.EOF needs to be skipped when
-			// the rpc is early finished (before the stream object is created.).
-			// TODO: It is probably better to move this into the generated code.
-			if !cs.desc.ClientStreams && cs.desc.ServerStreams {
-				err = nil
-			}
-			return
-		}
-		if _, ok := err.(transport.ConnectionError); !ok {
-			cs.closeTransportStream(err)
-		}
-		err = toRPCErr(err)
 	}()
 	var outPayload *stats.OutPayload
 	if cs.statsHandler != nil {
@@ -361,154 +384,143 @@ func (cs *clientStream) SendMsg(m interface{}) (err error) {
 			Client: true,
 		}
 	}
-	hdr, data, err := encode(cs.codec, m, cs.cp, bytes.NewBuffer([]byte{}), outPayload)
+	hdr, data, err := encode(cs.codec, m, cs.cp, outPayload, cs.comp)
 	if err != nil {
 		return err
 	}
-	if cs.c.maxSendMessageSize == nil {
-		return Errorf(codes.Internal, "callInfo maxSendMessageSize field uninitialized(nil)")
-	}
 	if len(data) > *cs.c.maxSendMessageSize {
-		return Errorf(codes.ResourceExhausted, "trying to send message larger than max (%d vs. %d)", 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)
 	}
-	err = cs.t.Write(cs.s, hdr, data, &transport.Options{Last: false})
-	if err == nil && outPayload != nil {
-		outPayload.SentTime = time.Now()
-		cs.statsHandler.HandleRPC(cs.statsCtx, outPayload)
+	if !cs.desc.ClientStreams {
+		cs.sentLast = true
 	}
-	return err
+	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
 }
 
 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.c.maxReceiveMessageSize == nil {
-		return Errorf(codes.Internal, "callInfo maxReceiveMessageSize field uninitialized(nil)")
+	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)
-	defer func() {
-		// err != nil indicates the termination of the stream.
-		if err != nil {
-			cs.finish(err)
-		}
-	}()
-	if err == nil {
-		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.ClientStreams || cs.desc.ServerStreams {
-			return
-		}
-		// Special handling for client streaming rpc.
-		// This recv expects EOF or errors, so we don't collect inPayload.
-		if cs.c.maxReceiveMessageSize == nil {
-			return Errorf(codes.Internal, "callInfo maxReceiveMessageSize field uninitialized(nil)")
-		}
-		err = recv(cs.p, cs.codec, cs.s, cs.dc, m, *cs.c.maxReceiveMessageSize, nil)
-		cs.closeTransportStream(err)
-		if err == nil {
-			return toRPCErr(errors.New("grpc: client streaming protocol violation: get <nil>, want <EOF>"))
-		}
+	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 se := cs.s.Status().Err(); se != nil {
-				return se
+			if statusErr := cs.s.Status().Err(); statusErr != nil {
+				return statusErr
 			}
-			cs.finish(err)
-			return nil
+			return io.EOF // indicates successful end of stream.
 		}
 		return toRPCErr(err)
 	}
-	if _, ok := err.(transport.ConnectionError); !ok {
-		cs.closeTransportStream(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 {
-		if statusErr := cs.s.Status().Err(); statusErr != nil {
-			return statusErr
-		}
-		// Returns io.EOF to indicate the end of the stream.
-		return
+		return cs.s.Status().Err() // non-server streaming Recv returns nil on success
 	}
 	return toRPCErr(err)
 }
 
-func (cs *clientStream) CloseSend() (err error) {
-	err = cs.t.Write(cs.s, nil, nil, &transport.Options{Last: true})
-	defer func() {
-		if err != nil {
-			cs.finish(err)
-		}
-	}()
-	if err == nil || err == io.EOF {
+func (cs *clientStream) CloseSend() error {
+	if cs.sentLast {
 		return nil
 	}
-	if _, ok := err.(transport.ConnectionError); !ok {
-		cs.closeTransportStream(err)
-	}
-	err = toRPCErr(err)
-	return
-}
-
-func (cs *clientStream) closeTransportStream(err error) {
-	cs.mu.Lock()
-	if cs.closed {
-		cs.mu.Unlock()
-		return
-	}
-	cs.closed = true
-	cs.mu.Unlock()
-	cs.t.CloseStream(cs.s, err)
+	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.
+	return nil
 }
 
 func (cs *clientStream) finish(err error) {
+	if err == io.EOF {
+		// Ending a stream with EOF indicates a success.
+		err = nil
+	}
 	cs.mu.Lock()
 	defer cs.mu.Unlock()
 	if cs.finished {
 		return
 	}
 	cs.finished = true
-	defer func() {
-		if cs.cancel != nil {
-			cs.cancel()
-		}
-	}()
+	cs.t.CloseStream(cs.s, err)
 	for _, o := range cs.opts {
 		o.after(cs.c)
 	}
 	if cs.done != nil {
-		updateRPCInfoInContext(cs.s.Context(), rpcInfo{
-			bytesSent:     cs.s.BytesSent(),
-			bytesReceived: cs.s.BytesReceived(),
+		cs.done(balancer.DoneInfo{
+			Err:           err,
+			BytesSent:     true,
+			BytesReceived: cs.s.BytesReceived(),
 		})
-		cs.done(balancer.DoneInfo{Err: err})
 		cs.done = nil
 	}
 	if cs.statsHandler != nil {
 		end := &stats.End{
 			Client:  true,
 			EndTime: time.Now(),
-		}
-		if err != io.EOF {
-			// end.Error is nil if the RPC finished successfully.
-			end.Error = toRPCErr(err)
+			Error:   err,
 		}
 		cs.statsHandler.HandleRPC(cs.statsCtx, end)
 	}
+	cs.cancel()
 	if !cs.tracing {
 		return
 	}
 	if cs.trInfo.tr != nil {
-		if err == nil || err == io.EOF {
+		if err == nil {
 			cs.trInfo.tr.LazyPrintf("RPC: [OK]")
 		} else {
 			cs.trInfo.tr.LazyPrintf("RPC: [%v]", err)
@@ -540,12 +552,16 @@ type ServerStream interface {
 
 // serverStream implements a server side Stream.
 type serverStream struct {
-	t                     transport.ServerTransport
-	s                     *transport.Stream
-	p                     *parser
-	codec                 Codec
-	cp                    Compressor
-	dc                    Decompressor
+	t     transport.ServerTransport
+	s     *transport.Stream
+	p     *parser
+	codec baseCodec
+
+	cp     Compressor
+	dc     Decompressor
+	comp   encoding.Compressor
+	decomp encoding.Compressor
+
 	maxReceiveMessageSize int
 	maxSendMessageSize    int
 	trInfo                *traceInfo
@@ -601,12 +617,12 @@ func (ss *serverStream) SendMsg(m interface{}) (err error) {
 	if ss.statsHandler != nil {
 		outPayload = &stats.OutPayload{}
 	}
-	hdr, data, err := encode(ss.codec, m, ss.cp, bytes.NewBuffer([]byte{}), outPayload)
+	hdr, data, err := encode(ss.codec, m, ss.cp, outPayload, ss.comp)
 	if err != nil {
 		return err
 	}
 	if len(data) > ss.maxSendMessageSize {
-		return Errorf(codes.ResourceExhausted, "trying to send message larger than max (%d vs. %d)", len(data), ss.maxSendMessageSize)
+		return status.Errorf(codes.ResourceExhausted, "trying to send message larger than max (%d vs. %d)", len(data), ss.maxSendMessageSize)
 	}
 	if err := ss.t.Write(ss.s, hdr, data, &transport.Options{Last: false}); err != nil {
 		return toRPCErr(err)
@@ -641,12 +657,12 @@ func (ss *serverStream) RecvMsg(m interface{}) (err error) {
 	if ss.statsHandler != nil {
 		inPayload = &stats.InPayload{}
 	}
-	if err := recv(ss.p, ss.codec, ss.s, ss.dc, m, ss.maxReceiveMessageSize, inPayload); err != nil {
+	if err := recv(ss.p, ss.codec, ss.s, ss.dc, m, ss.maxReceiveMessageSize, inPayload, ss.decomp); err != nil {
 		if err == io.EOF {
 			return err
 		}
 		if err == io.ErrUnexpectedEOF {
-			err = Errorf(codes.Internal, io.ErrUnexpectedEOF.Error())
+			err = status.Errorf(codes.Internal, io.ErrUnexpectedEOF.Error())
 		}
 		return toRPCErr(err)
 	}
@@ -655,3 +671,13 @@ func (ss *serverStream) RecvMsg(m interface{}) (err error) {
 	}
 	return nil
 }
+
+// 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
+}