async def METHOD_NAME(message_type):
    """The Endpoint.close_after_n_recv()"""
    async def server_node(ep):
        for _ in range(10):
            await _shutdown_send(ep, message_type)
    async def client_node(port):
        ep = await ucp.create_endpoint(
            ucp.get_address(),
            port,
        )
        ep.close_after_n_recv(10)
        for _ in range(10):
            await _shutdown_recv(ep, message_type)
        assert ep.closed()
        ep = await ucp.create_endpoint(
            ucp.get_address(),
            port,
        )
        for _ in range(5):
            await _shutdown_recv(ep, message_type)
        ep.close_after_n_recv(5)
        for _ in range(5):
            await _shutdown_recv(ep, message_type)
        assert ep.closed()
        ep = await ucp.create_endpoint(
            ucp.get_address(),
            port,
        )
        for _ in range(5):
            await _shutdown_recv(ep, message_type)
        ep.close_after_n_recv(10, count_from_ep_creation=True)
        for _ in range(5):
            await _shutdown_recv(ep, message_type)
        assert ep.closed()
        ep = await ucp.create_endpoint(
            ucp.get_address(),
            port,
        )
        for _ in range(10):
            await _shutdown_recv(ep, message_type)
        with pytest.raises(
            ucp.exceptions.UCXError,
            match="`n` cannot be less than current recv_count",
        ):
            ep.close_after_n_recv(5, count_from_ep_creation=True)
        ep.close_after_n_recv(1)
        with pytest.raises(
            ucp.exceptions.UCXError,
            match="close_after_n_recv has already been set to",
        ):
            ep.close_after_n_recv(1)
    listener = ucp.create_listener(
        server_node,
    )
    await client_node(listener.port)