Adds force_update flag to MappedObject with methods to set the flag and return the flag

Add check on results return in QueueHandler to avoid error of calling $results->records on non-
existant object.
Updates RestClient::apiCall to throw RestExceptions in all error cases
Updates PullBase to detect RestExceptions from RestClient.

Includes test code in PullBase to force exceptions to test new logic. SuspendQueueException may
not be the proper Queue Exception to throw - it seems to delete/‘release’ the problematic item when
it should be kept so it can be processed fully.

modules/salesforce_mapping/src/Entity/MappedObject.php

@@ -71,6 +71,13 @@ class MappedObject extends RevisionableContentEntityBase implements MappedObject
    */
   protected $drupal_entity = NULL;
 
+  /**
+   * Force update of mapped entity Flag
+   *
+   * @var boolean
+   */
+  protected $force_update = FALSE;
+
   /**
    * Overrides ContentEntityBase::__construct().
    */
@@ -522,4 +529,17 @@ class MappedObject extends RevisionableContentEntityBase implements MappedObject
     return defined('REQUEST_TIME') ? REQUEST_TIME : (int) $_SERVER['REQUEST_TIME'];
   }
 
+  /**
+   * Set the force entity update flag to desired state
+   *
+   * @param boolean $state\
+   */
+  public function setForceUpdate(boolean $state) {
+    $this->force_update = $state;
+  }
+
+  public function forceUpdate() {
+    return $this->force_update;
+  }
+
 }

modules/salesforce_pull/src/Plugin/QueueWorker/PullBase.php

@@ -13,11 +13,9 @@ use Drupal\Core\Extension\ModuleHandlerInterface;
 use Drupal\Core\Logger\LoggerChannelFactoryInterface;
 use Drupal\Core\Plugin\ContainerFactoryPluginInterface;
 use Drupal\Core\Queue\QueueWorkerBase;
+use Drupal\Core\Queue\SuspendQueueException;
+use Drupal\Core\State\State;
 use Drupal\Core\Utility\Error;
-use Drupal\salesforce\Exception;
-use Drupal\salesforce\Rest\RestClient;
-use Drupal\salesforce\SObject;
-use Drupal\salesforce\SalesforceEvents;
 use Drupal\salesforce_mapping\Entity\MappedObject;
 use Drupal\salesforce_mapping\Entity\MappedObjectInterface;
 use Drupal\salesforce_mapping\Entity\SalesforceMappingInterface;
@@ -26,10 +24,14 @@ use Drupal\salesforce_mapping\MappingConstants;
 use Drupal\salesforce_mapping\PushParams;
 use Drupal\salesforce_mapping\SalesforceMappingStorage;
 use Drupal\salesforce_mapping\SalesforcePullEvent;
+use Drupal\salesforce\Exception;
+use Drupal\salesforce\Rest\RestClient;
+use Drupal\salesforce\Rest\RestException;
+use Drupal\salesforce\SalesforceEvents;
+use Drupal\salesforce\SObject;
 use Psr\Log\LogLevel;
 use Symfony\Component\DependencyInjection\ContainerInterface;
 use Symfony\Component\EventDispatcher\EventDispatcherInterface;
-use Drupal\Core\State\State;
 
 /**
  * Provides base functionality for the Salesforce Pull Queue Workers.
@@ -136,6 +138,7 @@ abstract class PullBase extends QueueWorkerBase implements ContainerFactoryPlugi
     ]);
     // @TODO one-to-many: this is a blocker for OTM support:
     $mapped_object = current($mapped_object);
+    $mapped_object = NULL;
     if (!empty($mapped_object)) {
       return $this->updateEntity($mapping, $mapped_object, $sf_object);
     }
@@ -196,11 +199,11 @@ abstract class PullBase extends QueueWorkerBase implements ContainerFactoryPlugi
         $this->salesforcePullEvent($mapped_object, MappingConstants::SALESFORCE_MAPPING_SYNC_SF_UPDATE)
       );
 
-      // By default this is FALSE. To force true, set the state in the prepull
-      // event hook above.
+      // By default $mapped_object->forceUpdate() is FALSE. To force true, call
+      // $mapped_object->setForceUpdate() in the prepull event hook above.
       $force_update = $this->state->get('salesforce.pull_force_update', FALSE);
 
-      if ($sf_record_updated > $entity_updated || $force_update) {
+      if ($sf_record_updated > $entity_updated || $mapped_object->forceUpdate()) {
         // Set fields values on the Drupal entity.
         $mapped_object->pull();
         $this->logger->log(
@@ -287,6 +290,8 @@ abstract class PullBase extends QueueWorkerBase implements ContainerFactoryPlugi
             MappingConstants::SALESFORCE_MAPPING_SYNC_DRUPAL_UPDATE
           ])) {
         try {
+          throw new RestException(new \Drupal\salesforce\Rest\RestResponse(new \GuzzleHttp\Psr7\Response(403), 'test'));
+
           $params = new PushParams($mapping, $entity);
           $this->client->objectUpdate(
             $mapping->getSalesforceObjectType(),
@@ -294,7 +299,7 @@ abstract class PullBase extends QueueWorkerBase implements ContainerFactoryPlugi
             $params->getParams()
           );
         }
-        catch(\Exception $e) {
+        catch(RestException $e) {
           $this->logger->log(
             LogLevel::ERROR,
             'Unable to contact Salesforce API, suspending queue'

modules/salesforce_pull/src/QueueHandler.php

@@ -124,14 +124,16 @@ class QueueHandler {
     // Iterate over each field mapping to determine our query parameters.
     foreach ($this->mappings as $mapping) {
       $results = $this->doSfoQuery($mapping);
-      $this->insertIntoQueue($mapping, $results->records());
-      $this->handleLargeRequests($mapping, $results);
-      $this->state->set(
-        'salesforce_pull_last_sync_' . $mapping->getSalesforceObjectType(),
-        // @TODO Replace this with a better implementation when available,
-        // see https://www.drupal.org/node/2820345, https://www.drupal.org/node/2785211
-        $this->request->server->get('REQUEST_TIME')
-      );
+      if ($results) {
+        $this->insertIntoQueue($mapping, $results->records());
+        $this->handleLargeRequests($mapping, $results);
+        $this->state->set(
+          'salesforce_pull_last_sync_' . $mapping->getSalesforceObjectType(),
+          // @TODO Replace this with a better implementation when available,
+          // see https://www.drupal.org/node/2820345, https://www.drupal.org/node/2785211
+          $this->request->server->get('REQUEST_TIME')
+        );
+      }
     }
     return true;
   }

src/Rest/RestClient.php

@@ -9,10 +9,11 @@ use Drupal\Core\Cache\CacheBackendInterface;
 use Drupal\Core\Config\ConfigFactoryInterface;
 use Drupal\Core\State\StateInterface;
 use Drupal\Core\Url;
-use Drupal\salesforce\SFID;
-use Drupal\salesforce\SObject;
+use Drupal\salesforce\Rest\RestException;
 use Drupal\salesforce\SelectQuery;
 use Drupal\salesforce\SelectQueryResult;
+use Drupal\salesforce\SFID;
+use Drupal\salesforce\SObject;
 use GuzzleHttp\ClientInterface;
 use GuzzleHttp\Exception\RequestException;
 use GuzzleHttp\Psr7\Response;
@@ -128,7 +129,7 @@ class RestClient implements RestClientInterface {
 
       // Any exceptions besides 401 get bubbled up.
       if (!$this->response || $this->response->getStatusCode() != 401) {
-        throw $e;
+        throw new RestException($this->response, $e->getMessage());
       }
     }
 
@@ -142,13 +143,13 @@ class RestClient implements RestClientInterface {
       }
       catch (RequestException $e) {
         $this->response = $e->getResponse();
-        throw $e;
+        throw new RestException($this->response, $e->getMessage());
       }
     }
 
     if (empty($this->response)
     || ((int)floor($this->response->getStatusCode() / 100)) != 2) {
-      throw new \Exception('Unknown error occurred during API call');
+      throw new RestException($this->response, 'Unknown error occurred during API call');
     }
 
     if ($returnObject) {