Pipes: an outline

pipes.info
==========
; $Id$
name = Pipes
description = Allows for creation of pipes through the ui.
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.'),
      ),
      'parent' => array(
        'type' => 'int',
        'unsigned' => TRUE,
        'default' => 0,
        'description' => t("The {pipes_step_data}.sid of the step's parent (where it's output goes). NULL indicates that the step's output goes to root. Steps without output are not listed."),
      ),
    ),
    'indexes' => array(
      'parent' => array('parent'),
    ),
    'unique keys' => array('sid_parent' => array('sid', 'parent')),
  );
 
  $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$
 
// An array of the modules pipes core has .inc files for.
define('PIPES_CORE_SUPPORTED_MODULES', array());
 
/**
 * Lists all steps through invoking hook_pipes_steps.
 * @param $class
 *   If set, the class of a specific step to return information about.
 * @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_steps_list($class = NULL, $refresh = FALSE) {
  static $list;
  if ($refresh || !isset($list)) {
    $list = module_invoke_all('pipes_steps');
  }
  if (!is_null($class)) {
    if (isset($list[$class])) }
      return $list[$class];
    }
    return FALSE;
  }
  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_steps_load($class) {
  return pipes_steps_list($class);
}
 
/**
 * List all pipes.
 * @param $pid
 *   The pid of a a specific pipe to load. 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, $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 (!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_object($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['parent'] = $pipe->psh.parent;
        $data['settings'] = $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) {
    $value = call_user_func_array($info['class'], array_merge($values[$sid], $info['settings']));
    if (isset($info['step_data']['escapes'])) {
      foreach ($info['step_data']['escapes'] as $check => $action) {
        // Check for strict equality.
        if ($value === $check) {
          // A match has been found. Execute functions / return stuff.
          foreach ($action as $type => $function_or_return) {
            if ($type == 'function') {
              // We do not use the return value. If changes to $value are to be made, $value should be asked for by reference.
              $function_or_return($value);
            }
            if ($type == 'return') {
              return $function_or_return;
            }
          }
        }
      }
    }
    if (!is_null($value) && !is_null($info['parent'])) {
      $values[$info['parent']][] = $value;
    }
  }
 
  // Return final value, if we reach here.
  return $value;
}
 
/**
 * 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 parent, thus the parent is dependent on it.
    $dependencies[$step->parent][] = $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[$sid]);
        // We just made a change, so we are being productive.
        $being_productive = TRUE;
        // Cycle through dependencies and remove this one.
        foreach ($dependencies as $parent => $children) {
          if (($key = array_search($sid, $children, TRUE)) !== FALSE) {
            unset($dependencies[$parent][$key]);
          }
        }
      }
    }
  }
  if (!$being_productive) {
    // Circular dependency.
    return FALSE;
  }
  return $new_steps;
}
DrupalBin

Recent Posts

Tags

User login
