modules/subscribe/subscribe.module

<?php
// $Id: subscribe.module,v 1.26 2006/11/11 05:05:31 jvandyk Exp $

/**
 * @file
 * Allow site to receive content via XML-RPC from Drupal sites running the publish module.
 */

/********************************************************************
 * Drupal Hooks :: General Overview
 ********************************************************************/

/**
 * Implementation of hook_help().
 */
function subscribe_help($section) {
  switch ($section) {
    case 'admin/modules#description':
      return t('Allow site receive content from Drupal sites running the publish module.');
    case strstr($section, 'admin/subscribe/pull'):
      return t('<p>Manually request updated content from the channel. Note that there may be a long pause after you initiate content tranfer.</p>');
  }
}

/**
 * Implementation of hook_menu().
 */
function subscribe_menu($may_cache) {
  $items = array();
  $access = user_access('manage content subscriptions');

  if ($may_cache) {
    $items[] = array('path' => 'admin/subscribe', 'title' => t('subscribe'),
      'callback' => 'subscribe_overview',
      'access' => $access);
    $items[] = array('path' => 'admin/subscribe/overview', 'title' => t('list'),
      'type' => MENU_DEFAULT_LOCAL_TASK,
      'weight' => -10);
    $items[] = array('path' => 'admin/subscribe/sub/wizard', 'title' => t('subscribe to channel'),
      'callback' => 'subscribe_url_form',
      'access' => $access,
      'type' => MENU_LOCAL_TASK);
    $items[] = array('path' => 'admin/subscribe/sub/vocabmap', 'title' => t('Channel options'),
      'callback' => 'subscribe_vocabmap_form',
      'access' => $access,
      'type' => MENU_CALLBACK);
    $items[] = array('path' => 'admin/subscribe/delete', 'title' => t('Delete subscription'),
      'callback' => 'subscribe_delete_form',
      'access' => $access,
      'type' => MENU_CALLBACK);
    $items[] = array('path' => 'admin/subscribe/pull', 'title' => t('update content'),
      'callback' => 'subscribe_pull_form',
      'access' => $access,
      'type' => MENU_CALLBACK);
   }

  return $items;
}

/**
* Implementation of hook_perm().
*/
function subscribe_perm() {
  return array('manage content subscriptions');
}

/********************************************************************
 * Drupal Hooks :: Core
 ********************************************************************/

/**
 * Implementation of hook_xmlrpc().
 *
 * This is the server side of the xml-rpc request.
 * Registering xml-rpc methods to callback functions.
 */
function subscribe_xmlrpc() {
  return array('drupal.subscribe.receive' => 'subscribe_xmls_receive');
}

/**
 * Implementation of hook_nodeapi().
 */
function subscribe_nodeapi(&$node, $op, $teaser = NULL, $page = NULL) {
  switch ($op) {
    case 'load':
      if ($remote = subscribe_node_load($node->nid)) {
        foreach ($remote as $key => $value) {
          $node->$key = $value;
        }
      }
      break;

    case 'view':
      if (isset($node->remote_nid)) {
        $node = theme_external_node($node, 0, 1);
      }
      break;

    case 'delete':
      if (isset($node->remote_nid) && isset($node->sid)) {
        return db_query('DELETE FROM {subscribe_node} WHERE remote_nid = %d AND sid = %d', $node->remote_nid, $node->sid);
      }
      break;
  }
}

/**
 * Return the remote metadata for a node that's been imported.
 *
 * @param $nid
 *   the local nid of node
 */
function subscribe_node_load($nid) {
  return db_fetch_object(db_query('SELECT sn.remote_nid, sn.remote_uid, sn.remote_author, sn.sid, ss.base_url as remote_site_url, ss.name as remote_site FROM {subscribe_node} sn INNER JOIN {subscribe_subscriptions} ss ON ss.sid = sn.sid WHERE sn.nid = %d', $nid));

}

/**
 * Theme function to control the look and feel of an external node.
 */
function theme_external_node($node, $teaser = 0, $page = 0) {
  $node->title = t('%node-title (from %remote-url)', array('%node-title' => $node->title, '%remote-url' => $node->remote_site_url));

  $remote_url = $node->remote_site_url. '/';
  $item[] = t('<a href="%remote-url">Visit this page</a>', array('%remote-url' => $remote_url. "node/$node->remote_nid"));
  $item[] = t('Authored by: <a href="%remote-author-url">%remote-author</a>', array('%remote-author' => $node->remote_author, '%remote-author-url' => $remote_url. "user/$node->remote_uid"));

  $referring_info = theme('item_list', $item);

  $node->body .= $referring_info;
  $node->teaser .= $referring_info;

  return $node;
}

/********************************************************************
 * Module Functions :: Subscribing
 ********************************************************************/

 /**
 *
 * Display all subscriptions.
 */
function subscribe_overview() {
  $output = '<h3>'. t('Subscribed to:') .'</h3>';
  $header = array(t('Name'),t('URI'), t('Last Received'), t('Operations'));
  $result = db_query('SELECT * FROM {subscribe_subscriptions}');
  $row = array();
  while ($sub = db_fetch_object($result)) {
    $row[] = array(
      $sub->name,
      $sub->url,
      $sub->last_update ? format_date($sub->last_update) : t('never'),
      '<table><tr><td>' .
      // l(t('edit'), "admin/subscribe/sub/$sub->sid/edit") . '</td><td>' .
      l(t('pull'), "admin/subscribe/pull/$sub->sid") . '</td><td>' .
      l(t('delete'), "admin/subscribe/delete/$sub->sid") . '</td></tr></table>'
      );
  }
  $output .= ($row) ? theme('table', $header, $row) : '<p>' . t('You currently have no subscriptions.') . '</p><p>' . t('Would you like to') . ' ' . l(t('subscribe to a channel'),'admin/subscribe/sub/wizard') . '?</p>';

  return $output;
}

function subscribe_delete_form() {
  $sid = arg(3);
  $sub = subscribe_subscription_load($sid);
  if (!$sid || !$sub->sid) {
    drupal_set_message(t('Invalid channel ID.', 'error'));
    drupal_goto('admin/subscribe');
  }
  $form['sid'] = array('#type' => 'value', '#value' => $sid);
  $output = confirm_form('subscribe_delete_form', $form,
                   t('Are you sure you want to delete %title?', array('%title' => theme('placeholder', $sub->name))),
                   'admin/subscribe', t('This action cannot be undone.'),
                   t('Delete'), t('Cancel')  );
  return $output;
}

function subscribe_delete_form_submit($form, $form_values) {
  if ($form_values['confirm']) {
    subscribe_subscription_delete($form_values['sid']);
  }
  return 'admin/subscribe';
}

function subscribe_delete($edit) {
  $sub = subscribe_subscription_load($edit['sid']);
  if ($edit['confirm']) {
    global $user;
    db_query('DELETE FROM {subscribe_subscriptions} WHERE sid = %d', $sub->sid);
    watchdog('subscribe', t('%u: deleted %n', array('%u' => $user->name, '%n' => $sub->name)));
    $result = subscribe_xmlc_cancel_subscription($sub);
  }
  else {
    $extra = form_hidden('sid', $sub->sid);
    $output = theme('confirm',
                    t('Are you sure you want to delete %title?', array('%title' => '<em>'. $sub->name .'</em>')),
                    $_GET['destination'] ? $_GET['destination'] : 'admin/subscribe',
                    t('This action cannot be undone. Nodes imported from this channel will not be deleted.'),
                    t('Delete'),
                    t('Cancel'),
                    $extra);
  }

  return $output;
}

function subscribe_url_form() {
  // this form is a great opportunity for AJAXification
  $edit = isset($_SESSION['edit']) ? $_SESSION['edit'] : array();
  $form = array();
  $form['channel'] = array(
    '#type' => 'fieldset',
    '#title' => 'Subscribe to channel'
  );
  $form['channel']['url'] = array(
    '#type' => 'textfield',
    '#title' => t('URL of site including XML-RPC endpoint'),
    '#size' => '70',
    '#maxlength' => '255',
    '#description' => t('E.g. http://example.com/xmlrpc.php'),
    '#default_value' => isset($edit['url']) ? $edit['url'] : 'http://'
  );
  $form['channel']['channel']['channel_id'] = array(
    '#type' => 'textfield',
    '#title' => 'Channel identifier',
    '#size' => '10',
    '#maxlength' => '255',
    '#description' => t('Typically an integer corresponding to the channel ID'),
    '#default_value' => isset($edit['channel_id']) ? $edit['channel_id'] : ''
  );
  $form['channel']['username'] = array(
    '#type' => 'textfield',
    '#title' => t('Username'),
    '#size' => '10',
    '#maxlength' => '60',
    '#default_value' => isset($edit['username']) ? $edit['username'] : ''
  );
  $form['channel']['pass'] = array(
    '#type' => 'password',
    '#title' => t('Password'),
    '#size' => '10',
    '#maxlength' => '32'
  );
  $form['submit'] = array(
    '#type' => 'submit',
    '#value' => t('Continue')
  );
  unset($_SESSION['edit']);
  return drupal_get_form('subscribe_url_form', $form);
}

/**
 * Validator for initial subscription form.
 */
function subscribe_url_form_validate($form_id, $form_values) {
  if ($form_values['url'] == '') {
    form_set_error('url', t('You must specify the URL of the channel.'));
  }
  elseif (!valid_url($form_values['url'], TRUE)) {
    form_set_error('url', t('That is not a valid URL.'));
  }
  if (form_get_errors()) {
    $_SESSION['edit'] = $form_values;
  }
}

function subscribe_url_form_submit($form_id, $form_values) {
  $url = $form_values['url'];
  $channel_id = check_plain($form_values['channel_id']);
  $username = check_plain($form_values['username']);
  $pass = check_plain($form_values['pass']);
  $result = subscribe_xmlc_may_subscribe($url, $channel_id, $username, $pass);
  if ($result[0]) { // if user may subscribe
    $form_values = array();
    $form_values['name']         = $result[1];
    $form_values['node_types']   = $result[2];
    $form_values['vocabularies'] = $result[3];
    $form_values['base_url']     = $result[4];
    $form_values['channel_id']   = $channel_id;
    $form_values['username']     = $username;
    $form_values['pass']         = $pass;
    $form_values['url']          = $url;

    // now that we have a response we can do preliminary validation

    if (!$form_values['node_types']) {
      drupal_set_message(t('The channel is not currently publishing any node types. This is probably a misconfiguration on the publishing site.'), 'error');
      drupal_goto('admin/subscribe/sub/wizard');
    }

    $nodetypes = node_get_types();
    $missing = array();
    foreach (array_keys($form_values['node_types']) as $type) {
      if (!key_exists($type, $nodetypes)) {
        $missing[] = $type;
      }
    }
    if ($missing) {
      if (count($missing) == 1) {
        $type = $missing[0];
        drupal_set_message(t('The remote site is publishing nodes of type %type but you do not support nodes of this type on your site. Please enable the appropriate module.', array('%type' => theme('placeholder', $type))), 'error');
      }
      else {
        $phrase = '';
        foreach ($missing as $type) {
          $phrase .= ', ' . $type;
        }
        drupal_set_message(t('The remote site is publishing nodes of the following types: %types; but you do not support nodes of these types on your site. Please enable the appropriate modules.', array('%types' => theme('placeholder', ltrim($phrase, ',')), 'error')));
      }
      drupal_goto('admin/subscribe/sub/wizard');
    }
  }
  else {
    $msg = t('Unable to subscribe.');
    drupal_set_message($msg, 'error');
    $_SESSION['edit'] = $form_values;
    drupal_goto('admin/subscribe/sub/wizard');
  }
  $_SESSION['edit'] = $form_values;
  return ('admin/subscribe/sub/vocabmap');
}


/**
 * The form for mapping vocabularies from remote vocabularies to local vocabularies
 * if the publishing site is publishing vocabularies; also sets push/pull.
 */
function subscribe_vocabmap_form() {
  if (!isset($_SESSION['edit'])) {
    drupal_set_message(t('Unable to retrieve session data.'), 'error');
    drupal_goto('/subscribe/sub/wizard');
  }
  $edit = $_SESSION['edit'];
  $remote_vocabularies = array();

  foreach ($edit['vocabularies'] as $vid => $va) {
    $remote_vocabularies[$vid] = (object)$va;
  }

  $vocabularies = taxonomy_get_vocabularies();
  $voc_names = array();
  $voc_options = array();
  if (!$vocabularies && $remote_vocabularies) {
    drupal_set_message(t('In order to map remote vocabularies to local vocabularies, you must first create your local vocabularies using administer &gt; categories.'), 'error');
  }

  foreach ($vocabularies as $v) {
    $voc_names[$v->name] = $v->vid;
    $voc_options[$v->vid] = $v->name;
  }
  $voc_options['0'] = t('discard');

  // lots of error checking needs to go here to make sure
  // the local vocabulary select box only includes vocabularies
  // whose attributes match the remote vocabularies

  $form = array();
  $form['vocabmap'] = array();

  foreach ($remote_vocabularies as $vid => $v) {
    // autoselect
    $selected_voc = '';
    if (isset($voc_names[$v->name])) {
      $selected_voc = $voc_names[$v->name];
    }

    $form['vocabmap']['localvoc' . $vid] = array(
      '#type' => 'select',
      '#options' => $voc_options,
      '#default_value' => $selected_voc,
      '#vname' => $v->name
    );
  }

  $form['receiving'] = array(
    '#type' => 'fieldset',
    '#title' => t('Receiving Content')
  );
  $form['receiving']['pushpull'] = array(
    '#type' => 'radios',
    '#title' => t('Method'),
    '#default_value' => isset($edit['pushpull']) ? $edit['pushpull'] : 0,
    '#options' => array(t('Push: Receive content from channel automatically when the channel decides to send it.'), t('Pull: Receive content from channel only when you manually request it (your server must allow HTTP out).')),
    '#description' => t('Choose how you would like this site to receive content.')
  );
  $form['submit'] = array(
    '#type' => 'submit',
    '#value' => t('Subscribe to channel')
  );

  return drupal_get_form('subscribe_vocabmap_form', $form);
}

function theme_subscribe_vocabmap_form($form) {
  $header = array(t('Remote Vocabulary'), '', t('Local Vocabulary'));
  $rows = array();
  foreach (element_children($form['vocabmap']) as $key) {
    $rows[] = array(
      array('data' => $form['vocabmap'][$key]['#vname']),
      array('data' => '&#8594;'),
      array('data' => form_render($form['vocabmap'][$key]))
    );
  }
  if ($rows) {
    $output = theme('fieldset', array('#title' => t('Vocabulary Mapping'), '#children' => theme_table($header, $rows)));
  }
  $output .= form_render($form);
  return $output;
}

function subscribe_vocabmap_form_validate($form, $form_values) {
  if (!isset($_SESSION['edit'])) {
    drupal_set_message(t('Unable to retrieve session data.'), 'error');
    drupal_goto('/subscribe/sub/wizard');
  }
  $edit = $_SESSION['edit'];
   if (!subscribe_subscription_validate($edit['url'], $edit['channel_id'])) {
    drupal_set_message(t('You are already subscribed to this channel.'), 'error');
    drupal_goto('admin/subscribe');
   }
}

function subscribe_vocabmap_form_submit($form, $edit) {
  if (!isset($_SESSION['edit'])) {
    drupal_set_message(t('Unable to retrieve session data.'), 'error');
    drupal_goto('/subscribe/sub/wizard');
  }
  $edit = array_merge($_SESSION['edit'], $edit);
  unset($_SESSION['edit']);
  $edit['channel_id'] = check_plain($edit['channel_id']);
  $edit['username'] = check_plain($edit['username']);
  $edit['pass'] = check_plain($edit['pass']);

  $remote_vids = array();
  $local_vids = array();
  $vocab_map = _subscribe_extract_vocab_map($edit);
  foreach ($vocab_map as $rvid => $local_vid) {
    if ($local_vid != 0) {
      $remote_vids[] = $rvid;
      $local_vids[] = $local_vid;
    }
  }

  $result = subscribe_xmlc_subscribe($edit['url'], $edit['channel_id'], $edit['username'], $edit['pass'], $edit['pushpull'], $local_vids, $remote_vids, array());
  if ($result['status'] = 'ok') {
    $edit['rsid'] = $result['sid'];
    $edit['method'] = $edit['pushpull'];
    $edit['sid'] = subscribe_subscription_save($edit);
    subscribe_vocabmap_save($vocab_map, $edit['sid']);
    drupal_set_message(t('Subscription to channel %name established.', array('%name' => theme('placeholder', $edit['name']))));
    drupal_goto('admin/subscribe');
  }
  else {
    drupal_set_message(t('Could not subscribe.'), 'error');
    drupal_goto('admin/subscribe');
  }
}

/**
 * Save local-remote vocabulary mapping.
 *
 * @param $local_vocabularies
 *   Array which must contain local/remote mappings
 *   encoded as e.g., $edit['localvoc2'] = 4.
 * @param $sid subscription id
 *
 * @return
 *
 */
function subscribe_vocabmap_save($local_vocabularies, $sid) {
  db_query("DELETE FROM {subscribe_vocab_map} WHERE sid = '%s'", $sid);
  foreach ($local_vocabularies as $lvid => $rvid) {
    db_query("INSERT INTO {subscribe_vocab_map} (sid, remote_vid, local_vid) VALUES (%d, %d, %d)", $sid, $rvid, $lvid);
  }
}

/**
 * Generate a mapping of local vocabularies for a subscription
 * keyed by remote vocabulary ID: $vocabmap[3] = 4 means that
 * vocabulary 4 on the remote site maps to vocabulary 3 on this site.
 *
 * @param $sid
 *   the local subscription ID
 */
function subscribe_get_vocabmap($sid) {
  $vocabmap = array();
  $result = db_query("SELECT * FROM {subscribe_vocab_map} WHERE sid = %d", $sid);
  while ($data = db_fetch_object($result)) {
    $vocabmap[$data->remote_vid] = taxonomy_get_vocabulary($data->local_vid);
  }

  return $vocabmap;
}


/**
 * Using the URL of the channel, determine whether we already
 * have a subscription to this channel.
 *
 * @param $url
 *   the URL of the channel
 * @param $channel_id
 *   the id of the channel
 */
function subscribe_subscription_validate($url, $channel_id) {
  $result = db_query("SELECT * FROM {subscribe_subscriptions} WHERE url = '%s' AND channel_id = %d", $url, $channel_id);
  $data = db_fetch_object($result);
  return !$data;
}

/**
 * Return a subscription object representing a database row for a given sid.
 *
 * @param $sid
 *   the local ID of the subscription
 */
function subscribe_subscription_load($sid) {
  return db_fetch_object(db_query('SELECT * FROM {subscribe_subscriptions} WHERE sid = %d', $sid));
}

/**
 * Save a subscription to a channel to the database.
 *
 * @param $edit
 *   array from form;
 */
function subscribe_subscription_save($edit) {
  global $base_url;
  global $user;
  $fields = _subscribe_db_fields('subscribe_subscriptions');
  //  sid      rsid      name      base_url    url      channel_id    domain      username      pass      token      method   last_update

  $parts = parse_url($edit['url']);
  $edit['domain'] = $parts['host'];

  if ($edit['sid'] && db_result(db_query('SELECT COUNT(sid) FROM {subscribe_subscriptions} WHERE sid = %d', $edit['sid']))) { // existing subscription
    // Prepare the query:
    foreach ($edit as $key => $value) {
      if (in_array($key, $fields)) {
        $q[] = db_escape_string($key) ." = '%s'";
        $v[] = $value;
      }
    }
    db_query('UPDATE {subscribe_subscriptions} SET '. implode(', ', $q) .' WHERE sid = '. db_escape_string($edit['sid']), $v);
  }
  else { // new subscription
    $edit['sid'] = db_next_id('subscribe');
    $edit['token'] = md5($edit['username'] . $edit['pass'] . $base_url);
    // Prepare the query:
    foreach ($edit as $key => $value) {
      if (in_array((string) $key, $fields)) {
        $k[] = db_escape_string($key);
        $v[] = $value;
        $s[] = "'%s'";
      }
    }
    // Insert the subscription record into the database:
    db_query('INSERT INTO {subscribe_subscriptions} ('. implode(", ", $k) .') VALUES('. implode(', ', $s) .')', $v);
    watchdog('subscribe', t('%u subscribed to channel %c', array('%u' => $user->name, '%c' => $edit['name'])));
  }

  return $edit['sid'];
}

function subscribe_subscription_delete($sid) {
  global $user;
  $sub = subscribe_subscription_load($sid);
  db_query('DELETE FROM {subscribe_subscriptions} WHERE sid = %d', $sid);
  db_query('DELETE FROM {subscribe_vocab_map WHERE sid = %d', $sid);
  watchdog('subscribe', t('%u: deleted subscription %n', array('%u' => $user->name, '%n' => $sub->name)));
  $result = subscribe_xmlc_cancel_subscription($sub);
}

/**
 * Present the form where a user can request to receive updated data
 * from a channel.
 *
 * @param $sid
 *   the local subscription ID
 */
function subscribe_pull_form() {

  $sid = arg(3);
  $sub = subscribe_subscription_load($sid);
  if (!$sid || !$sub->sid) {
    drupal_set_message(t('Invalid channel ID.', 'error'));
    drupal_goto('admin/subscribe');
  }
  $form['sid'] = array(
    '#type' => 'value',
    '#value' => $sid
  );
  $form['update'] = array(
    '#type' => 'fieldset',
    '#title' => t('Request content from channel %channel', array('%channel' => theme('placeholder', $sub->name)))
  );
  $form['update']['submit'] = array('#type' => 'submit',
    '#value' => t('Update Content')
  );

  return drupal_get_form('subscribe_pull_form', $form);
}

function subscribe_pull_form_submit($form, $form_values) {
  if ($form_values['sid']) {
    $sub = subscribe_subscription_load($form_values['sid']);
    subscribe_pull($sub);
  }
  return 'admin/subscribe';
}

/**
 * Used to invoke content transfer at subscribing site's request.
 *
 * @param $subscription
 *   a subscription object, e.g. as returned by subscribe_subscription_load()
 * @param $cond
 *   an array of triplets to constrain the results (see publish_publish())
 */
function subscribe_pull($subscription, $cond = array()) {
  $changed_filter_present = FALSE;
  foreach ($cond as $triplet) {
    if ($triplet[0] == 'changed') {
      $changed_filter_present = TRUE;
    }
  }

  if (!$changed_filter_present) {
    $cond[] = array('changed', '>', $subscription->last_update);
  }

  $result = subscribe_xmlc_pull($subscription->url, $subscription->channel_id, $subscription->username, $subscription->pass, $subscription->rsid, $cond);
  if (!$result) {
    $error = xmlrpc_error_msg();
    drupal_set_message($error, 'error');
  }
  if ($result['count']) {
    $site_metadata = array('site_url' => $result['site_url'], 'site_name' => $result['site_name'], 'sid' => $subscription->sid);
    list($n_counter, $u_counter, $t_counter) = subscribe_import($result['nodes'], $site_metadata);
    drupal_set_message(t("New content from %name. Nodes imported: %ncount; nodes updated: %ucount; terms imported: %tcount.", array('%ncount' => $n_counter, '%ucount' => $u_counter, '%tcount' => $t_counter, '%name' => $subscription->name)));
    db_query("UPDATE {subscribe_subscriptions} SET last_update = %d WHERE sid = %d", time(), $subscription->sid);
    drupal_goto('admin/subscribe');
  }
  else {
    drupal_set_message(t('Pull attempt failed for %url.', array('%url' => theme('placeholder', check_plain($subscription->url)))), 'error');
  }
}

/********************************************************************
 * Module Functions :: Node Importation
 ********************************************************************/

/**
 * Import nodes into this Drupal site.
 *
 * @param $nodes
 *   an array of nodes, typically from a remote site
 * @param $site_metadata
 *   url and name of remote site and local subscription ID
 *   corresponding to the channel of the remote site
 *
 * @return an array of count of nodes imported, count of taxonomy terms imported
 */
function subscribe_import($nodes, $site_metadata) {

  // required metadata
  $site_url  = $site_metadata['site_url'];
  $site_name = $site_metadata['site_name'];
  $sid       = $site_metadata['sid'];

  $vocab_map = subscribe_get_vocabmap($sid);
  $n_counter = 0;
  $u_counter = 0;
  $t_counter = 0;

  foreach ($nodes as $n) {
    $node = (object)$n;

    // get "external" information
    $remote_url    = $node->url;
    $remote_nid    = $node->nid;
    $remote_terms  = $node->taxonomy_terms;
    $remote_uid    = $node->uid;
    $remote_author = $node->name;

    // discard remote taxonomy
    unset($node->taxonomy);

    // validate or map user
    // currently we just assign to uid 1
    $node->uid = '1';

    // do we already have a copy of this node?
    $local_nid = db_result(db_query("SELECT nid FROM {subscribe_node} WHERE remote_nid = %d AND sid = %d", $node->nid, $sid));

    if ($local_nid) {
      // substitute the local nid for the remote nid
      $node->nid = $local_nid;

      // we assign the revision ID of the old local node to the incoming, updated node
      $local_node = node_load($local_nid);
      $node->vid = $local_node->vid;

      node_save($node);
      $u_counter = $u_counter + 1;

      // will terms be orphaned now if remote terms have changed?
    }
    else {
      // force a new local nid
      $node->nid = '';

      // import node
      node_save($node);
      $local_nid = $node->nid;
      watchdog('subscribe', t('%type: added %title from site %site.', array('%type' => theme('placeholder', t($node->type)), '%title' => theme('placeholder', $node->title), '%site' => theme('placeholder', $site_name))), WATCHDOG_NOTICE, l(t('view'), 'node/'. $node->nid));
      $n_counter = $n_counter + 1;

      // save map of remote/local node ID's (and uids?)
      db_query("INSERT INTO {subscribe_node} (nid, sid, remote_nid, remote_uid, remote_author) VALUES (%d, %d, %d, %d, '%s')", $local_nid, $sid, $remote_nid, $remote_uid, $remote_author);

      // add vocabulary information
      foreach ($node->taxonomy_terms as $remote_term) {
       	// which vocabulary will we be adding it to?
      	$local_vid = $vocab_map[$remote_term['vid']]->vid;
      	if (!$local_vid) {
      		watchdog('subscribe', t('Discarding term %term because remote vocabulary id %vid is not mapped to a local vocabulary.', array('%term' => check_plain($remote_term->name), '%vid' => $remote_term->vid)));
      		continue;
      	}
      	
      	$db_result = db_query("SELECT * FROM {term_data} WHERE LOWER('%s') LIKE LOWER(name) AND vid = %d", trim($remote_term['name']), $local_vid);
        $local_term = db_fetch_object($db_result);

        if ($local_term) { // existing term
          // only add if this term is not already mapped to this node
          $result = db_query("SELECT nid FROM {term_node} WHERE nid = %d AND tid = %d", $local_nid, $local_term->tid);
          $nid = db_result($result);
          if (!$nid) { // this nid-tid pair has not yet been mapped
            db_query('INSERT INTO {term_node} (nid, tid) VALUES (%d, %d)', $local_nid, $local_term->tid);  
          }
        }
        else { // term does not exist locally; add new term
          unset($remote_term['tid']); // local Drupal will assign a new tid
          $remote_term['vid'] = $local_vid;
          $remote_term['weight'] = 0; // discard remote weights
          taxonomy_save_term($remote_term); // sets $remote_term->tid
          $t_counter = $t_counter + 1;
          db_query('INSERT INTO {term_node} (nid, tid) VALUES (%d, %d)', $local_nid, $remote_term->tid);
        }
      }
    }
  }
  return array($n_counter, $u_counter, $t_counter);
}


/********************************************************************
 * Module Functions :: Sending XML-RPC
 ********************************************************************/

/**
 * Ask the channel if we may subscribe to it.
 *
 * @param $edit
 *   array containing a url key with url of the channel
 *
 * @return
 *   array containing maySubscribe (boolean), ppName (string),
 *   nodeTypes (array of published node types), vocabularies (array
 *   of published vocabularies)
 */
function subscribe_xmlc_may_subscribe($url, $channel_id, $username, $pass) {

  $result = xmlrpc($url, 'drupal.publish.maySubscribe', (int)$channel_id, $username, $pass);
  if (!$result) {
    $error = xmlrpc_error_msg();
    if (!$error) {
      $error = t('Are you sure you entered a valid XML-RPC endpoint?');
    }
    drupal_set_message($error, 'error');
    watchdog('error', $error);
  }
  else {
    return $result;
  }
}

/**
 * XMLRPC client to request that remote site which has the publish
 * module installed establish a publishing relationship with client
 * site (that is, this site).
 *
 * @param $url
 *   the URL of the channel on the publishing site
 *   which generally looks like http://drupal.org/publish/6
 * @param $username
 *   the username with access on the publishing site
 * @param $password
 *   the password with access on the publishing site
 * @param $cond
 *   array of condition triplets; see publish_publish()
 *   e.g. ['nid', '=', '3']
 *
 * @return
 *   Subscription ID, a numeric key to to our subscription on the
 *   publishing site (similar to a user ID).
 */
 function subscribe_xmlc_subscribe($url, $channel_id, $username, $pass, $method, $local_vids, $remote_vids, $cond = array()) {
  global $base_url;
  $destination = $base_url . '/xmlrpc.php'; // so the remote server knows where to send updates

  $result = xmlrpc($url, 'drupal.publish.subscribe', (int)$channel_id, $username, $pass, $destination, $method, $local_vids, $remote_vids, $cond);

  if (!$result) {
    $error = xmlrpc_error_msg();
    drupal_set_message($error, 'error');
    watchdog('error', $error);
  }
  else {
    return $result;
  }
}

 /**
 * XMLRPC client to request that remote site which has the publish
 * module installed send an array of nodes to the client
 * site (that is, this site).
 *
 * @param $rsid
 *   the subscription ID of this client on the publishing site
 * @param $username
 *   the username with access on the publishing site
 * @param $password
 *   the password with access on the publishing site
 * @param $url
 *   the URI of the XML-RPC endpoint on the publishing site
 *   which generally looks like http://example.org/xmlrpc.php
 * @param $cond
 *   array of condition triplets; see publish_publish()
 *   e.g. ['nid', '=', '3']
 *
 * @return
 *   Subscription ID, a numeric key to to our subscription on the
 *   publishing site (similar to a user ID).
 */
 function subscribe_xmlc_pull($url, $channel_id, $username, $pass, $rsid, $cond = array()) {

  $result = xmlrpc($url, 'drupal.publish.pull', (int)$channel_id, (int)$rsid, $username, $pass, $cond);

  if (!$result) {
    $error = xmlrpc_error_msg();
    drupal_set_message($error, 'error');
    watchdog('error', $error);
  }
  else {
    return $result;
  }
}

function subscribe_xmlc_cancel_subscription($sub) {
  $result = xmlrpc($sub->url, 'drupal.publish.cancelSubscription', (int)$sub->rsid, $sub->token);
  if (!$result) {
    $error = xmlrpc_error_msg();
    drupal_set_message($error, 'error');
    watchdog('error', $error);
  }

  return $result;
}


/********************************************************************
 * Module Functions :: Receiving XML-RPC
 ********************************************************************/

/*
 * Receive nodes that a remote site is pushing to us
 *
      $message = array(
      'token'    => token,
      'count'     => count($nodes),
      'site_url'  => $base_url,
      'site_name' => variable_get('site_name', 'drupal'),
      'nodes'     => $nodes
      );
 *
 */

function subscribe_xmls_receive($message) {
  $subscription = db_fetch_object(db_query("SELECT * FROM {subscribe_subscriptions} WHERE token = '%s' AND method = 'push' AND rsid = %d", $message['token'], $message['rsid']));
  if (!$subscription) {
    return xmlrpc_error(801, t('No such subscription'));
  }


  // verify that it's from a domain that we trust
  $ip = $_SERVER['REMOTE_ADDR'];
  if ($ip != '127.0.0.1') {
    $host = gethostbyaddr($ip);
    $parts = parse_url($host);
    if ($parts['host'] != $subscription->domain) {
      return xmlrpc_error(701, t('Publishing domain not trusted.'));
    }
  }

  $site_metadata = array('site_url' => $message['site_url'], 'site_name' => $message['site_name'], 'sid' => $subscription->sid);
  list($n_counter, $u_counter, $t_counter) = subscribe_import($message['nodes'], $site_metadata);
  $result = t("New content from %name. Nodes imported: %ncount; noded updated: %ucount; terms imported: %tcount.", array('%ncount' => $n_counter, '%ucount' => $u_counter, '%tcount' => $t_counter, '%name' => $subscription->name));

  return $result;
}

/********************************************************************
 * Module Functions :: Private
 ********************************************************************/

/**
 * Utility function to get vocabulary mapping out of our encoded form.
 *
 * @return
 *   associative array keyed by local vocabulary ID
 */
function _subscribe_extract_vocab_map($edit) {
  $localvoc = array();
  foreach ($edit as $key => $val) {
    if (strstr($key, 'localvoc')) {
      $lvid = substr($key, 8);
      if (is_numeric($lvid) && ($lvid > 0)) $localvoc[$lvid] = $val;
    }
  }
  return $localvoc;
}

/**
 * Return an array of database column names for a given table.
 *
 * This function makes changing the database and module updates a little easier.
 */
function _subscribe_db_fields($table_name) {
  switch ($table_name) {
    case 'subscribe_subscriptions':
      return array('sid', 'rsid', 'name', 'base_url', 'url', 'channel_id', 'domain', 'username', 'pass', 'token', 'method', 'last_update');
  }
}