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

Code before:
				// message encoding; attempt to find a registered compressor that does.
				a.decompressorV0 = nil
				a.decompressorV1 = encoding.GetCompressor(ct)
			}
		} else {
			// No compression is used; disable our decompressor.
			a.decompressorV0 = nil
		}
		// Only initialize this state once per stream.
		a.decompressorSet = true
	}
	if err := recv(a.parser, cs.codec, a.transportStream, a.decompressorV0, m, *cs.callInfo.maxReceiveMessageSize, payInfo, a.decompressorV1, false); err != nil {
		if err == io.EOF {
			if statusErr := a.transportStream.Status().Err(); statusErr != nil {
				return statusErr
			}
			if !cs.desc.ServerStreams && !cs.recvFirstMsg {
				return status.Errorf(codes.Internal, "client streaming cardinality violation")
			}
			return io.EOF // indicates successful end of stream.
		}

		return toRPCErr(err)
	}
	if !cs.desc.ServerStreams {
		cs.recvFirstMsg = true
	}
	if a.trInfo != nil {
		a.mu.Lock()
		if a.trInfo.tr != nil {
			a.trInfo.tr.LazyLog(&payload{sent: false, msg: m}, true)
		}
		a.mu.Unlock()
	}
	for _, sh := range a.statsHandlers {
		sh.HandleRPC(a.ctx, &stats.InPayload{
			Client:           true,
			RecvTime:         time.Now(),
			Payload:          m,
			WireLength:       payInfo.compressedLength + headerLen,
			CompressedLength: payInfo.compressedLength,
			Length:           payInfo.uncompressedBytes.Len(),
		})
	}
	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.
	if err := recv(a.parser, cs.codec, a.transportStream, a.decompressorV0, m, *cs.callInfo.maxReceiveMessageSize, nil, a.decompressorV1, false); err == io.EOF {

Review comment (bug):
Should there also be something here that fails if `recvFirstMsg` is already set?  (That would indicate we received multiple messages in a unary RPC.)

Code after:
			if a.decompressorV0 == nil || a.decompressorV0.Type() != ct {
				// No configured decompressor, or it does not match the incoming
				// message encoding; attempt to find a registered compressor that does.
				a.decompressorV0 = nil
				a.decompressorV1 = encoding.GetCompressor(ct)
			}
		} else {
			// No compression is used; disable our decompressor.
			a.decompressorV0 = nil
		}
		// Only initialize this state once per stream.
		a.decompressorSet = true
	}
	if err := recv(a.parser, cs.codec, a.transportStream, a.decompressorV0, m, *cs.callInfo.maxReceiveMessageSize, payInfo, a.decompressorV1, false); err != nil {
		if err == io.EOF {
			if statusErr := a.transportStream.Status().Err(); statusErr != nil {
				return statusErr
			}
			// Received no msg and status OK for non-server streaming rpcs.
			if !cs.desc.ServerStreams {
				return status.Error(codes.Internal, "cardinality violation: received no response message from non-streaming RPC")
			}
			return io.EOF // indicates successful end of stream.
		}

		return toRPCErr(err)
	}
	if a.trInfo != nil {
		a.mu.Lock()
		if a.trInfo.tr != nil {
			a.trInfo.tr.LazyLog(&payload{sent: false, msg: m}, true)
		}
		a.mu.Unlock()
	}
	for _, sh := range a.statsHandlers {
		sh.HandleRPC(a.ctx, &stats.InPayload{
			Client:           true,
			RecvTime:         time.Now(),
			Payload:          m,
			WireLength:       payInfo.compressedLength + headerLen,
			CompressedLength: payInfo.compressedLength,
			Length:           payInfo.uncompressedBytes.Len(),
		})
	}
	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.
	if err := recv(a.parser, cs.codec, a.transportStream, a.decompressorV0, m, *cs.callInfo.maxReceiveMessageSize, nil, a.decompressorV1, false); err == io.EOF {