Method HandleStreams

internal/transport/handler_server.go:392–467

(ctx context.Context, startStream func(*ServerStream))

}

func (ht *serverHandlerTransport) HandleStreams(ctx context.Context, startStream func(*ServerStream)) {
	// With this transport type there will be exactly 1 stream: this HTTP request.
	var cancel context.CancelFunc
	if ht.timeoutSet {
		ctx, cancel = context.WithTimeout(ctx, ht.timeout)
	} else {
		ctx, cancel = context.WithCancel(ctx)
	}

	// requestOver is closed when the status has been written via WriteStatus.
	requestOver := make(chan struct{})
	go func() {
		select {
		case <-requestOver:
		case <-ht.closedCh:
		case <-ht.req.Context().Done():
		}
		cancel()
		ht.Close(errors.New("request is done processing"))
	}()

	ctx = metadata.NewIncomingContext(ctx, ht.headerMD)
	req := ht.req
	s := &ServerStream{
		Stream: Stream{
			id:             0, // irrelevant
			ctx:            ctx,
			method:         req.URL.Path,
			recvCompress:   req.Header.Get("grpc-encoding"),
			contentSubtype: ht.contentSubtype,
		},
		cancel:           cancel,
		st:               ht,
		headerWireLength: 0, // won't have access to header wire length until golang/go#18997.
	}
	s.Stream.buf.init()
	s.readRequester = s
	s.trReader = transportReader{
		reader:        recvBufferReader{ctx: s.ctx, ctxDone: s.ctx.Done(), recv: &s.buf},
		windowHandler: s,
	}

	// readerDone is closed when the Body.Read-ing goroutine exits.
	readerDone := make(chan struct{})
	go func() {
		defer close(readerDone)

		for {
			buf := ht.bufferPool.Get(http2MaxFrameLen)
			n, err := req.Body.Read(*buf)
			if n > 0 {
				*buf = (*buf)[:n]
				s.buf.put(recvMsg{buffer: mem.NewBuffer(buf, ht.bufferPool)})
			} else {
				ht.bufferPool.Put(buf)
			}
			if err != nil {
				s.buf.put(recvMsg{err: mapRecvMsgError(err)})