diff --git a/modules/salesforce_push/salesforce_push.module b/modules/salesforce_push/salesforce_push.module index 718af73..00227be 100644 --- a/modules/salesforce_push/salesforce_push.module +++ b/modules/salesforce_push/salesforce_push.module @@ -155,6 +155,7 @@ function salesforce_push_sync_rest($entity_type, $entity, $mapping, $sf_sync_tri $mapped_objects = entity_create('salesforce_mapping_object', array( 'entity_id' => $entity_wrapper->getIdentifier(), 'entity_type' => $entity_type, + // 'bundle' => $entity_wrapper->getBundle(), //only applies if bundle added to salesforce_mapping_object 'salesforce_id' => $data['id'], )); } @@ -209,7 +211,7 @@ function salesforce_push_cron() { if (!$sfapi->isAuthorized()) { return; } - + $payloadKeys = array(); //only applies if bundle added to salesforce_mapping_object $queue = DrupalQueue::get(SALESFORCE_PUSH_QUEUE); $limit = variable_get('salesforce_pull_limit', 50); $use_soap = module_exists('salesforce_soap'); @@ -220,27 +222,37 @@ function salesforce_push_cron() { if (!$item) { break; } + $mapping = $item->data['mapping']; - + // Duplicate entity in the queue. if ($item->data['entity_type'] == $entity_type && $item->data['entity_id'] == $entity_id) { $queue->deleteItem($item); continue; } - + $entity_type = $item->data['entity_type']; $entity_id = $item->data['entity_id']; $mapped_objects = salesforce_mapping_object_load_by_drupal($entity_type, $entity_id); $entity = entity_load_single($entity_type, $entity_id); + if ($use_soap) { if ($item->data['trigger'] == SALESFORCE_MAPPING_SYNC_DRUPAL_DELETE && $mapped_objects) { $delete_list[$delta] = $mapped_objects->salesforce_id; } else { $wrapper = entity_metadata_wrapper($item->data['entity_type'], $entity); - $params = salesforce_push_map_params($mapping, $wrapper, $key_field, $key_value); - - $synced_entities[$delta] = array( + $params = salesforce_push_map_params($mapping, $wrapper, $key_field, $key_value, $use_soap); + + //only applies if bundle added to salesforce_mapping_object + //Some bundles may have the same key + if($payloadKeys[$key_value] == $key_value){ + continue; + }else{ + $payloadKeys[$key_value] = $key_value; + } + + $synced_entities[$mapping->salesforce_object_type][$delta] = array( 'entity_wrapper' => $wrapper, 'mapped_objects' => $mapped_objects, ); @@ -248,16 +260,27 @@ function salesforce_push_cron() { $sobject = new stdClass(); $sobject->type = $mapping->salesforce_object_type; foreach ($params as $key => $value) { - $sobject->fields[$key] = $value; + $sobject->fields[$key] = htmlspecialchars($value); } if ($mapped_objects && $mapped_objects->salesforce_id) { $sobject->Id = $mapped_objects->salesforce_id; - $update_list[$delta] = $sobject; + + $synced_entities['update_list'][$mapping->salesforce_object_type][$delta] = array( + 'entity_wrapper' => $wrapper, + 'mapped_objects' => $mapped_objects, + ); + + + $update_list[$mapping->salesforce_object_type][$delta] = $sobject; } else { if ($key_field && $key_value) { - $upsert_list[$key_field][$delta] = $sobject; + $synced_entities['upsert_list'][$mapping->salesforce_object_type][$delta] = array( + 'entity_wrapper' => $wrapper, + 'mapped_objects' => $mapped_objects, + ); + $upsert_list[$mapping->salesforce_object_type][$key_field][$delta] = $sobject; } else { $create_list[$delta] = $sobject; @@ -272,7 +295,7 @@ function salesforce_push_cron() { // Remove item from queue. $queue->deleteItem($item); } - + // Use soap API to batch process records. if ($use_soap) { module_load_include('inc', 'salesforce_soap'); @@ -288,15 +311,23 @@ function salesforce_push_cron() { } if (!empty($update_list)) { - $results = $soap->update($update_list); - salesforce_push_process_soap_results('Update', $results, $synced_entities); + foreach ($update_list as $sObjectType => $update_items) { + $synced_entities_indexed = array_values($synced_entities['update_list'][$sObjectType]); + $update_items = array_values($update_items); + $results = $soap->update($update_items); + salesforce_push_process_soap_results('Update', $results, $synced_entities_indexed); + } } if (!empty($upsert_list)) { - foreach ($upsert_list as $key => $upsert_item) { - $results = $soap->upsert($key, $upsert_item); - salesforce_push_process_soap_results('Upsert', $results, $synced_entities); - } + foreach ($upsert_list as $sObjectType => $upsert_items) { + foreach ($upsert_items as $key => $upsert_item) { + $synced_entities_indexed = array_values($synced_entities['upsert_list'][$sObjectType]); + $upsert_item = array_values($upsert_item); + $results = $soap->upsert($key, $upsert_item); + salesforce_push_process_soap_results('Upsert', $results, $synced_entities_indexed); + } + } } } } @@ -312,7 +343,16 @@ function salesforce_push_cron() { * Entities that were synced with Salesforce. */ function salesforce_push_process_soap_results($op, $results, $synced_entities) { + //Handle single item result. Multiple items will be array, single item is an object + if(is_object($results)){ + $tempSObject = $results; + unset($results); + $results = array(); + $results[0] = $tempSObject; + } + foreach ($results as $key => $result) { + if ($result->success) { $mapped_objects = $synced_entities[$key]['mapped_objects']; if (drupal_strtolower($op) == 'delete' && $mapped_objects) { @@ -321,12 +361,26 @@ function salesforce_push_process_soap_results($op, $results, $synced_entities) { else { // Create mapping object, saved below. $wrapper = $synced_entities[$key]['entity_wrapper']; + + //Check if entity exists already + $query = new EntityFieldQuery(); + $entityFieldQueryResult = $query->entityCondition('entity_type', 'salesforce_mapping_object') + ->propertyCondition('entity_id', $wrapper->getIdentifier()) + // ->propertyCondition('bundle', $wrapper->getBundle()) //only applies if bundle added to salesforce_mapping_object + ->execute(); + + if(isset($entityFieldQueryResult['salesforce_mapping_object']) && is_array($entityFieldQueryResult['salesforce_mapping_object'])){ + //Update entity? + }else{ $mapped_objects = entity_create('salesforce_mapping_object', array( 'entity_id' => $wrapper->getIdentifier(), 'entity_type' => $wrapper->type(), + // 'bundle' => $wrapper->getBundle(), //only applies if bundle added to salesforce_mapping_object 'salesforce_id' => $result->id, )); - $mapped_objects->save(); + $mapped_objects->save(); + } + } watchdog('salesforce_push', '%op: Salesforce object %id', @@ -364,7 +418,7 @@ function salesforce_push_process_soap_results($op, $results, $synced_entities) { * @return array * Associative array of key value pairs. */ -function salesforce_push_map_params($mapping, $entity_wrapper, &$key_field, &$key_value) { +function salesforce_push_map_params($mapping, $entity_wrapper, &$key_field, &$key_value, $use_soap = FALSE) { foreach ($mapping->field_mappings as $fieldmap) { // Skip fields that aren't being pushed to Salesforce. if (!in_array($fieldmap['direction'], array(SALESFORCE_MAPPING_DIRECTION_DRUPAL_SF, SALESFORCE_MAPPING_DIRECTION_SYNC))) { @@ -379,7 +433,9 @@ function salesforce_push_map_params($mapping, $entity_wrapper, &$key_field, &$ke $key_field = $fieldmap['salesforce_field']['name']; $key_value = $value; // If key is set, remove from $params to avoid UPSERT errors. - unset($params[$key_field]); + if(!$use_soap){ + unset($params[$key_field]); + } } }