Repository: grpc/grpc-go
File: stream.go (Go)

Code before:
	} else if ac.cc.dopts.compressorV0 != nil {
		callHdr.SendCompress = ac.cc.dopts.compressorV0.Type()
		cp = ac.cc.dopts.compressorV0
	}
	if c.creds != nil {
		callHdr.Creds = c.creds
	}

	// Use a special addrConnStream to avoid retry.
	as := &addrConnStream{
		callHdr:          callHdr,
		ac:               ac,
		ctx:              ctx,
		cancel:           cancel,
		opts:             opts,
		callInfo:         c,
		desc:             desc,
		codec:            c.codec,
		sendCompressorV0: cp,
		sendCompressorV1: comp,
		decompressorV0:   ac.cc.dopts.dc,
		transport:        t,
	}

	s, err := as.transport.NewStream(as.ctx, as.callHdr, nil)
	if err != nil {
		err = toRPCErr(err)
		return nil, err
	}
	as.transportStream = s
	as.parser = parser{r: s, bufferPool: ac.dopts.copts.BufferPool}
	ac.incrCallsStarted()
	if desc != unaryStreamDesc {
		// Listen on stream context to cleanup when the stream context is
		// canceled.  Also listen for the addrConn's context in case the
		// addrConn is closed or reconnects to a different address.  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
		// addrConnStream.finish.
		go func() {
			ac.mu.Lock()
			acCtx := ac.ctx
			ac.mu.Unlock()
			select {
			case <-acCtx.Done():
				as.finish(status.Error(codes.Canceled, "grpc: the SubConn is closing"))
			case <-ctx.Done():
				as.finish(toRPCErr(ctx.Err()))
			}
		}()

Review comment (bug):
Is this how the health and ORCA streams get a `nil` stats handler? We want to eliminate this duplication between the regular client stream and the non-retry client stream at some point. We tried to prioritize that work for Q1, but it didn't happen. This has been a maintenance burden for a while now, and a source of hard to find bugs.

Code after:
	} else if ac.cc.dopts.compressorV0 != nil {
		callHdr.SendCompress = ac.cc.dopts.compressorV0.Type()
		cp = ac.cc.dopts.compressorV0
	}
	if c.creds != nil {
		callHdr.Creds = c.creds
	}

	// Use a special addrConnStream to avoid retry.
	as := &addrConnStream{
		callHdr:          callHdr,
		ac:               ac,
		ctx:              ctx,
		cancel:           cancel,
		opts:             opts,
		callInfo:         c,
		desc:             desc,
		codec:            c.codec,
		sendCompressorV0: cp,
		sendCompressorV1: comp,
		decompressorV0:   ac.cc.dopts.dc,
		transport:        t,
	}

	// nil stats handler: internal streams like health and ORCA do not support telemetry.
	s, err := as.transport.NewStream(as.ctx, as.callHdr, nil)
	if err != nil {
		err = toRPCErr(err)
		return nil, err
	}
	as.transportStream = s
	as.parser = parser{r: s, bufferPool: ac.dopts.copts.BufferPool}
	ac.incrCallsStarted()
	if desc != unaryStreamDesc {
		// Listen on stream context to cleanup when the stream context is
		// canceled.  Also listen for the addrConn's context in case the
		// addrConn is closed or reconnects to a different address.  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
		// addrConnStream.finish.
		go func() {
			ac.mu.Lock()
			acCtx := ac.ctx
			ac.mu.Unlock()
			select {
			case <-acCtx.Done():
				as.finish(status.Error(codes.Canceled, "grpc: the SubConn is closing"))
			case <-ctx.Done():
				as.finish(toRPCErr(ctx.Err()))
			}