diff --git a/sf_import/sf_import.module b/sf_import/sf_import.module
index 14d0e79..134e31b 100644
--- a/sf_import/sf_import.module
+++ b/sf_import/sf_import.module
@@ -1,28 +1,86 @@
- array(
- 'page callback' => 'drupal_get_form',
- 'page arguments' => array('sf_import_create'),
+ 'page callback' => 'sf_import_overview',
'type' => MENU_LOCAL_TASK,
'access arguments' => array('administer salesforce'),
'title' => 'Import',
),
+ SALESFORCE_PATH_ADMIN_IMPORT . '/create' => array(
+ 'page callback' => 'drupal_get_form',
+ 'page arguments' => array('sf_import_form'),
+ 'type' => MENU_LOCAL_TASK,
+ 'access arguments' => array('administer salesforce'),
+ 'title' => 'Create Import Configuration',
+ ),
+ SALESFORCE_PATH_ADMIN_IMPORT . '/overview' => array(
+ 'page callback' => 'sf_import_overview',
+ 'type' => MENU_DEFAULT_LOCAL_TASK,
+ 'access arguments' => array('administer salesforce'),
+ 'title' => 'Index',
+ ),
+ SALESFORCE_PATH_ADMIN_IMPORT . '/%/edit' => array(
+ 'page callback' => 'drupal_get_form',
+ 'page arguments' => array('sf_import_form', $i),
+ 'type' => MENU_LOCAL_TASK,
+ 'access callback' => '_sf_import_accesscallback',
+ 'access arguments' => array('administer salesforce', 'edit', $i),
+ 'title' => 'Edit Import',
+ ),
+ SALESFORCE_PATH_ADMIN_IMPORT . '/%/delete' => array(
+ 'page callback' => 'drupal_get_form',
+ 'page arguments' => array('sf_import_delete_form', $i),
+ 'type' => MENU_CALLBACK,
+ 'access callback' => '_sf_import_accesscallback',
+ 'access arguments' => array('administer salesforce', 'delete', $i),
+ 'title' => 'Delete Import Configuration',
+ ),
);
}
-function sf_import_create(&$form_state, $ongoing = 0) {
+/**
+ * Access callback for CRUD links.
+ */
+function _sf_import_accesscallback($perm, $op, $id = NULL) {
+ if (!empty($id) && is_numeric($id)) {
+ if (sf_import_load($id)) {
+ return user_access($perm);
+ }
+ }
+ return FALSE;
+}
+/**
+ * Add/Edit form for import configurations
+ */
+function sf_import_form(&$form_state, $id = NULL) {
$form = $options = array();
+ if (!empty($id)) {
+ $import_config = sf_import_load($id);
+ if (empty($import_config)) {
+ drupal_set_message(t('Invalid import config id.'), 'error');
+ return;
+ }
+ }
$fieldmaps = salesforce_api_salesforce_field_map_load_all();
+
foreach ($fieldmaps as $map) {
- $options[$map->name] =
+ $options[$map->name] =
salesforce_api_fieldmap_object_label('salesforce', $map->salesforce) . ' => ' .
salesforce_api_fieldmap_object_label('drupal', $map->drupal);
+ // Add description if there is one
+ if ($map->description) {
+ $options[$map->name] .= ' (' . $map->description . ')';
+ }
}
// Add a message if no objects have been mapped.
@@ -30,21 +88,22 @@ function sf_import_create(&$form_state, $ongoing = 0) {
drupal_set_message(t('You have not yet defined any fieldmaps.'), 'error');
return;
}
-
+
// Admin should select a mapping to use for the import.
$form['label'] = array(
- '#type' => 'markup',
- '#value' => '' . ($ongoing ? t('Create Ongoing Import') : t('Perform One-time Import')) . '
',
+ '#type' => 'item',
+ '#title' => $id ? t('Edit Import Configuration') : t('Import data from Salesforce'),
);
-
+
$form['fieldmap'] = array(
'#title' => t('Please choose a fieldmap to use for the import'),
'#description' => t('Salesforce Object => Drupal Content Type'),
'#type' => 'radios',
'#required' => TRUE,
'#options' => $options,
+ '#default_value' => $import_config->name,
);
-
+
$form['extra-options'] = array(
'#title' => t('Extra Options'),
'#type' => 'fieldset',
@@ -52,35 +111,72 @@ function sf_import_create(&$form_state, $ongoing = 0) {
'#collapsed' => FALSE,
);
- $form['extra-options']['extra-linked'] = array(
- '#title' => t('Link nodes to Salesforce objects on import?'),
- '#description' => t('Links the imported Drupal node to the salesforce object allowing the ability to issue manual syncronization of data to and from Drupal and Salesforce business objects. Linking also enables the ability to use node reference to relate business objects in Drupal (like accounts to contacts).'),
+ // If we are not dealing with an existing import config, checking this cb will
+ // create a cronned import config on submit.
+ if (empty($id)) {
+ $form['extra-options']['extra-ongoing'] = array(
'#type' => 'checkbox',
- );
-
+ '#title' => t('Create ongoing import'),
+ '#description' => t('Ongoing imports will run once immediately, then poll Salesforce on cron runs to gather updated records and synchronize them with Drupal. !view', array('!view' => l('View all ongoing imports', SALESFORCE_PATH_ADMIN_IMPORT . '/overview'))),
+ );
+ }
+
$form['extra-options']['extra-where'] = array(
- '#title' => t('Conditions'),
- '#description' => t("Advanced: Enter any additional SOQL \"Where\" conditions to use for this import query, e.g.
Type != 'One-to-One Individual'
Learn more here: Salesforce.com SOQL Where clause"),
- '#type' => 'textarea',
+ '#title' => t('Conditions'),
+ '#description' => t("Advanced: Enter any additional SOQL \"Where\" conditions to use for this import query, e.g.
Type != 'One-to-One Individual'
Learn more here: Salesforce.com SOQL Where clause"),
+ '#type' => 'textarea',
+ '#default_value' => $import_config->conf['extra-where'],
);
-
- $form['ongoing'] = array('#type' => 'value', '#value' => $ongoing);
- $form['submit'] = array('#type' => 'submit', '#value' => 'submit');
-
+
+ $form['submit'] = array('#type' => 'submit', '#value' => t('Submit'));
+ if ($id) {
+ $form['id'] = array('#type' => 'value', '#value' => $id);
+ $form['delete'] = array('#type' => 'submit', '#value' => t('Delete'));
+ }
return $form;
}
-function sf_import_create_submit($form, &$form_state, $ongoing = 0) {
+/**
+ * Submit handler for import config add/edit form.
+ * @see sf_import_form()
+ */
+function sf_import_form_submit($form, &$form_state) {
// Create a new batch job to do the import
-
$extra_options = array();
- $extra_options['extra-linked'] = $form_state['values']['extra-linked'];
$extra_options['extra-where'] = $form_state['values']['extra-where'];
- $batch = sf_import_create_batchjob($form_state['values']['fieldmap'], $extra_options);
- batch_set($batch);
+ // If we are creating or updating an import config, write to the database.
+ if ($form_state['values']['extra-ongoing'] || $form_state['values']['id']) {
+ $object = (object)array('name' => $form_state['values']['fieldmap'], 'conf' => $extra_options);
+ if ($form_state['values']['id']) {
+ $object->id = $form_state['values']['id'];
+ $conf = db_result(db_query('SELECT conf FROM {salesforce_import} WHERE id = %d', $object->id));
+ // Preserve any existing values that were not in the form.
+ if (!empty($conf)) {
+ $conf = unserialize($conf);
+ if (is_array($conf)) {
+ // Clear any exceptions to allow this import config to be run again.
+ unset($conf['cron-exception']);
+ $object->conf = array_merge($conf, $object->conf);
+ }
+ }
+ $object->conf = serialize($object->conf);
+ db_query('REPLACE INTO {salesforce_import} SET (id, name, last, conf) VALUES (%d, "%s", %d, "%s")', $object->id, $object->name, time(), $object->conf);
+ }
+ else {
+ $object->conf = serialize($object->conf);
+ db_query('INSERT INTO {salesforce_import} (conf, name, last) VALUES ("%s", "%s", %d)', $object->conf, $object->name, time());
+ }
+ }
+
+ // Start the batch.
+ $batch = sf_import_create_batchjob($form_state['values']['fieldmap'], $extra_options);
+ batch_set($batch);
}
+/**
+ * Helper function to build a Batch-API formatted array to process the import.
+ */
function sf_import_create_batchjob($fieldmap, $extra = NULL) {
$params = array('fieldmap_key' => $fieldmap, 'extra' => $extra);
return array(
@@ -88,21 +184,27 @@ function sf_import_create_batchjob($fieldmap, $extra = NULL) {
'operations' => array(
array('sf_import_batchjob', $params)),
'finished' => 'sf_import_batchjob_finalize',
- );
+ );
}
+/**
+ * Finalize callback for Batch API import job.
+ */
function sf_import_batchjob_finalize($success, $results, $operations) {
if ($success) {
drupal_set_message(t('Import complete.'));
- if (count($results) > 0) {
- drupal_set_message(theme('item_list', $results));
- }
}
else {
drupal_set_message(t('Import failed.'));
}
+ if (count($results) > 0) {
+ drupal_set_message(theme('item_list', $results));
+ }
}
+/**
+ * The Batch API worker callback.
+ */
function sf_import_batchjob($fieldmap_key, $extra, &$context) {
// Always log in to salesforce.
if (empty($context['sandbox'])) {
@@ -119,9 +221,9 @@ function sf_import_batchjob($fieldmap_key, $extra, &$context) {
$context['sandbox']['salesforce']['map'] = $map;
// Load the object definitions.
- $context['sandbox']['salesforce']['drupal_object'] =
+ $context['sandbox']['salesforce']['drupal_object'] =
salesforce_api_fieldmap_objects_load('drupal', $map->drupal);
- // $context['sandbox']['salesforce']['salesforce_object'] =
+ // $context['sandbox']['salesforce']['salesforce_object'] =
// salesforce_api_fieldmap_objects_load('salesforce', $map['salesforce']);
$sql = 'SELECT oid, sfid FROM {salesforce_object_map} WHERE name = "%s"';
$result = db_query($sql, $map->name);
@@ -131,28 +233,30 @@ function sf_import_batchjob($fieldmap_key, $extra, &$context) {
$soql = 'SELECT '
// "Id" must be included in the SOQL query.
- . implode(', ', array_keys($map->fields + array('Id' => '')))
+ . implode(', ', array_keys($map->fields + array('Id' => '', 'LastModifiedDate' => '')))
. ' FROM ' . $map->salesforce;
if (!empty($extra['extra-where'])) {
$soql .= ' WHERE ' . $extra['extra-where'];
}
try {
$sf = salesforce_api_connect();
- $context['sandbox']['salesforce']['query'] = $query = $sf->client->query($soql);
- } catch (Exception $e) {
+ $context['sandbox']['salesforce']['query'] = $query = $sf->client->query($soql);
+ }
+ catch (Exception $e) {
$context['finished'] = 1;
$context['message'] = $e->getMessage();
- return;
+ $context['sandbox']['exception'] = 1;
+ return;
}
if (empty($query->records)) {
- $context['finished'] = 1;
+ $context['finished'] = 1;
$context['message'] = 'Empty resultset returned from Salesforce query.';
}
$context['sandbox']['progress'] = 0;
$context['sandbox']['position'] = 0;
$context['sandbox']['imported'] = 0;
$context['finished'] = 0;
- }
+ }
$map = $context['sandbox']['salesforce']['map'];
$query = $context['sandbox']['salesforce']['query'];
@@ -176,6 +280,7 @@ function sf_import_batchjob($fieldmap_key, $extra, &$context) {
} catch (Exception $e) {
$context['finished'] = 1;
$context['message'] = $e->getMessage();
+ $context['sandbox']['exception'] = 1;
return;
}
}
@@ -192,16 +297,15 @@ function sf_import_batchjob($fieldmap_key, $extra, &$context) {
if (!is_array($record)) {
$record = get_object_vars($record);
}
- $created = isset($existing[$record['Id']]);
+ $created = !array_key_exists($record['Id'], $existing);
$type = $map->drupal;
- // "node" mappings are like "node_contenttype".
+ // "node" mappings are like "node_contenttype".
// others are like "user", "uc_order", etc.
if (strpos($type, 'node_') === 0) {
$type = 'node';
}
$function = 'sf_' . $type . '_import';
-
if (function_exists($function)) {
$oid = $function($record, $map->name, $existing[$record['Id']]);
}
@@ -220,3 +324,168 @@ function sf_import_batchjob($fieldmap_key, $extra, &$context) {
$context['sandbox']['position']++;
$context['finished'] = $context['sandbox']['progress'] / $size;
}
+
+/**
+ * Load function for import configurations.
+ * @param $id - numeric id of the import config.
+ * @return the import config object
+ */
+function sf_import_load($id = NULL) {
+ $sql = 'SELECT id, name, last, conf FROM {salesforce_import}';
+ $args = array();
+ if (!empty($id)) {
+ $sql .= ' WHERE id = %d';
+ $object = db_fetch_object(db_query($sql, $id));
+ $object->conf = unserialize($object->conf);
+ return $object;
+ }
+ else {
+ $result = db_query($sql);
+ $objects = array();
+ while ($object = db_fetch_object($result)) {
+ $object->conf = unserialize($object->conf);
+ $objects[] = $object;
+ }
+ return $objects;
+ }
+}
+
+/**
+ * Index of import configs.
+ */
+function sf_import_overview() {
+ $configs = sf_import_load();
+ $rows = array();
+ if (empty($configs)) {
+ return t('No existing import configurations.');
+ }
+ foreach ($configs as $config) {
+ $map_link = l('edit fieldmap', SALESFORCE_PATH_FIELDMAPS . '/' . $config->name . '/edit');
+
+ $details = 'Last run ' . format_date($config->last);
+ if ($config->conf['cron-remaining']) {
+ $details .= '
Remaining records: ' . $config->conf['cron-remaining'];
+ }
+ if ($config->conf['cron-exception']) {
+ $details .= '
Last Error: ' . $config->conf['cron-exception'] . '
';
+ }
+
+ $edit_link = l('edit', SALESFORCE_PATH_ADMIN_IMPORT . '/' . $config->id . '/edit');
+ $del_link = l('delete', SALESFORCE_PATH_ADMIN_IMPORT . '/' . $config->id . '/delete');
+
+ $rows[] = array(
+ $config->name . ' [' . $map_link . ']',
+ $details,
+ $edit_link . ' | ' . $del_link,
+ );
+ }
+ $header = array(t('Fieldmap'), t('Details'), t('Import Configuration Actions'));
+ return theme('table', $header, $rows);
+}
+
+/**
+ * Confirm form for deleting import configs.
+ */
+function sf_import_delete_form($form, $id = NULL) {
+ if (empty($id)) {
+ drupal_not_found();
+ exit;
+ }
+ $config = sf_import_load($id);
+ if (empty($config)) {
+ drupal_not_found();
+ exit;
+ }
+ $form = array('id' => array('#type' => 'value', '#value' => $id));
+ return confirm_form($form, 'Are you sure you want to delete this import configuration?', SALESFORCE_PATH_ADMIN_IMPORT, 'This action cannot be undone.', 'Delete', 'Cancel', 'confirm');
+}
+
+/**
+ * Submit handler for import config delete form.
+ * @see sf_import_delete_form()
+ */
+function sf_import_delete_form_submit($form, &$form_state) {
+ $form_state['redirect'] = SALESFORCE_PATH_ADMIN_IMPORT;
+ $sql = 'DELETE FROM {salesforce_import} WHERE id = %d';
+ db_query($sql, $form_state['values']['id']);
+}
+
+/**
+ * Implements hook_cron.
+ * Process one import config per cron run, based on the oldest run. Use at most
+ * 25% of the entire cron run. If we run out of time during a run, then update
+ * the "last" run timestamp to be equal to the LastModifiedDate of the
+ * last-processed record. Then, on subsequent runs, we'll proceed from that
+ * record forward.
+ */
+function sf_import_cron() {
+ $result = db_query('SELECT id, name, last, conf FROM {salesforce_import} ORDER BY last');
+ while ($object = db_fetch_object($result)) {
+ $object->conf = unserialize($object->conf);
+ // Don't try to run an import that fired an exception.
+ if (!empty($object->conf['cron-exception'])) {
+ watchdog('sf_import', 'Import configuration was not run due to Salesforce exception(s). !message
', array('!message' => $object->conf['cron-exception']), WATCHDOG_ERROR, l('edit config', SALESFORCE_PATH_ADMIN_IMPORT . '/' . $object->id . '/edit'));
+ continue;
+ }
+ else {
+ break;
+ }
+ }
+ if (empty($object)) {
+ return;
+ }
+
+ $conf = $object->conf;
+ // Limit our imports to a specific time period. We need only gather Salesforce
+ // records modified since our last run.
+ $soql_where = '';
+ if (!empty($conf['extra-where'])) {
+ $soql_where .= $conf['extra-where'] . ' AND ';
+ }
+ $soql_where .= ' LastModifiedDate > ' . gmdate(DATE_ATOM, $object->last);
+ $conf['extra-where'] = $soql_where . ' ORDER BY LastModifiedDate';
+
+ // Fake the batch api harness and start the actual processing.
+ $context = array('finished' => 0);
+ $request_time = time();
+ // TODO: make this configurable:
+ $duration = .25;
+ $limit = ini_get('max_execution_time') * $duration;
+ while ($context['finished'] < 1 && time() < $request_time + $limit) {
+ sf_import_batchjob($object->name, $conf, $context);
+ }
+
+ // If the batch stopped prematurely, save our place so we can begin where we
+ // left off on the next run.
+ if ($context['finished'] < 1 || $context['sandbox']['exception']) {
+ // If there was an exception, mark the config so that we don't keep
+ // re-running a faulty import.
+ // TODO: smart error handling. Some exceptions are temporary, e.g. network
+ // disconnects, and some are fatal, e.g. invalid syntax in WHERE clause.
+ // Temporary exceptions should not block future import processing.
+ if ($context['sandbox']['exception']) {
+ $object->conf['cron-exception'] = $context['message'];
+ }
+ else {
+ // Fudge the "last" value to point to the timestamp of the last-processed
+ // Salesforce record.
+ // TODO: implement a better way to keep track of our progress. This
+ // implementation relies on no two Salesforce records having the exact
+ // same modification timestamp, which may not be realistic.
+ // Maybe store $context to the database and skip the query-portion of the
+ // batch altogether on subsequent runs?
+ $query = $context['sandbox']['salesforce']['query'];
+ $pos = $context['sandbox']['position'];
+ $record = $query_array->records[$pos - 1];
+ $object->last = strtotime($record->LastModifiedDate);
+ $object->conf['cron-remaining'] = $query_array['size'] - $pos;
+ }
+ db_query('UPDATE {salesforce_import} SET conf = "%s", last = %d WHERE id = %d', serialize($object->conf), $object->last, $object->id);
+ }
+ else {
+ // If everything went smoothly, remove any irrelevant configuration data
+ // and update the import config's timestamp.
+ unset($object->conf['cron-remaining'], $object->conf['cron-exception']);
+ db_query('UPDATE {salesforce_import} SET last = %d, conf = "%s" WHERE id = %d', $request_time, serialize($object->conf), $object->id);
+ }
+}