Method RecvMsg

stream.go:1497–1553

(m any)

}

func (as *addrConnStream) RecvMsg(m any) (err error) {
	defer func() {
		if err != nil || !as.desc.ServerStreams {
			// err != nil or non-server-streaming indicates end of stream.
			as.finish(err)
		}
	}()

	if !as.decompressorSet {
		// Block until we receive headers containing received message encoding.
		if ct := as.transportStream.RecvCompress(); ct != "" && ct != encoding.Identity {
			if as.decompressorV0 == nil || as.decompressorV0.Type() != ct {
				// No configured decompressor, or it does not match the incoming
				// message encoding; attempt to find a registered compressor that does.
				as.decompressorV0 = nil
				as.decompressorV1 = encoding.GetCompressor(ct)
			}
			// Validate that the compression method is acceptable for this call.
			if !acceptedCompressorAllows(as.callInfo.acceptedResponseCompressors, ct) {
				return status.Errorf(codes.Internal, "grpc: peer compressed the response with %q which is not allowed by AcceptCompressors", ct)
			}
		} else {
			// No compression is used; disable our decompressor.
			as.decompressorV0 = nil
		}
		// Only initialize this state once per stream.
		as.decompressorSet = true
	}
	if err := recv(&as.parser, as.codec, as.transportStream, as.decompressorV0, m, *as.callInfo.maxReceiveMessageSize, nil, as.decompressorV1, false); err != nil {
		if err == io.EOF {
			if statusErr := as.transportStream.Status().Err(); statusErr != nil {
				return statusErr
			}
			// Received no msg and status OK for non-server streaming rpcs.
			if !as.desc.ServerStreams && !as.receivedFirstMsg {
				return status.Error(codes.Internal, "cardinality violation: received no response message from non-server-streaming RPC")
			}
			return io.EOF // indicates successful end of stream.
		}
		return toRPCErr(err)
	}
	as.receivedFirstMsg = true

	if as.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(&as.parser, as.codec, as.transportStream, as.decompressorV0, m, *as.callInfo.maxReceiveMessageSize, nil, as.decompressorV1, false); err == io.EOF {
		return as.transportStream.Status().Err() // non-server streaming Recv returns nil on success
	} else if err != nil {
		return toRPCErr(err)
	}
	return status.Error(codes.Internal, "cardinality violation: expected <EOF> for non server-streaming RPCs, but received another message")
}