Method handleStream

server.go:1800–1891

(t transport.ServerTransport, stream *transport.ServerStream)

}

func (s *Server) handleStream(t transport.ServerTransport, stream *transport.ServerStream) {
	ctx := stream.Context()
	ctx = contextWithServer(ctx, s)
	if envconfig.LabelServerGoroutines&envconfig.GoroutineLabelServerMethod != 0 {
		// This method always runs in its own goroutine, so we can set a
		// goroutine label without needing to restore a previous context.
		ctx = pprof.WithLabels(ctx, pprof.Labels("grpc.method", stream.Method()))
		pprof.SetGoroutineLabels(ctx)
	}
	var ti *traceInfo
	if EnableTracing {
		tr := newTrace("grpc.Recv."+methodFamily(stream.Method()), stream.Method())
		ctx = newTraceContext(ctx, tr)
		ti = &traceInfo{
			tr: tr,
			firstLine: firstLine{
				client:     false,
				remoteAddr: t.Peer().Addr,
			},
		}
		if dl, ok := ctx.Deadline(); ok {
			ti.firstLine.deadline = time.Until(dl)
		}
	}

	sm, found := strings.CutPrefix(stream.Method(), "/")
	if !found {
		s.handleMalformedMethodName(stream, ti)
		return
	}
	pos := strings.LastIndex(sm, "/")
	if pos == -1 {
		s.handleMalformedMethodName(stream, ti)
		return
	}
	service := sm[:pos]
	method := sm[pos+1:]

	// FromIncomingContext is expensive: skip if there are no statsHandlers
	if s.statsHandler != nil {
		md, _ := metadata.FromIncomingContext(ctx)
		ctx = s.statsHandler.TagRPC(ctx, &stats.RPCTagInfo{FullMethodName: stream.Method()})
		s.statsHandler.HandleRPC(ctx, &stats.InHeader{
			FullMethod:  stream.Method(),
			RemoteAddr:  t.Peer().Addr,
			LocalAddr:   t.Peer().LocalAddr,
			Compression: stream.RecvCompress(),
			WireLength:  stream.HeaderWireLength(),
			Header:      md,
		})
	}
	// To have calls in stream callouts work. Will delete once all stats handler
	// calls come from the gRPC layer.
	stream.SetContext(ctx)

	srv, knownService := s.services[service]
	if knownService {
		if md, ok := srv.methods[method]; ok {