abstract class MigrationBase in Migrate 7.2

Same name and namespace in other branches

6.2 includes/base.inc \MigrationBase

The base class for all objects representing distinct steps in a migration process. Most commonly these will be Migration objects which actually import data from a source into a Drupal destination, but by deriving classes directly from MigrationBase one can have other sorts of tasks (e.g., enabling/disabling of modules) occur during the migration process.

Hierarchy

class \MigrationBase

Expanded class hierarchy of MigrationBase

1 string reference to 'MigrationBase'

migrate_migrations in ./migrate.module: Retrieve a list of all active migrations, ordered by dependencies. To be recognized, a class must be non-abstract, and derived from MigrationBase.

File

includes/base.inc, line 15: Defines the base class for migration processes.

View source

abstract class MigrationBase {

  /**
   * Track the migration currently running, so handlers can easily determine it
   * without having to pass a Migration object everywhere.
   *
   * @var Migration
   */
  protected static $currentMigration;
  public static function currentMigration() {
    return self::$currentMigration;
  }

  /**
   * The machine name of this Migration object, derived by removing the
   * 'Migration' suffix from the class name. Used to construct default
   * map/message table names, displayed in drush migrate-status, key to
   * migrate_status table...
   *
   * @var string
   */
  protected $machineName;
  public function getMachineName() {
    return $this->machineName;
  }

  /**
   * A migration group object, used to collect related migrations.
   *
   * @var MigrateGroup
   */
  protected $group;
  public function getGroup() {
    return $this->group;
  }

  /**
   * Detailed information describing the migration.
   *
   * @var string
   */
  protected $description;
  public function getDescription() {
    return $this->description;
  }
  public function setDescription($description) {
    $this->description = $description;
  }

  /**
   * Save options passed to current operation
   *
   * @var array
   */
  protected $options;
  public function getOption($option_name) {
    if (isset($this->options[$option_name])) {
      return $this->options[$option_name];
    }
    else {
      return NULL;
    }
  }
  public function getItemLimit() {
    if (isset($this->options['limit']) && ($this->options['limit']['unit'] == 'items' || $this->options['limit']['unit'] == 'item')) {
      return $this->options['limit']['value'];
    }
    else {
      return NULL;
    }
  }
  public function getTimeLimit() {
    if (isset($this->options['limit']) && ($this->options['limit']['unit'] == 'seconds' || $this->options['limit']['unit'] == 'second')) {
      return $this->options['limit']['value'];
    }
    else {
      return NULL;
    }
  }

  /**
   * Indicates that we are processing a rollback or import - used to avoid
   * excess writes in endProcess()
   *
   * @var boolean
   */
  protected $processing = FALSE;

  /**
   * Are we importing, rolling back, or doing nothing?
   *
   * @var enum
   */
  protected $status = MigrationBase::STATUS_IDLE;

  /**
   * When the current operation started.
   *
   * @var int
   */
  protected $starttime;

  /**
   * Whether to maintain a history of migration processes in migrate_log
   *
   * @var boolean
   */
  protected $logHistory = TRUE;

  /**
   * Primary key of the current history record (inserted at the beginning of
   * a process, to be updated at the end)
   *
   * @var int
   */
  protected $logID;

  /**
   * Number of "items" processed in the current migration process (whatever that
   * means for the type of process)
   *
   * @var int
   */
  protected $total_processed = 0;

  /**
   * List of other Migration classes which should be imported before this one.
   * E.g., a comment migration class would typically have node and user
   * migrations as dependencies.
   *
   * @var array
   */
  protected $dependencies = array(), $softDependencies = array();
  public function getHardDependencies() {
    return $this->dependencies;
  }
  public function setHardDependencies(array $dependencies) {
    $this->dependencies = $dependencies;
  }
  public function addHardDependencies(array $dependencies) {
    $this->dependencies = array_merge($this->dependencies, $dependencies);
  }
  public function getSoftDependencies() {
    return $this->softDependencies;
  }
  public function setSoftDependencies(array $dependencies) {
    $this->softDependencies = $dependencies;
  }
  public function addSoftDependencies(array $dependencies) {
    $this->softDependencies = array_merge($this->softDependencies, $dependencies);
  }
  public function getDependencies() {
    return array_merge($this->dependencies, $this->softDependencies);
  }

  /**
   * Name of a function for displaying feedback. It must take the message to
   * display as its first argument, and a (string) message type as its second
   * argument
   * (see drush_log()).
   *
   * @var string
   */
  protected static $displayFunction;
  public static function setDisplayFunction($display_function) {
    self::$displayFunction = $display_function;
  }

  /**
   * Track whether or not we've already displayed an encryption warning
   *
   * @var bool
   */
  protected static $showEncryptionWarning = TRUE;

  /**
   * The fraction of the memory limit at which an operation will be interrupted.
   * Can be overridden by a Migration subclass if one would like to push the
   * envelope. Defaults to 85%.
   *
   * @var float
   */
  protected $memoryThreshold = 0.85;

  /**
   * The PHP memory_limit expressed in bytes.
   *
   * @var int
   */
  protected $memoryLimit;

  /**
   * The fraction of the time limit at which an operation will be interrupted.
   * Can be overridden by a Migration subclass if one would like to push the
   * envelope. Defaults to 90%.
   *
   * @var float
   */
  protected $timeThreshold = 0.9;

  /**
   * The PHP max_execution_time.
   *
   * @var int
   */
  protected $timeLimit;

  /**
   * A time limit in seconds appropriate to be used in a batch
   * import. Defaults to 240.
   *
   * @var int
   */
  protected $batchTimeLimit = 240;

  /**
   * MigrateTeamMember objects representing people involved with this
   * migration.
   *
   * @var array
   */
  protected $team = array();
  public function getTeam() {
    return $this->team;
  }
  public function setTeam(array $team) {
    $this->team = $team;
  }

  /**
   * If provided, an URL for an issue tracking system containing :id where
   * the issue number will go (e.g., 'http://example.com/project/ticket/:id').
   *
   * @var string
   */
  protected $issuePattern;
  public function getIssuePattern() {
    return $this->issuePattern;
  }
  public function setIssuePattern($issue_pattern) {
    $this->issuePattern = $issue_pattern;
  }

  /**
   * If we set an error handler (during import), remember the previous one so
   * it can be restored.
   *
   * @var callback
   */
  protected $previousErrorHandler = NULL;

  /**
   * Arguments configuring a migration.
   *
   * @var array
   */
  protected $arguments = array();
  public function getArguments() {
    return $this->arguments;
  }
  public function setArguments(array $arguments) {
    $this->arguments = $arguments;
  }
  public function addArguments(array $arguments) {
    $this->arguments = array_merge($this->arguments, $arguments);
  }

  /**
   * Disabling a migration prevents it from running with --all, or individually
   * without --force
   *
   * @var boolean
   */
  protected $enabled = TRUE;
  public function getEnabled() {
    return $this->enabled;
  }
  public function setEnabled($enabled) {
    $this->enabled = $enabled;
  }

  /**
   * Any module hooks which should be disabled during migration processes.
   *
   * @var array
   *  Key: Hook name (e.g., 'node_insert')
   *  Value: Array of modules for which to disable this hook (e.g.,
   *   array('pathauto')).
   */
  protected $disableHooks = array();
  public function getDisableHooks() {
    return $this->disableHooks;
  }

  /**
   * An array to track 'mail_system' variable if disabled.
   */
  protected $mailSystem;

  /**
   * Have we already warned about obsolete constructor argumentss on this
   * request?
   *
   * @var bool
   */
  protected static $groupArgumentWarning = FALSE;
  protected static $emptyArgumentsWarning = FALSE;

  /**
   * Codes representing the result of a rollback or import process.
   */
  const RESULT_COMPLETED = 1;

  // All records have been processed
  const RESULT_INCOMPLETE = 2;

  // The process has interrupted itself (e.g., the
  // memory limit is approaching)
  const RESULT_STOPPED = 3;

  // The process was stopped externally (e.g., via
  // drush migrate-stop)
  const RESULT_FAILED = 4;

  // The process had a fatal error
  const RESULT_SKIPPED = 5;

  // Dependencies are unfulfilled - skip the process
  const RESULT_DISABLED = 6;

  // This migration is disabled, skipping

  /**
   * Codes representing the current status of a migration, and stored in the
   * migrate_status table.
   */
  const STATUS_IDLE = 0;
  const STATUS_IMPORTING = 1;
  const STATUS_ROLLING_BACK = 2;
  const STATUS_STOPPING = 3;
  const STATUS_DISABLED = 4;

  /**
   * Message types to be passed to saveMessage() and saved in message tables.
   * MESSAGE_INFORMATIONAL represents a condition that did not prevent the
   * operation from succeeding - all others represent different severities of
   * conditions resulting in a source record not being imported.
   */
  const MESSAGE_ERROR = 1;
  const MESSAGE_WARNING = 2;
  const MESSAGE_NOTICE = 3;
  const MESSAGE_INFORMATIONAL = 4;

  /**
   * Get human readable name for a message constant.
   *
   * @return string
   *  Name.
   */
  public function getMessageLevelName($constant) {
    $map = array(
      MigrationBase::MESSAGE_ERROR => t('Error'),
      MigrationBase::MESSAGE_WARNING => t('Warning'),
      MigrationBase::MESSAGE_NOTICE => t('Notice'),
      MigrationBase::MESSAGE_INFORMATIONAL => t('Informational'),
    );
    return $map[$constant];
  }

  /**
   * Construction of a MigrationBase instance.
   *
   * @param array $arguments
   */
  public function __construct($arguments = array()) {

    // Support for legacy code passing a group object as the first parameter.
    if (is_object($arguments) && is_a($arguments, 'MigrateGroup')) {
      $this->group = $arguments;
      $this->arguments['group_name'] = $arguments
        ->getName();
      if (!self::$groupArgumentWarning && variable_get('migrate_deprecation_warnings', 1)) {
        self::displayMessage(t('Passing a group object to a migration constructor is now deprecated - pass through the arguments array passed to the leaf class instead.'));
        self::$groupArgumentWarning = TRUE;
      }
    }
    else {
      if (empty($arguments)) {
        $this->arguments = array();
        if (!self::$emptyArgumentsWarning && variable_get('migrate_deprecation_warnings', 1)) {
          self::displayMessage(t('Passing an empty first parameter to a migration constructor is now deprecated - pass through the arguments array passed to the leaf class instead.'));
          self::$emptyArgumentsWarning = TRUE;
        }
      }
      else {
        $this->arguments = $arguments;
      }
      if (empty($this->arguments['group_name'])) {
        $this->arguments['group_name'] = 'default';
      }
      $this->group = MigrateGroup::getInstance($this->arguments['group_name']);
    }
    if (isset($this->arguments['machine_name'])) {
      $this->machineName = $this->arguments['machine_name'];
    }
    else {

      // Deprecated - this supports old code which does not pass the arguments
      // array through to the base constructor. Remove in the next version.
      $this->machineName = $this
        ->machineFromClass(get_class($this));
    }

    // Make any group arguments directly accessible to the specific migration,
    // other than group dependencies.
    $group_arguments = $this->group
      ->getArguments();
    unset($group_arguments['dependencies']);
    $this->arguments += $group_arguments;

    // Record the memory limit in bytes
    $limit = trim(ini_get('memory_limit'));
    if ($limit == '-1') {
      $this->memoryLimit = PHP_INT_MAX;
    }
    else {
      if (!is_numeric($limit)) {
        $last = drupal_strtolower($limit[strlen($limit) - 1]);
        $limit = substr($limit, 0, -1);
        switch ($last) {
          case 'g':
            $limit *= 1024;
          case 'm':
            $limit *= 1024;
          case 'k':
            $limit *= 1024;
            break;
          default:
            throw new Exception(t('Invalid PHP memory_limit !limit', array(
              '!limit' => $limit,
            )));
        }
      }
      $this->memoryLimit = $limit;
    }

    // Record the time limit
    $this->timeLimit = ini_get('max_execution_time');

    // Make sure we clear our semaphores in case of abrupt exit
    drupal_register_shutdown_function(array(
      $this,
      'endProcess',
    ));

    // Save any hook disablement information.
    if (isset($this->arguments['disable_hooks']) && is_array($this->arguments['disable_hooks'])) {
      $this->disableHooks = $this->arguments['disable_hooks'];
    }
  }

  /**
   * Initialize static members, before any class instances are created.
   */
  public static function staticInitialize() {

    // Default the displayFunction outputFunction based on context
    if (function_exists('drush_log')) {
      self::$displayFunction = 'drush_log';
    }
    else {
      self::$displayFunction = 'drupal_set_message';
    }
  }

  /**
   * Register a new migration process in the migrate_status table. This will
   * generally be used in two contexts - by the class detection code for
   * static (one instance per class) migrations, and by the module implementing
   * dynamic (parameterized class) migrations.
   *
   * @param string $class_name
   * @param string $machine_name
   * @param array $arguments
   */
  public static function registerMigration($class_name, $machine_name = NULL, array $arguments = array()) {

    // Support for legacy migration code - in later releases, the machine_name
    // should always be explicit.
    if (!$machine_name) {
      $machine_name = self::machineFromClass($class_name);
    }
    if (!preg_match('|^[a-z0-9_]+$|i', $machine_name)) {
      throw new Exception(t('!name is not a valid Migration machine name. Use only alphanumeric or underscore characters.', array(
        '!name' => $machine_name,
      )));
    }

    // We no longer have any need to store the machine_name in the arguments.
    if (isset($arguments['machine_name'])) {
      unset($arguments['machine_name']);
    }
    if (isset($arguments['group_name'])) {
      $group_name = $arguments['group_name'];
      unset($arguments['group_name']);
    }
    else {
      $group_name = 'default';
    }
    $arguments = self::encryptArguments($arguments);

    // Register the migration if it's not already there; if it is,
    // update the class and arguments in case they've changed.
    db_merge('migrate_status')
      ->key(array(
      'machine_name' => $machine_name,
    ))
      ->fields(array(
      'class_name' => $class_name,
      'group_name' => $group_name,
      'arguments' => serialize($arguments),
    ))
      ->execute();
  }

  /**
   * Deregister a migration - remove all traces of it from the database (without
   * touching any content which was created by this migration).
   *
   * @param string $machine_name
   */
  public static function deregisterMigration($machine_name) {
    $rows_deleted = db_delete('migrate_status')
      ->condition('machine_name', $machine_name)
      ->execute();

    // Make sure the group gets deleted if we were the only member.
    MigrateGroup::deleteOrphans();
  }

  /**
   * The migration machine name is stored in the arguments.
   *
   * @return string
   */
  protected function generateMachineName() {
    return $this->arguments['machine_name'];
  }

  /**
   * Given only a class name, derive a machine name (the class name with the
   * "Migration" suffix, if any, removed).
   *
   * @param $class_name
   *
   * @return string
   */
  protected static function machineFromClass($class_name) {
    if (preg_match('/Migration$/', $class_name)) {
      $machine_name = drupal_substr($class_name, 0, strlen($class_name) - strlen('Migration'));
    }
    else {
      $machine_name = $class_name;
    }
    return $machine_name;
  }

  /**
   * Return the single instance of the given migration.
   *
   * @param string $machine_name
   */

  /**
   * Return the single instance of the given migration.
   *
   * @param $machine_name
   *  The unique machine name of the migration to retrieve.
   * @param string $class_name
   *  Deprecated - no longer used, class name is retrieved from migrate_status.
   * @param array $arguments
   *  Deprecated - no longer used, arguments are retrieved from migrate_status.
   *
   * @return MigrationBase
   */
  public static function getInstance($machine_name, $class_name = NULL, array $arguments = array()) {
    $migrations =& drupal_static(__FUNCTION__, array());

    // Otherwise might miss cache hit on case difference
    $machine_name_key = drupal_strtolower($machine_name);
    if (!isset($migrations[$machine_name_key])) {

      // See if we know about this migration
      $row = db_select('migrate_status', 'ms')
        ->fields('ms', array(
        'class_name',
        'group_name',
        'arguments',
      ))
        ->condition('machine_name', $machine_name)
        ->execute()
        ->fetchObject();
      if ($row) {
        $class_name = $row->class_name;
        $arguments = unserialize($row->arguments);
        $arguments = self::decryptArguments($arguments);
        $arguments['group_name'] = $row->group_name;
      }
      else {

        // Can't find a migration with this name
        self::displayMessage(t('No migration found with machine name !machine', array(
          '!machine' => $machine_name,
        )));
        return NULL;
      }
      $arguments['machine_name'] = $machine_name;
      if (class_exists($class_name)) {
        try {
          $migrations[$machine_name_key] = new $class_name($arguments);
        } catch (Exception $e) {
          self::displayMessage(t('Migration !machine could not be constructed.', array(
            '!machine' => $machine_name,
          )));
          self::displayMessage($e
            ->getMessage());
          return NULL;
        }
      }
      else {
        self::displayMessage(t('No migration class !class found', array(
          '!class' => $class_name,
        )));
        return NULL;
      }
      if (isset($arguments['dependencies'])) {
        $migrations[$machine_name_key]
          ->setHardDependencies($arguments['dependencies']);
      }
      if (isset($arguments['soft_dependencies'])) {
        $migrations[$machine_name_key]
          ->setSoftDependencies($arguments['soft_dependencies']);
      }
    }
    return $migrations[$machine_name_key];
  }

  /**
   * @deprecated - No longer a useful distinction between "status" and "dynamic"
   *  migrations.
   */
  public static function isDynamic() {
    return FALSE;
  }

  /**
   * Default to printing messages, but derived classes are expected to save
   * messages indexed by current source ID.
   *
   * @param string $message
   *  The message to record.
   * @param int $level
   *  Optional message severity (defaults to MESSAGE_ERROR).
   */
  public function saveMessage($message, $level = MigrationBase::MESSAGE_ERROR) {
    switch ($level) {
      case MigrationBase::MESSAGE_ERROR:
        $level = 'error';
        break;
      case MigrationBase::MESSAGE_WARNING:
        $level = 'warning';
        break;
      case MigrationBase::MESSAGE_NOTICE:
        $level = 'notice';
        break;
      case MigrationBase::MESSAGE_INFORMATIONAL:
        $level = 'status';
        break;
    }
    self::displayMessage($message, $level);
  }

  /**
   * Output the given message appropriately
   * (drush_print/drupal_set_message/etc.)
   *
   * @param string $message
   *  The message to output.
   * @param int $level
   *  Optional message severity as understood by drupal_set_message and
   *   drush_log
   *  (defaults to 'error').
   */
  public static function displayMessage($message, $level = 'error') {
    call_user_func(self::$displayFunction, $message, $level);
  }

  /**
   * Custom PHP error handler.
   * TODO: Redundant with hook_watchdog?
   *
   * @param $error_level
   *   The level of the error raised.
   * @param $message
   *   The error message.
   * @param $filename
   *   The filename that the error was raised in.
   * @param $line
   *   The line number the error was raised at.
   * @param $context
   *   An array that points to the active symbol table at the point the error
   *   occurred.
   */
  public function errorHandler($error_level, $message, $filename, $line, $context) {
    if ($error_level & error_reporting()) {
      $message .= "\n" . t('File !file, line !line', array(
        '!line' => $line,
        '!file' => $filename,
      ));

      // Record notices and continue
      if ($error_level == E_NOTICE || $error_level == E_USER_NOTICE) {
        $this
          ->saveMessage($message . "(file: {$filename}, line {$line})", MigrationBase::MESSAGE_INFORMATIONAL);
      }
      elseif (!($error_level == E_STRICT || $error_level == 8192 || $error_level == 16384)) {
        throw new MigrateException($message, MigrationBase::MESSAGE_ERROR);
      }
    }
  }

  /**
   * Takes an Exception object and both saves and displays it, pulling
   * additional information on the location triggering the exception.
   *
   * @param Exception $exception
   *  Object representing the exception.
   * @param boolean $save
   *  Whether to save the message in the migration's mapping table. Set to
   *   FALSE
   *  in contexts where this doesn't make sense.
   */
  public function handleException($exception, $save = TRUE) {
    $result = _drupal_decode_exception($exception);
    $message = $result['!message'] . ' (' . $result['%file'] . ':' . $result['%line'] . ')';
    if ($save) {
      $this
        ->saveMessage($message);
    }
    self::displayMessage($message);
  }

  /**
   * Check the current status of a migration.
   *
   * @return int
   *  One of the MigrationBase::STATUS_* constants
   */
  public function getStatus() {
    if (!$this->enabled) {
      return MigrationBase::STATUS_DISABLED;
    }
    $status = db_select('migrate_status', 'ms')
      ->fields('ms', array(
      'status',
    ))
      ->condition('machine_name', $this->machineName)
      ->execute()
      ->fetchField();
    if (!isset($status)) {
      $status = MigrationBase::STATUS_IDLE;
    }
    return $status;
  }

  /**
   * Retrieve the last time an import operation completed successfully.
   *
   * @return string
   *  Date/time string, formatted... How? Default DB server format?
   */
  public function getLastImported() {
    $last_imported = db_select('migrate_log', 'ml')
      ->fields('ml', array(
      'endtime',
    ))
      ->condition('machine_name', $this->machineName)
      ->isNotNull('endtime')
      ->orderBy('endtime', 'DESC')
      ->range(0, 1)
      ->execute()
      ->fetchField();
    if ($last_imported) {
      $last_imported = date('Y-m-d H:i:s', $last_imported / 1000);
    }
    else {
      $last_imported = '';
    }
    return $last_imported;
  }

  /**
   * Fetch the current highwater mark for updated content.
   *
   * @return string
   *  The highwater mark.
   */
  public function getHighwater() {
    $highwater = db_select('migrate_status', 'ms')
      ->fields('ms', array(
      'highwater',
    ))
      ->condition('machine_name', $this->machineName)
      ->execute()
      ->fetchField();
    return $highwater;
  }

  /**
   * Save the highwater mark for this migration (but not when using an idlist).
   *
   * @param mixed $highwater
   *  Highwater mark to save
   * @param boolean $force
   *  If TRUE, save even if it's lower than the previous value.
   */
  protected function saveHighwater($highwater, $force = FALSE) {
    if (!isset($this->options['idlist'])) {
      $query = db_update('migrate_status')
        ->fields(array(
        'highwater' => $highwater,
      ))
        ->condition('machine_name', $this->machineName);
      if (!$force) {
        if (!empty($this->highwaterField['type']) && $this->highwaterField['type'] == 'int') {

          // If the highwater is an integer type, we need to force the DB server
          // to treat the varchar highwater field as an integer (otherwise it will
          // think '5' > '10').
          switch (Database::getConnection()
            ->databaseType()) {
            case 'pgsql':
              $query
                ->where('(CASE WHEN highwater=\'\' THEN 0 ELSE CAST(highwater AS INTEGER) END) < :highwater', array(
                ':highwater' => intval($highwater),
              ));
              break;
            default:

              // MySQL casts as integers as SIGNED or UNSIGNED.
              $query
                ->where('(CASE WHEN highwater=\'\' THEN 0 ELSE CAST(highwater AS SIGNED) END) < :highwater', array(
                ':highwater' => intval($highwater),
              ));
          }
        }
        else {
          $query
            ->condition('highwater', $highwater, '<');
        }
      }
      $query
        ->execute();
    }
  }

  /**
   * Retrieve the last throughput for current Migration (items / minute).
   *
   * @return integer
   */
  public function getLastThroughput() {
    $last_throughput = 0;
    $row = db_select('migrate_log', 'ml')
      ->fields('ml', array(
      'starttime',
      'endtime',
      'numprocessed',
    ))
      ->condition('machine_name', $this->machineName)
      ->condition('process_type', 1)
      ->isNotNull('endtime')
      ->orderBy('starttime', 'DESC')
      ->execute()
      ->fetchObject();
    if ($row) {
      $elapsed = ($row->endtime - $row->starttime) / 1000;
      if ($elapsed > 0) {
        $last_throughput = round($row->numprocessed / $elapsed * 60);
      }
    }
    return $last_throughput;
  }

  /**
   * Reports whether this migration process is complete. For a Migration, for
   * example, this would be whether all available source rows have been
   * processed. Other MigrationBase classes will need to return TRUE/FALSE
   * appropriately.
   */
  public abstract function isComplete();

  /**
   * Reports whether all (hard) dependencies have completed migration
   */
  protected function dependenciesComplete($rollback = FALSE) {
    if ($rollback) {
      foreach (migrate_migrations() as $migration) {
        $dependencies = $migration
          ->getHardDependencies();
        if (array_search($this->machineName, $dependencies) !== FALSE) {
          if (method_exists($migration, 'importedCount') && $migration
            ->importedCount() > 0) {
            return FALSE;
          }
        }
      }
    }
    else {
      foreach ($this->dependencies as $dependency) {
        $migration = MigrationBase::getInstance($dependency);
        if (!$migration || !$migration
          ->isComplete()) {
          return FALSE;
        }
      }
    }
    return TRUE;
  }

  /**
   * Returns an array of the migration's dependencies that are incomplete.
   */
  public function incompleteDependencies() {
    $incomplete = array();
    foreach ($this
      ->getDependencies() as $dependency) {
      $migration = MigrationBase::getInstance($dependency);
      if (!$migration || !$migration
        ->isComplete()) {
        $incomplete[] = $dependency;
      }
    }
    return $incomplete;
  }

  /**
   * Begin a process, ensuring only one process can be active
   * at once on a given migration.
   *
   * @param int $newStatus
   *  MigrationBase::STATUS_IMPORTING or MigrationBase::STATUS_ROLLING_BACK
   */
  protected function beginProcess($newStatus) {

    // So hook_watchdog() knows what migration (if any) is running
    self::$currentMigration = $this;

    // Try to make the semaphore handling atomic (depends on DB support)
    $transaction = db_transaction();

    // Save the current mail system, prior to disabling emails.
    $this
      ->saveMailSystem();

    // Prevent emails from being sent out during migrations.
    $this
      ->disableMailSystem();
    $this->starttime = microtime(TRUE);

    // Check to make sure there's no process already running for this migration
    $status = $this
      ->getStatus();
    if ($status != MigrationBase::STATUS_IDLE) {
      throw new MigrateException(t('There is already an active process on !machine_name', array(
        '!machine_name' => $this->machineName,
      )));
    }
    $this->processing = TRUE;
    $this->status = $newStatus;
    db_merge('migrate_status')
      ->key(array(
      'machine_name' => $this->machineName,
    ))
      ->fields(array(
      'class_name' => get_class($this),
      'status' => $newStatus,
    ))
      ->execute();

    // Set an error handler for imports
    if ($newStatus == MigrationBase::STATUS_IMPORTING) {
      $this->previousErrorHandler = set_error_handler(array(
        $this,
        'errorHandler',
      ));
    }

    // Save the initial history record
    if ($this->logHistory) {
      $this->logID = db_insert('migrate_log')
        ->fields(array(
        'machine_name' => $this->machineName,
        'process_type' => $newStatus,
        'starttime' => round(microtime(TRUE) * 1000),
        'initialHighwater' => $this
          ->getHighwater(),
      ))
        ->execute();
    }

    // If we're disabling any hooks, reset the static module_implements cache so
    // it is rebuilt with the specified hooks removed by our
    // hook_module_implements_alter(). By setting #write_cache to FALSE, we
    // ensure that our munged version of the hooks array does not get written
    // to the persistent cache and interfere with other Drupal processes.
    if (!empty($this->disableHooks)) {
      $implementations =& drupal_static('module_implements');
      $implementations = array();
      $implementations['#write_cache'] = FALSE;
    }
  }

  /**
   * End a rollback or import process, releasing the semaphore. Note that it
   * must be public to be callable as the shutdown function.
   */
  public function endProcess() {
    if ($this->previousErrorHandler) {
      set_error_handler($this->previousErrorHandler);
      $this->previousErrorHandler = NULL;
    }
    if ($this->processing) {
      $this->status = MigrationBase::STATUS_IDLE;

      // Restore the previous mail handler.
      $this
        ->restoreMailSystem();
      $fields = array(
        'class_name' => get_class($this),
        'status' => MigrationBase::STATUS_IDLE,
      );
      db_merge('migrate_status')
        ->key(array(
        'machine_name' => $this->machineName,
      ))
        ->fields($fields)
        ->execute();

      // Complete the log record
      if ($this->logHistory) {
        try {
          db_merge('migrate_log')
            ->key(array(
            'mlid' => $this->logID,
          ))
            ->fields(array(
            'endtime' => round(microtime(TRUE) * 1000),
            'finalhighwater' => $this
              ->getHighwater(),
            'numprocessed' => $this->total_processed,
          ))
            ->execute();
        } catch (PDOException $e) {
          Migration::displayMessage(t('Could not log operation on migration !name - possibly MigrationBase::beginProcess() was not called', array(
            '!name' => $this->machineName,
          )));
        }
      }
      $this->processing = FALSE;
    }
    self::$currentMigration = NULL;

    // If we're disabling any hooks, reset the static module_implements cache so
    // it is rebuilt again and the specified hooks removed by our
    // hook_module_implements_alter() is recollected. By setting #write_cache to
    // FALSE, we ensure that our munged version of the hooks array does not get
    // written to the persistent cache and interfere with other Drupal processes.
    if (!empty($this->disableHooks)) {
      $implementations =& drupal_static('module_implements');
      $implementations = array();
      $implementations['#write_cache'] = FALSE;
    }
  }

  /**
   * Signal that any current import or rollback process should end itself at
   * the earliest opportunity
   */
  public function stopProcess() {

    // Do not change the status of an idle migration
    db_update('migrate_status')
      ->fields(array(
      'status' => MigrationBase::STATUS_STOPPING,
    ))
      ->condition('machine_name', $this->machineName)
      ->condition('status', MigrationBase::STATUS_IDLE, '<>')
      ->execute();
  }

  /**
   * Reset the status of the migration to IDLE (to be used when the status
   * gets stuck, e.g. if a process core-dumped)
   */
  public function resetStatus() {

    // Do not change the status of an already-idle migration
    db_update('migrate_status')
      ->fields(array(
      'status' => MigrationBase::STATUS_IDLE,
    ))
      ->condition('machine_name', $this->machineName)
      ->condition('status', MigrationBase::STATUS_IDLE, '<>')
      ->execute();
  }

  /**
   * Perform an operation during the rollback phase.
   *
   * @param array $options
   *  List of options provided (usually from a drush command). Specific to
   *  the derived class.
   */
  public function processRollback(array $options = array()) {
    if ($this->enabled) {
      $return = MigrationBase::RESULT_COMPLETED;
      if (method_exists($this, 'rollback')) {
        $this->options = $options;
        if (!isset($options['force'])) {
          if (!$this
            ->dependenciesComplete(TRUE)) {
            return MigrationBase::RESULT_SKIPPED;
          }
        }
        $this
          ->beginProcess(MigrationBase::STATUS_ROLLING_BACK);
        try {
          $return = $this
            ->rollback();
        } catch (Exception $exception) {

          // If something bad happened, make sure we clear the semaphore
          $this
            ->endProcess();
          throw $exception;
        }
        $this
          ->endProcess();
      }
    }
    else {
      $return = MigrationBase::RESULT_DISABLED;
    }
    return $return;
  }

  /**
   * Perform an operation during the import phase
   *
   * @param array $options
   *  List of options provided (usually from a drush command). Specific to
   *  the derived class.
   */
  public function processImport(array $options = array()) {
    if ($this->enabled) {
      $return = MigrationBase::RESULT_COMPLETED;
      if (method_exists($this, 'import')) {
        $this->options = $options;
        if (!isset($options['force']) || !$options['force']) {
          if (!$this
            ->dependenciesComplete()) {
            return MigrationBase::RESULT_SKIPPED;
          }
        }
        $this
          ->beginProcess(MigrationBase::STATUS_IMPORTING);
        try {
          $return = $this
            ->import();
        } catch (Exception $exception) {

          // If something bad happened, make sure we clear the semaphore
          $this
            ->endProcess();
          throw $exception;
        }
        if ($return == MigrationBase::RESULT_COMPLETED && isset($this->total_successes)) {
          $time = microtime(TRUE) - $this->starttime;
          if ($time > 0) {
            $overallThroughput = round(60 * $this->total_successes / $time);
          }
          else {
            $overallThroughput = 9999;
          }
        }
        else {
          $overallThroughput = 0;
        }
        $this
          ->endProcess($overallThroughput);
      }
    }
    else {
      $return = MigrationBase::RESULT_DISABLED;
    }
    return $return;
  }

  /**
   * Set the PHP time limit. This method may be called from batch callbacks
   * before calling the processImport method.
   */
  public function setBatchTimeLimit() {
    drupal_set_time_limit($this->batchTimeLimit);
  }

  /**
   * A derived migration class does the actual rollback or import work in these
   * methods - we cannot declare them abstract because some classes may define
   * only one.
   *
   * abstract protected function rollback();
   * abstract protected function import();
   */

  /**
   * Test whether we've exceeded the desired memory threshold. If so, output a
   * message.
   *
   * @return boolean
   *  TRUE if the threshold is exceeded, FALSE if not.
   */
  protected function memoryExceeded() {
    $usage = memory_get_usage();
    $pct_memory = $usage / $this->memoryLimit;
    if ($pct_memory > $this->memoryThreshold) {
      self::displayMessage(t('Memory usage is !usage (!pct% of limit !limit), resetting statics', array(
        '!pct' => round($pct_memory * 100),
        '!usage' => format_size($usage),
        '!limit' => format_size($this->memoryLimit),
      )), 'warning');

      // First, try resetting Drupal's static storage - this frequently releases
      // plenty of memory to continue
      drupal_static_reset();
      $usage = memory_get_usage();
      $pct_memory = $usage / $this->memoryLimit;

      // Use a lower threshold - we don't want to be in a situation where we keep
      // coming back here and trimming a tiny amount
      if ($pct_memory > 0.9 * $this->memoryThreshold) {
        self::displayMessage(t('Memory usage is now !usage (!pct% of limit !limit), not enough reclaimed, starting new batch', array(
          '!pct' => round($pct_memory * 100),
          '!usage' => format_size($usage),
          '!limit' => format_size($this->memoryLimit),
        )), 'warning');
        return TRUE;
      }
      else {
        self::displayMessage(t('Memory usage is now !usage (!pct% of limit !limit), reclaimed enough, continuing', array(
          '!pct' => round($pct_memory * 100),
          '!usage' => format_size($usage),
          '!limit' => format_size($this->memoryLimit),
        )), 'warning');
        return FALSE;
      }
    }
    else {
      return FALSE;
    }
  }

  /**
   * Test whether we're approaching the PHP time limit.
   *
   * @return boolean
   *  TRUE if the threshold is exceeded, FALSE if not.
   */
  protected function timeExceeded() {
    if ($this->timeLimit == 0) {
      return FALSE;
    }
    $time_elapsed = time() - REQUEST_TIME;
    $pct_time = $time_elapsed / $this->timeLimit;
    if ($pct_time > $this->timeThreshold) {
      return TRUE;
    }
    else {
      return FALSE;
    }
  }

  /**
   * Test whether we've exceeded the designated time limit.
   *
   * @return boolean
   *  TRUE if the threshold is exceeded, FALSE if not.
   */
  protected function timeOptionExceeded() {
    if (!($timelimit = $this
      ->getTimeLimit())) {
      return FALSE;
    }
    $time_elapsed = time() - REQUEST_TIME;
    if ($time_elapsed >= $timelimit) {
      return TRUE;
    }
    else {
      return FALSE;
    }
  }

  /**
   * Encrypt an incoming value. Detects for existence of the Drupal 'Encrypt'
   *  module.
   *
   * @param string $value
   *
   * @return string The encrypted value.
   */
  public static function encrypt($value) {
    if (module_exists('encrypt')) {
      $value = encrypt($value);
    }
    else {
      if (self::$showEncryptionWarning) {
        MigrationBase::displayMessage(t('Encryption of secure migration information is not supported. Ensure the <a href="@encrypt">Encrypt module</a> is installed for this functionality.', array(
          '@encrypt' => 'http://drupal.org/project/encrypt',
        )), 'warning');
        self::$showEncryptionWarning = FALSE;
      }
    }
    return $value;
  }

  /**
   * Decrypt an incoming value.
   *
   * @param string $value
   *
   * @return string The encrypted value
   */
  public static function decrypt($value) {
    if (module_exists('encrypt')) {
      $value = decrypt($value);
    }
    else {
      if (self::$showEncryptionWarning) {
        MigrationBase::displayMessage(t('Encryption of secure migration information is not supported. Ensure the <a href="@encrypt">Encrypt module</a> is installed for this functionality.', array(
          '@encrypt' => 'http://drupal.org/project/encrypt',
        )), 'warning');
        self::$showEncryptionWarning = FALSE;
      }
    }
    return $value;
  }

  /**
   * Make sure any arguments we want to be encrypted get encrypted.
   *
   * @param array $arguments
   *
   * @return array
   */
  public static function encryptArguments(array $arguments) {
    if (isset($arguments['encrypted_arguments'])) {
      foreach ($arguments['encrypted_arguments'] as $argument_name) {
        if (isset($arguments[$argument_name])) {
          $arguments[$argument_name] = self::encrypt(serialize($arguments[$argument_name]));
        }
      }
    }
    return $arguments;
  }

  /**
   * Make sure any arguments we want to be decrypted get decrypted.
   *
   * @param array $arguments
   *
   * @return array
   */
  public static function decryptArguments(array $arguments) {
    if (isset($arguments['encrypted_arguments'])) {
      foreach ($arguments['encrypted_arguments'] as $argument_name) {
        if (isset($arguments[$argument_name])) {
          $decrypted_string = self::decrypt($arguments[$argument_name]);

          // A decryption failure will return FALSE and issue a notice. We need
          // to distinguish a failure from a serialized FALSE.
          $unserialized_value = @unserialize($decrypted_string);
          if ($unserialized_value === FALSE && $decrypted_string != serialize(FALSE)) {
            self::displayMessage(t('Failed to decrypt argument %argument_name', array(
              '%argument_name' => $argument_name,
            )));
            unset($arguments[$argument_name]);
          }
          else {
            $arguments[$argument_name] = $unserialized_value;
          }
        }
      }
    }
    return $arguments;
  }

  /**
   * Convert an incoming string (which may be a UNIX timestamp, or an
   * arbitrarily-formatted date/time string) to a UNIX timestamp.
   *
   * @param string $value
   *   The time string to convert.
   * @param string $timezone
   *   Optional timezone for the time string. NULL to leave the timezone unset.
   *
   * @return string
   *   The UNIX timestamp.
   */
  public static function timestamp($value, $timezone = NULL) {

    // Does it look like it's already a timestamp? Just return it
    if (is_numeric($value)) {
      return $value;
    }

    // Default empty values to now
    if (empty($value)) {
      return time();
    }
    if (isset($timezone)) {
      $timezone = new DateTimeZone($timezone);
    }
    $date = new DateTime($value, $timezone);
    $time = $date
      ->format('U');
    if ($time == FALSE) {

      // Handles form YYYY-MM-DD HH:MM:SS.garbage
      if (drupal_strlen($value) > 19) {
        $time = strtotime(drupal_substr($value, 0, 19));
      }
    }
    return $time;
  }

  /**
   * Saves the current mail system, or set a system default if there is none.
   */
  public function saveMailSystem() {
    global $conf;
    $this->mailSystem = empty($conf['mail_system']) ? NULL : $conf['mail_system'];
  }

  /**
   * Disables mail system to prevent emails from being sent during migrations.
   */
  public function disableMailSystem() {
    global $conf;
    if (!empty($conf['mail_system'])) {
      foreach ($conf['mail_system'] as $system => $class) {
        $conf['mail_system'][$system] = 'MigrateMailIgnore';
      }
    }
    else {
      $conf['mail_system'] = array(
        'default-system' => 'MigrateMailIgnore',
      );
    }
  }

  /**
   * Restores the original saved mail system for migrations that require it.
   */
  public function restoreMailSystem() {
    global $conf;
    $conf['mail_system'] = $this->mailSystem;
  }

}

Members

Name	Modifiers	Type	Description	Overrides
MigrationBase::$arguments	protected	property	Arguments configuring a migration.
MigrationBase::$batchTimeLimit	protected	property	A time limit in seconds appropriate to be used in a batch import. Defaults to 240.
MigrationBase::$currentMigration	protected static	property	Track the migration currently running, so handlers can easily determine it without having to pass a Migration object everywhere.
MigrationBase::$dependencies	protected	property	List of other Migration classes which should be imported before this one. E.g., a comment migration class would typically have node and user migrations as dependencies.
MigrationBase::$description	protected	property	Detailed information describing the migration.
MigrationBase::$disableHooks	protected	property	Any module hooks which should be disabled during migration processes.
MigrationBase::$displayFunction	protected static	property	Name of a function for displaying feedback. It must take the message to display as its first argument, and a (string) message type as its second argument (see drush_log()).
MigrationBase::$emptyArgumentsWarning	protected static	property
MigrationBase::$enabled	protected	property	Disabling a migration prevents it from running with --all, or individually without --force
MigrationBase::$group	protected	property	A migration group object, used to collect related migrations.
MigrationBase::$groupArgumentWarning	protected static	property	Have we already warned about obsolete constructor argumentss on this request?
MigrationBase::$issuePattern	protected	property	If provided, an URL for an issue tracking system containing :id where the issue number will go (e.g., 'http://example.com/project/ticket/:id').
MigrationBase::$logHistory	protected	property	Whether to maintain a history of migration processes in migrate_log
MigrationBase::$logID	protected	property	Primary key of the current history record (inserted at the beginning of a process, to be updated at the end)
MigrationBase::$machineName	protected	property	The machine name of this Migration object, derived by removing the 'Migration' suffix from the class name. Used to construct default map/message table names, displayed in drush migrate-status, key to migrate_status table...
MigrationBase::$mailSystem	protected	property	An array to track 'mail_system' variable if disabled.
MigrationBase::$memoryLimit	protected	property	The PHP memory_limit expressed in bytes.
MigrationBase::$memoryThreshold	protected	property	The fraction of the memory limit at which an operation will be interrupted. Can be overridden by a Migration subclass if one would like to push the envelope. Defaults to 85%.
MigrationBase::$options	protected	property	Save options passed to current operation
MigrationBase::$previousErrorHandler	protected	property	If we set an error handler (during import), remember the previous one so it can be restored.
MigrationBase::$processing	protected	property	Indicates that we are processing a rollback or import - used to avoid excess writes in endProcess()
MigrationBase::$showEncryptionWarning	protected static	property	Track whether or not we've already displayed an encryption warning
MigrationBase::$starttime	protected	property	When the current operation started.
MigrationBase::$status	protected	property	Are we importing, rolling back, or doing nothing?
MigrationBase::$team	protected	property	MigrateTeamMember objects representing people involved with this migration.
MigrationBase::$timeLimit	protected	property	The PHP max_execution_time.
MigrationBase::$timeThreshold	protected	property	The fraction of the time limit at which an operation will be interrupted. Can be overridden by a Migration subclass if one would like to push the envelope. Defaults to 90%.
MigrationBase::$total_processed	protected	property	Number of "items" processed in the current migration process (whatever that means for the type of process)
MigrationBase::addArguments	public	function
MigrationBase::addHardDependencies	public	function
MigrationBase::addSoftDependencies	public	function
MigrationBase::beginProcess	protected	function	Begin a process, ensuring only one process can be active at once on a given migration.	1
MigrationBase::currentMigration	public static	function
MigrationBase::decrypt	public static	function	Decrypt an incoming value.
MigrationBase::decryptArguments	public static	function	Make sure any arguments we want to be decrypted get decrypted.
MigrationBase::dependenciesComplete	protected	function	Reports whether all (hard) dependencies have completed migration
MigrationBase::deregisterMigration	public static	function	Deregister a migration - remove all traces of it from the database (without touching any content which was created by this migration).	1
MigrationBase::disableMailSystem	public	function	Disables mail system to prevent emails from being sent during migrations.
MigrationBase::displayMessage	public static	function	Output the given message appropriately (drush_print/drupal_set_message/etc.)
MigrationBase::encrypt	public static	function	Encrypt an incoming value. Detects for existence of the Drupal 'Encrypt' module.
MigrationBase::encryptArguments	public static	function	Make sure any arguments we want to be encrypted get encrypted.
MigrationBase::endProcess	public	function	End a rollback or import process, releasing the semaphore. Note that it must be public to be callable as the shutdown function.	1
MigrationBase::errorHandler	public	function	Custom PHP error handler. TODO: Redundant with hook_watchdog?
MigrationBase::generateMachineName	protected	function	The migration machine name is stored in the arguments.	1
MigrationBase::getArguments	public	function
MigrationBase::getDependencies	public	function
MigrationBase::getDescription	public	function
MigrationBase::getDisableHooks	public	function
MigrationBase::getEnabled	public	function
MigrationBase::getGroup	public	function
MigrationBase::getHardDependencies	public	function
MigrationBase::getHighwater	public	function	Fetch the current highwater mark for updated content.
MigrationBase::getInstance	public static	function	Return the single instance of the given migration.
MigrationBase::getIssuePattern	public	function
MigrationBase::getItemLimit	public	function
MigrationBase::getLastImported	public	function	Retrieve the last time an import operation completed successfully.
MigrationBase::getLastThroughput	public	function	Retrieve the last throughput for current Migration (items / minute).
MigrationBase::getMachineName	public	function
MigrationBase::getMessageLevelName	public	function	Get human readable name for a message constant.
MigrationBase::getOption	public	function
MigrationBase::getSoftDependencies	public	function
MigrationBase::getStatus	public	function	Check the current status of a migration.
MigrationBase::getTeam	public	function
MigrationBase::getTimeLimit	public	function
MigrationBase::handleException	public	function	Takes an Exception object and both saves and displays it, pulling additional information on the location triggering the exception.
MigrationBase::incompleteDependencies	public	function	Returns an array of the migration's dependencies that are incomplete.
MigrationBase::isComplete	abstract public	function	Reports whether this migration process is complete. For a Migration, for example, this would be whether all available source rows have been processed. Other MigrationBase classes will need to return TRUE/FALSE appropriately.	3
MigrationBase::isDynamic Deprecated	public static	function		1
MigrationBase::machineFromClass	protected static	function	Given only a class name, derive a machine name (the class name with the "Migration" suffix, if any, removed).
MigrationBase::memoryExceeded	protected	function	Test whether we've exceeded the desired memory threshold. If so, output a message.
MigrationBase::MESSAGE_ERROR		constant	Message types to be passed to saveMessage() and saved in message tables. MESSAGE_INFORMATIONAL represents a condition that did not prevent the operation from succeeding - all others represent different severities of conditions resulting in a source…
MigrationBase::MESSAGE_INFORMATIONAL		constant
MigrationBase::MESSAGE_NOTICE		constant
MigrationBase::MESSAGE_WARNING		constant
MigrationBase::processImport	public	function	Perform an operation during the import phase
MigrationBase::processRollback	public	function	Perform an operation during the rollback phase.
MigrationBase::registerMigration	public static	function	Register a new migration process in the migrate_status table. This will generally be used in two contexts - by the class detection code for static (one instance per class) migrations, and by the module implementing dynamic (parameterized class)…	1
MigrationBase::resetStatus	public	function	Reset the status of the migration to IDLE (to be used when the status gets stuck, e.g. if a process core-dumped)
MigrationBase::restoreMailSystem	public	function	Restores the original saved mail system for migrations that require it.
MigrationBase::RESULT_COMPLETED		constant	Codes representing the result of a rollback or import process.
MigrationBase::RESULT_DISABLED		constant
MigrationBase::RESULT_FAILED		constant
MigrationBase::RESULT_INCOMPLETE		constant
MigrationBase::RESULT_SKIPPED		constant
MigrationBase::RESULT_STOPPED		constant
MigrationBase::saveHighwater	protected	function	Save the highwater mark for this migration (but not when using an idlist).
MigrationBase::saveMailSystem	public	function	Saves the current mail system, or set a system default if there is none.
MigrationBase::saveMessage	public	function	Default to printing messages, but derived classes are expected to save messages indexed by current source ID.	1
MigrationBase::setArguments	public	function
MigrationBase::setBatchTimeLimit	public	function	Set the PHP time limit. This method may be called from batch callbacks before calling the processImport method.
MigrationBase::setDescription	public	function
MigrationBase::setDisplayFunction	public static	function
MigrationBase::setEnabled	public	function
MigrationBase::setHardDependencies	public	function
MigrationBase::setIssuePattern	public	function
MigrationBase::setSoftDependencies	public	function
MigrationBase::setTeam	public	function
MigrationBase::staticInitialize	public static	function	Initialize static members, before any class instances are created.
MigrationBase::STATUS_DISABLED		constant
MigrationBase::STATUS_IDLE		constant	Codes representing the current status of a migration, and stored in the migrate_status table.
MigrationBase::STATUS_IMPORTING		constant
MigrationBase::STATUS_ROLLING_BACK		constant
MigrationBase::STATUS_STOPPING		constant
MigrationBase::stopProcess	public	function	Signal that any current import or rollback process should end itself at the earliest opportunity
MigrationBase::timeExceeded	protected	function	Test whether we're approaching the PHP time limit.
MigrationBase::timeOptionExceeded	protected	function	Test whether we've exceeded the designated time limit.
MigrationBase::timestamp	public static	function	Convert an incoming string (which may be a UNIX timestamp, or an arbitrarily-formatted date/time string) to a UNIX timestamp.
MigrationBase::__construct	public	function	Construction of a MigrationBase instance.	3

You are here

abstract class MigrationBase in Migrate 7.2

Hierarchy

File

Members

API Navigation