Pipes

pipes.info
==========
; $Id$
name = Pipes
description = Provides a framework for creation and execution of Drupal pipes.
core = "6.x"
 
pipes.install
=============
<?php
// $Id$
 
/**
 * Implementation of hook_install().
 */
function pipes_install() {
  drupal_install_schema('pipes');
}
 
/**
 * Implementation of hook_uninstall().
 */
function pipes_uninstall() {
  drupal_uninstall_schema('pipes');
}
 
/**
 * Implementation of hook_schema().
 */
function pipes_schema() {
  $schema['pipes_step_data'] = array(
    'description' => t('Stores step information.'),
    'fields' => array(
      'sid' => array(
        'type' => 'serial',
        'unsigned' => TRUE,
        'not null' => TRUE,
        'description' => t('Primary Key: Unique term ID.'),
      ),
      'pid' => array(
        'type' => 'int',
        'unsigned' => TRUE,
        'not null' => TRUE,
        'default' => 0,
        'description' => t('The {pipes}.pid of the pipe to which the step is assigned.'),
      ),
      'class' => array(
        'type' => 'varchar',
        'length' => 255,
        'not null' => TRUE,
        'default' => '',
        'description' => t('The step class (the function name).'),
      ),
      'settings' => array(
        'type' => 'text',
        'size' => 'big',
        'not null' => TRUE,
        'description' => t('Serialized array of any settings that have been set for this step (not including hierarchical settings).'),
      ),
      'weight' => array(
        'type' => 'int',
        'not null' => TRUE,
        'default' => 0,
        'size' => 'tiny',
        'description' => t('The saved weight of this step in relation to other steps. Lighter steps are fired first.'),
      ),
    ),
    'primary key' => array('sid'),
    'indexes' => array('pid' => array('pid')),
  );
 
  $schema['pipes_step_hierarchy'] = array(
    'description' => t('Stores the hierarchical relationship between steps.'),
    'fields' => array(
      'sid' => array(
        'type' => 'int',
        'unsigned' => TRUE,
        'not null' => TRUE,
        'default' => 0,
        'description' => t('The {pipes_step_data}.sid of the step.'),
      ),
      'destination' => array(
        'type' => 'int',
        'unsigned' => TRUE,
        'default' => 0,
        'description' => t("The {pipes_step_data}.sid of the step's destination (where it's output goes). NULL indicates that the step's output goes to root. Steps without output are not listed."),
      ),
      'param_id' => array(
        'type' => 'int',
        'not null' => TRUE,
        'default' => 0,
        'size' => 'tiny',
        'description' => t('The number of the parameter that this step\'s output needs to go to.'),
      ),
    ),
    'indexes' => array(
      'destination' => array('destination'),
    ),
    'unique keys' => array('sid_destination_param_id' => array('sid', 'destination', 'param_id')),
  );
 
  $schema['pipes'] = array(
    'description' => t('Stores pipe information.'),
    'fields' => array(
      'pid' => array(
        'type' => 'serial',
        'unsigned' => TRUE,
        'not null' => TRUE,
        'description' => t('Primary Key: Unique pipe ID.'),
      ),
      'name' => array(
        'type' => 'varchar',
        'length' => 255,
        'not null' => TRUE,
        'default' => '',
        'description' => t('Name of the pipe.'),
      ),
      'module' => array(
        'type' => 'varchar',
        'length' => 255,
        'not null' => TRUE,
        'default' => '',
        'description' => t('The module which created and manages the pipe.'),
      ),
    ),
    'primary key' => array('pid'),
  );
 
  return $schema;
}
 
pipes.module
=============
<?php
// $Id$
 
/**
 * Lists all steps through invoking hook_pipes_steps.
 * @param $refresh
 *   If TRUE, will re-invoke the hook even if already stored in the static variable.
 * @return
 *   Either an array of steps or a specific step in the array, depending on whether $class was set or not.
 */
function pipes_list_steps($refresh = FALSE) {
  static $list;
  if ($refresh || !isset($list)) {
    $list = module_invoke_all('pipes_steps');
    foreach ($list as $class => $info) {
      // Allow modules to alter pipe steps.
      drupal_alter('pipes_step', $info, $class);
    }
  }
  return $list;
}
 
/**
 * Load a specific step through invoking hook_pipes_steps.
 * @param $class
 *   The class of the step to load.
 * @return
 *   The specified step, or FALSE on failure.
 */
function pipes_step_load($class) {
  $steps = pipes_list_steps();
  if (!isset($steps[$class])) {
    return FALSE;
  }
  return $steps[$class];
}
 
/**
 * List all pipes.
 * @param $pid
 *   The pid of a a specific pipe to load. If not set, all will be loaded.
 * @param $module
 *   The module whose pipe(s) should be loaded. If not set, all will be loaded.
 * @param $refresh
 *   If TRUE, will re-load the data even if already stored in the static variable.
 * @return
 *   Either an array of pipes or a specific pipe in the array, depending on whether $pid was set or not.
 */
function pipes_list($pid = NULL, $module = NULL, $refresh = FALSE) {
  static $list;
  if (!is_null($pid)) {
    if (isset($list[$pid])) {
      return $list[$pid];
    }
  }
  $regenerate = ($refresh || (is_null($pid) ? !isset($list) : (!isset($list[$pid]))));
  if ($regenerate) {
    $list = array();
    $query = 'SELECT p.* psd.* psh.* FROM {pipes} p INNER JOIN {pipes_step_data} psd ON psd.pid = p.pid LEFT JOIN {pipes_step_hierarchy} psh ON psh.sid = p.sid';
    $where = array();
    $vars = array();
    if (!is_null($pid)) {
      $where[] = 'p.pid = %d';
      $vars[] = $pid;
    }
    if (!is_null($module)) {
      $where[] = 'p.module = %s';
      $vars[] = $module;
    }
    if (!empty($where)) {
      $query .= ' WHERE '. implode(' AND ', $where) .' ORDER BY psd.weight';
      $result = call_user_func_array('db_query', array_merge(array($query), $vars));
    }
    else {
      $result = db_query($query .' ORDER BY psd.weight');
    }
    while ($pipe = db_fetch_array($result)) {
      foreach (array('name', 'module') as $element) {
        if (!isset($list[$pipe['p.pid']][$element])) {
          $list[$pipe['p.pid']][$element] = $pipe['p.'. $element];
        }
        // Multiple steps per pipe.
        $data = array();
        $data['class'] = $pipe['psd.class'];
        $data['step_data'] = pipes_step_load($pipe['psd.class']);
        $data['destination'][] = array('step' => $pipe['psh.destination'], 'param' => $pipe['psh.param_id']);
        $data['settings'] = unserialize($pipe['psd.settings']);
        $list[$pipe['p.pid']][$pipe['psd.sid']] = $data;
      }
    }
  }
  if (!is_null($pid)) {
    if (isset($list[$pid])) {
      return $list[$pid];
    }
    return FALSE;
  }
  return $list;
}
 
/**
 * Fetch a pipe based on a given pipe id.
 * @param $pid
 *   The pid of the pipe to load.
 * @return
 *   The specified step, or FALSE on failure.
 */
function pipes_load($pid) {
  return pipes_list($pid);
}
 
/**
 * Main pipe callback. Executes an array of steps.
 * @param $steps
 *   Either an array of steps, or the id of a pipe to load.
 * @param $reorder
 *   Defaults to FALSE. If set to TRUE, will cause the steps to be re-ordered.
 * @return
 *   Return value of the pipe. Unknown.
 */
function pipes_execute($steps, $reorder = FALSE) {
  if (is_numeric($steps)) {
    // It's a pid, so load the pipe.
    $steps = pipes_load($steps);
  }
  // Convert to an array of necessary.
  $steps = array($steps);
  // Reorder the pipes if called for.
  if ($reorder) {
    $steps = pipes_order_steps($steps);
  }
  // Preserve a copy of the steps array before trimming.
  $pipe = $steps;
  // Trim irrelevant data from the steps array in order to protect it during foreach() loop.
  pipes_trim_steps($steps);
 
  $values = array();
 
  // Cycle through each of the steps in the pipe.
  foreach ($steps as $sid => $info) {
    foreach ($info['step_data']['defaults'] as $key => $var) {
      if (is_int($var)) {
        $info['step_data']['defaults'][$key] = arg($var);
      }
    }
    $value = call_user_func_array($info['class'], ($values[$sid] + $info['settings'] + $info['step_data']['defaults']));
    if (isset($info['step_data']['decision'])) {
      $decision = pipes_evaluate_condition($info['step_data']['decision'], $value, $steps, $values);
      if ($decision !== NULL) {
        $value = $decision;
      }
    }
    if (isset($info['step_data']['escape'])) {
      $escape = pipes_evaluate_condition($info['step_data']['escape'], $value, $steps, $values);
      if ($escape !== NULL) {
        return $escape;
      }
    }
    if (!is_null($info['destination'])) {
      foreach ($info['destination'] as $destination_data) {
        $values[$destination_data['step']][$destination_data['param']] = $value;
      }
    }
  }
 
  // Return final value, if we reach here.
  return $value;
}
 
/**
 * Helper function for pipes_execute().
 * @param $condition
 *   The condition to be evaluated.
 * @param $value
 *   The value, passed in by reference.
 * @param $steps
 *   The steps, passed in by reference.
 * @param $values
 *   The values, passed in by reference.
 * @return
 *   The produced data.
 */
function pipes_evaluate_condition($condition, &$value, &$steps, &$destination_data) {
  foreach ($condition as $function => $transition) {
    $result = FALSE;
    switch ($function) {
      case 'empty':
        $result = empty($value);
        break;
      case 'isset':
        $result = isset($value);
        break;
      default:
        $result = $function($value);
    }
    if ($result) {
      preg_match('/(\w+)(\(\))$/', $transition, $matches);
      if ($matches[2] == '()') {
        // It's a function.
        $function = $matches[1];
        return $function($value, $steps, $destination_data);
      }
      else {
        return $transition;
      }
    }
  }
}
 
/**
 * Removes non-step data from a pipe array.
 * @param &$steps
 *   An array of steps that need non-step data trimmed. Passed in by reference.
 */
function pipes_trim_steps(&$steps) {
  unset($steps['name']);
  unset($steps['module']);
}
 
/**
 * Orders steps based on which step needs which information.
 * @param $steps
 *   An array of steps to order.
 */
function pipes_order_steps($steps) {
  $dependencies = array();
  $new_steps = array();
  foreach ($steps as $sid => $step) {
    // The step gives data to its destination, thus the destination is dependent on it.
    $dependencies[$step->destination][] = $sid;
  }
  $being_productive = TRUE;
  while ($being_productive && !empty($steps)) {
    $being_productive = FALSE;
    foreach ($steps as $sid => $step) {
      if (empty($dependencies[$sid])) {
        // It has no unhandled dependencies, so might as well put it in.
        $new_steps[$sid] = $step;
        // We no longer have to worry about it.
        unset($steps[
DrupalBin

Recent Posts

Tags

User login
