Issue #3033925 by AaronBauman: Expose standalone pull mechanism, analogous to standalone push

config/schema/salesforce.schema.yml

@@ -24,8 +24,8 @@ salesforce.settings:
       description: 'Set the maximum number of items which can be enqueued for pull at any given time. Note this setting is not exactly analogous to the push queue limit, since Drupal Cron API does not offer such granularity. Use 0 for no limit.'
     standalone:
       type: boolean
-      label: 'Provide standalone push queue processing endpoint'
-      description: 'Enable standalone push processing, and do not process push mappings during cron. Note: when enabled, you must set up your own service to query this endpoint.'
+      label: 'Provide standalone queue processing endpoint and disable cron processing.'
+      description: 'Enable standalone queue processing, and do not process push mappings during cron. Pull queue will be populated and processed via standalone endpoint, and may also be processed during cron. Note: when enabled, you must set up your own service to query this endpoint.'
     show_all_objects:
       type: boolean
       label: 'Show all Salesforce objects in mapping UI, including system and non-writeable tables'

modules/salesforce_mapping/config/schema/salesforce_mapping.schema.yml

@@ -28,6 +28,9 @@ salesforce_mapping.salesforce_mapping.*:
     push_standalone:
       type: boolean
       label: 'Standalone push queue processing'
+    pull_standalone:
+      type: boolean
+      label: 'Standalone pull queue processing'
     pull_trigger_date:
       type: string
       label: 'Pull Trigger Date Field'

modules/salesforce_mapping/src/Entity/SalesforceMapping.php

@@ -36,6 +36,7 @@ use Drupal\salesforce_mapping\MappingConstants;
  *    "key",
  *    "async",
  *    "push_standalone",
+ *    "pull_standalone",
  *    "pull_trigger_date",
  *    "pull_where_clause",
  *    "sync_triggers",
@@ -115,6 +116,13 @@ class SalesforceMapping extends ConfigEntityBase implements SalesforceMappingInt
    */
   protected $push_standalone = FALSE;
 
+  /**
+   * Whether a standalone push endpoint is enabled for this mapping.
+   *
+   * @var bool
+   */
+  protected $pull_standalone = FALSE;
+
   /**
    * The Salesforce field to use for determining whether or not to pull.
    *
@@ -499,6 +507,13 @@ class SalesforceMapping extends ConfigEntityBase implements SalesforceMappingInt
     return $this->push_standalone;
   }
 
+  /**
+   * {@inheritdoc}
+   */
+  public function doesPullStandalone() {
+    return $this->pull_standalone;
+  }
+
   /**
    * {@inheritdoc}
    */

modules/salesforce_mapping/src/Entity/SalesforceMappingInterface.php

@@ -98,6 +98,15 @@ interface SalesforceMappingInterface extends ConfigEntityInterface, EntityWithPl
    */
   public function doesPushStandalone();
 
+  /**
+   * Getter for push_standalone property.
+   *
+   * @return bool
+   *   TRUE if this mapping is set to process push queue via a standalone
+   *   endpoint instead of during cron.
+   */
+  public function doesPullStandalone();
+
   /**
    * Checks mappings for any push operation.
    *

modules/salesforce_mapping/src/SalesforceMappingStorage.php

@@ -48,16 +48,32 @@ class SalesforceMappingStorage extends ConfigEntityStorage {
   }
 
   /**
-   * Get Mapping entities who are standalone push-enabled.
+   * Get push Mappings to be processed during cron.
    *
    * @return \Drupal\salesforce_mapping\Entity\SalesforceMappingInterface[]
-   *   The push-standalong mappings Mappings.
+   *   The Mappings to process.
    */
   public function loadCronPushMappings() {
+    if ($this->configFactory->get('salesforce.settings')->get('standalone')) {
+      return [];
+    }
     $properties["push_standalone"] = FALSE;
     return $this->loadPushMappingsByProperties($properties);
   }
 
+  /**
+   * Get pull Mappings to be processed during cron.
+   *
+   * @return \Drupal\salesforce_mapping\Entity\SalesforceMappingInterface[]
+   *   The pull Mappings.
+   */
+  public function loadCronPullMappings() {
+    if ($this->configFactory->get('salesforce.settings')->get('standalone')) {
+      return [];
+    }
+    return $this->loadPullMappingsByProperties(["pull_standalone" => FALSE]);
+  }
+
   /**
    * Return an array push-enabled mappings by properties.
    *
@@ -83,6 +99,31 @@ class SalesforceMappingStorage extends ConfigEntityStorage {
     return $push_mappings;
   }
 
+  /**
+   * Return an array push-enabled mappings by properties.
+   *
+   * @param array $properties
+   *   Properties array for storage handler.
+   *
+   * @return \Drupal\salesforce_mapping\Entity\SalesforceMappingInterface[]
+   *   The pull mappings.
+   *
+   * @see ::loadByProperties()
+   */
+  public function loadPullMappingsByProperties(array $properties) {
+    $mappings = $this->loadByProperties($properties);
+    foreach ($mappings as $key => $mapping) {
+      if (!$mapping->doesPull()) {
+        continue;
+      }
+      $push_mappings[$key] = $mapping;
+    }
+    if (empty($push_mappings)) {
+      return [];
+    }
+    return $push_mappings;
+  }
+
   /**
    * Return an array of SalesforceMapping entities who are pull-enabled.
    *

modules/salesforce_mapping_ui/src/Form/SalesforceMappingFormCrudBase.php

@@ -231,6 +231,34 @@ abstract class SalesforceMappingFormCrudBase extends SalesforceMappingFormBase {
         '#default_value' => $mapping->pull_frequency,
         '#description' => t('Enter a frequency, in seconds, for how often this mapping should be used to pull data to Drupal. Enter 0 to pull as often as possible. FYI: 1 hour = 3600; 1 day = 86400. <em>NOTE: pull frequency is shared per-Salesforce Object. The setting is exposed here for convenience.</em>'),
       ];
+
+      $description = t('Check this box to disable cron pull processing for this mapping, and allow standalone processing only. A URL will be generated after saving the mapping.');
+      if ($mapping->id()) {
+        $standalone_url = Url::fromRoute(
+          'salesforce_pull.endpoint.salesforce_mapping',
+          [
+            'salesforce_mapping' => $mapping->id(),
+            'key' => \Drupal::state()->get('system.cron_key'),
+          ],
+          ['absolute' => TRUE])
+          ->toString();
+        $description = t('Check this box to disable cron pull processing for this mapping, and allow standalone processing via this URL: <a href=":url">:url</a>', [':url' => $standalone_url]);
+      }
+      $form['pull']['pull_standalone'] = [
+        '#title' => t('Enable standalone pull queue processing'),
+        '#type' => 'checkbox',
+        '#description' => $description,
+        '#default_value' => $mapping->pull_standalone,
+      ];
+
+      // If global standalone is enabled, then we force this mapping's
+      // standalone property to true.
+      if ($this->config('salesforce.settings')->get('standalone')) {
+        $settings_url = Url::fromRoute('salesforce.global_settings')->toString();
+        $form['pull']['pull_standalone']['#default_value'] = TRUE;
+        $form['pull']['pull_standalone']['#disabled'] = TRUE;
+        $form['pull']['pull_standalone']['#description'] .= ' ' . t('See also <a href="@url">global standalone processing settings</a>.', ['@url' => $settings_url]);
+      }
     }
 
     if ($this->moduleHandler->moduleExists('salesforce_push')) {
@@ -309,7 +337,7 @@ abstract class SalesforceMappingFormCrudBase extends SalesforceMappingFormBase {
       // If global standalone is enabled, then we force this mapping's
       // standalone property to true.
       if ($this->config('salesforce.settings')->get('standalone')) {
-        $settings_url = Url::fromRoute('salesforce.global_settings');
+        $settings_url = Url::fromRoute('salesforce.global_settings')->toString();
         $form['push']['push_standalone']['#default_value'] = TRUE;
         $form['push']['push_standalone']['#disabled'] = TRUE;
         $form['push']['push_standalone']['#description'] .= ' ' . t('See also <a href="@url">global standalone processing settings</a>.', ['@url' => $settings_url]);

modules/salesforce_pull/salesforce_pull.install

@@ -91,3 +91,17 @@ function salesforce_pull_update_8004() {
   \Drupal::state()->set('salesforce.mapping_pull_info', $mapping_pull_info);
   \Drupal::state()->delete('salesforce.sobject_pull_info');
 }
+
+/**
+ * Update mappings with "pull standalone" property.
+ */
+function salesforce_pull_update_8005() {
+  $mappings = \Drupal::entityTypeManager()->getStorage('salesforce_mapping')->loadPullMappings();
+  foreach ($mappings as $mapping) {
+    if (empty($mapping->get('pull_standalone'))) {
+      $mapping
+        ->set('pull_standalone', FALSE)
+        ->save();
+    }
+  }
+}

modules/salesforce_pull/salesforce_pull.module

@@ -9,6 +9,11 @@
  * Implements hook_cron().
  */
 function salesforce_pull_cron() {
+  if (\Drupal::config('salesforce.settings')->get('standalone')) {
+    // If global standalone processing is enabled, stop here.
+    return;
+  }
+
   if (\Drupal::service('plugin.manager.salesforce.auth_providers')->getToken()) {
     \Drupal::service('salesforce_pull.queue_handler')->getUpdatedRecords();
     \Drupal::service('salesforce_pull.delete_handler')->processDeletedRecords();

modules/salesforce_pull/salesforce_pull.routing.yml

+salesforce_pull.endpoint:
+  path: '/salesforce_pull/endpoint/{key}'
+  defaults:
+    _controller: '\Drupal\salesforce_pull\Controller\PullController::endpoint'
+  options:
+    no_cache: TRUE
+  requirements:
+    _access_system_cron: 'TRUE'
+
+salesforce_pull.endpoint.salesforce_mapping:
+  path: '/salesforce_pull/{salesforce_mapping}/endpoint/{key}'
+  defaults:
+    _controller: '\Drupal\salesforce_pull\Controller\PullController::endpoint'
+  options:
+    no_cache: TRUE
+  requirements:
+    _access_system_cron: 'TRUE'
+
+salesforce_pull.endpoint.single_record:
+  path: '/salesforce_pull/{salesforce_mapping}/endpoint/{key}/record/{id}'
+  defaults:
+    _controller: '\Drupal\salesforce_pull\Controller\PullController::endpoint'
+  options:
+    no_cache: TRUE
+  requirements:
+    _access_system_cron: 'TRUE'

modules/salesforce_pull/src/Controller/PullController.php

+<?php
+
+namespace Drupal\salesforce_pull\Controller;
+
+use Drupal\Component\Datetime\Time;
+use Drupal\Core\Config\ConfigFactoryInterface;
+use Drupal\Core\Controller\ControllerBase;
+use Drupal\Core\Entity\EntityTypeManagerInterface;
+use Drupal\Core\Queue\QueueFactory;
+use Drupal\Core\Queue\QueueWorkerManagerInterface;
+use Drupal\Core\Queue\RequeueException;
+use Drupal\Core\Queue\SuspendQueueException;
+use Drupal\Core\State\StateInterface;
+use Drupal\salesforce\Event\SalesforceEvents;
+use Drupal\salesforce\Event\SalesforceNoticeEvent;
+use Drupal\salesforce\SFID;
+use Drupal\salesforce_mapping\Entity\SalesforceMappingInterface;
+use Drupal\salesforce_pull\DeleteHandler;
+use Drupal\salesforce_pull\QueueHandler;
+use Symfony\Component\DependencyInjection\ContainerInterface;
+use Symfony\Component\EventDispatcher\EventDispatcherInterface;
+use Symfony\Component\HttpFoundation\RedirectResponse;
+use Symfony\Component\HttpFoundation\RequestStack;
+use Symfony\Component\HttpFoundation\Response;
+use Symfony\Component\HttpKernel\Exception\AccessDeniedHttpException;
+
+/**
+ * Push controller.
+ */
+class PullController extends ControllerBase {
+
+  const DEFAULT_TIME_LIMIT = 30;
+
+  /**
+   * Pull queue handler service.
+   *
+   * @var \Drupal\salesforce_pull\QueueHandler
+   */
+  protected $queueHandler;
+
+  /**
+   * Pull delete handler service.
+   *
+   * @var \Drupal\salesforce_pull\DeleteHandler
+   */
+  protected $deleteHandler;
+
+  /**
+   * Mapping storage.
+   *
+   * @var \Drupal\Core\Entity\EntityStorageInterface
+   */
+  protected $mappingStorage;
+
+  /**
+   * State.
+   *
+   * @var \Drupal\Core\State\StateInterface
+   */
+  protected $state;
+
+  /**
+   * Queue factory service.
+   *
+   * @var \Drupal\Core\Queue\QueueFactory
+   */
+  protected $queueService;
+
+  /**
+   * Queue worker manager.
+   *
+   * @var \Drupal\Core\Queue\QueueWorkerManagerInterface
+   */
+  protected $queueWorkerManager;
+
+  /**
+   * Event dispatcher.
+   *
+   * @var \Symfony\Component\EventDispatcher\EventDispatcherInterface
+   */
+  protected $eventDispatcher;
+
+  /**
+   * Time.
+   *
+   * @var \Drupal\Component\Datetime\Time
+   */
+  protected $time;
+
+  /**
+   * Current Request.
+   *
+   * @var \Symfony\Component\HttpFoundation\Request
+   */
+  protected $request;
+
+  /**
+   * PushController constructor.
+   *
+   * @throws \Drupal\Component\Plugin\Exception\InvalidPluginDefinitionException
+   * @throws \Drupal\Component\Plugin\Exception\PluginNotFoundException
+   */
+  public function __construct(QueueHandler $queueHandler, DeleteHandler $deleteHandler, EntityTypeManagerInterface $etm, ConfigFactoryInterface $config, StateInterface $state, QueueFactory $queueService, QueueWorkerManagerInterface $queueWorkerManager, EventDispatcherInterface $eventDispatcher, Time $time, RequestStack $requestStack) {
+    $this->queueHandler = $queueHandler;
+    $this->deleteHandler = $deleteHandler;
+    $this->mappingStorage = $etm->getStorage('salesforce_mapping');
+    $this->config = $config;
+    $this->state  = $state;
+    $this->queueService = $queueService;
+    $this->queueWorkerManager = $queueWorkerManager;
+    $this->eventDispatcher = $eventDispatcher;
+    $this->time = $time;
+    $this->request = $requestStack->getCurrentRequest();
+  }
+
+
+  /**
+   * {@inheritdoc}
+   */
+  public static function create(ContainerInterface $container) {
+    return new static(
+      $container->get('salesforce_pull.queue_handler'),
+      $container->get('salesforce_pull.delete_handler'),
+      $container->get('entity_type.manager'),
+      $container->get('config.factory'),
+      $container->get('state'),
+      $container->get('queue'),
+      $container->get('plugin.manager.queue_worker'),
+      $container->get('event_dispatcher'),
+      $container->get('datetime.time'),
+      $container->get('request_stack')
+    );
+  }
+
+  /**
+   * Page callback to process push queue for a given mapping.
+   */
+  public function endpoint(SalesforceMappingInterface $salesforce_mapping = NULL, $key = NULL, $id = NULL) {
+    // If standalone for this mapping is disabled, and global standalone is
+    // disabled, then "Access Denied" for this mapping.
+    if ($key != $this->state->get('system.cron_key')) {
+      throw new AccessDeniedHttpException();
+    }
+    $global_standalone = $this->config('salesforce.settings')->get('standalone');
+    if (!$salesforce_mapping && !$global_standalone) {
+      throw new AccessDeniedHttpException();
+    }
+    if ($salesforce_mapping && !$salesforce_mapping->doesPullStandalone() && !$global_standalone) {
+      throw new AccessDeniedHttpException();
+    }
+    if ($id) {
+      try {
+        $id = new SFID($id);
+      }
+      catch (\Exception $e) {
+        throw new AccessDeniedHttpException();
+      }
+    }
+    $this->populateQueue($salesforce_mapping, $id);
+    $this->processQueue();
+    if ($this->request->get('destination')) {
+      return new RedirectResponse($this->request->get('destination'));
+    }
+    return new Response('', 204);
+  }
+
+  protected function populateQueue(SalesforceMappingInterface $mapping = NULL, SFID $id = NULL) {
+    $mappings = [];
+    if ($id) {
+      return $this->queueHandler->getSingleUpdatedRecord($mapping, $id, TRUE);
+    }
+
+    if ($mapping != NULL) {
+      $mappings[] = $mapping;
+    }
+    else {
+      $mappings = $this->mappingStorage->loadByProperties([["pull_standalone" => TRUE]]);
+    }
+
+    foreach ($mappings as $mapping) {
+      $this->queueHandler->getUpdatedRecordsForMapping($mapping);
+    }
+  }
+
+  protected function getTimeLimit() {
+    return self::DEFAULT_TIME_LIMIT;
+  }
+
+  protected function processQueue() {
+    $start = microtime(true);
+    $worker = $this->queueWorkerManager->createInstance(QueueHandler::PULL_QUEUE_NAME);
+    $end = time() + $this->getTimeLimit();
+    $queue = $this->queueService->get(QueueHandler::PULL_QUEUE_NAME);
+    $count = 0;
+    while ((!$this->getTimeLimit() || time() < $end) && ($item = $queue->claimItem())) {
+      try {
+        $this->eventDispatcher->dispatch(SalesforceEvents::NOTICE, new SalesforceNoticeEvent(NULL, 'Processing item @id from @name queue.', ['@name' => QueueHandler::PULL_QUEUE_NAME, '@id' => $item->item_id]));
+        $worker->processItem($item->data);
+        $queue->deleteItem($item);
+        $count++;
+      }
+      catch (RequeueException $e) {
+        // The worker requested the task to be immediately requeued.
+        $queue->releaseItem($item);
+      }
+      catch (SuspendQueueException $e) {
+        // If the worker indicates there is a problem with the whole queue,
+        // release the item.
+        $queue->releaseItem($item);
+        throw new \Exception($e->getMessage());
+      }
+    }
+    $elapsed = microtime(true) - $start;
+    $this->eventDispatcher->dispatch(SalesforceEvents::NOTICE, new SalesforceNoticeEvent(NULL, 'Processed @count items from the @name queue in @elapsed sec.', ['@count' => $count, '@name' => QueueHandler::PULL_QUEUE_NAME, '@elapsed' => round($elapsed, 2)]));
+  }
+}

modules/salesforce_pull/src/QueueHandler.php

@@ -10,6 +10,7 @@ use Drupal\salesforce\Event\SalesforceErrorEvent;
 use Drupal\salesforce\Event\SalesforceEvents;
 use Drupal\salesforce\Event\SalesforceNoticeEvent;
 use Drupal\salesforce\Rest\RestClientInterface;
+use Drupal\salesforce\SFID;
 use Drupal\salesforce\SObject;
 use Drupal\salesforce\SelectQueryResult;
 use Drupal\salesforce_mapping\Entity\SalesforceMappingInterface;
@@ -25,7 +26,7 @@ use Drupal\Component\Datetime\TimeInterface;
 class QueueHandler {
 
   const PULL_MAX_QUEUE_SIZE = 100000;
-
+  const PULL_QUEUE_NAME = 'cron_salesforce_pull';
   /**
    * Salesforce client.
    *
@@ -89,13 +90,13 @@ class QueueHandler {
    */
   public function __construct(RestClientInterface $sfapi, EntityTypeManagerInterface $entity_type_manager, QueueDatabaseFactory $queue_factory, ConfigFactoryInterface $config, EventDispatcherInterface $event_dispatcher, TimeInterface $time) {
     $this->sfapi = $sfapi;
-    $this->queue = $queue_factory->get('cron_salesforce_pull');
+    $this->queue = $queue_factory->get(self::PULL_QUEUE_NAME);
     $this->config = $config->get('salesforce.settings');
     $this->eventDispatcher = $event_dispatcher;
     $this->time = $time;
     $this->mappings = $entity_type_manager
       ->getStorage('salesforce_mapping')
-      ->loadPullMappings();
+      ->loadCronPullMappings();
   }
 
   /**
@@ -177,6 +178,32 @@ class QueueHandler {
     }
   }
 
+  /**
+   * Given a single mapping/id pair, enqueue it.
+   *
+   * @param \Drupal\salesforce_mapping\Entity\SalesforceMappingInterface $mapping
+   *   The mapping.
+   * @param \Drupal\salesforce\SFID $id
+   *   The record id.
+   * @param bool $force_pull
+   *   Whether to force a pull. TRUE by default.
+   *
+   * @return bool
+   *   TRUE if the record was enqueued successfully. Otherwise FALSE.
+   */
+  public function getSingleUpdatedRecord(SalesforceMappingInterface $mapping, SFID $id, $force_pull = TRUE) {
+    if (!$mapping->doesPull()) {
+      return FALSE;
+    }
+    $record = $this->sfapi->objectRead($mapping->getSalesforceObjectType(), (string)$id);
+    if ($record) {
+      $results = SelectQueryResult::createSingle($record);
+      $this->enqueueAllResults($mapping, $results, $force_pull);
+      return TRUE;
+    }
+    return FALSE;
+  }
+
   /**
    * Perform the SFO Query for a mapping and its mapped fields.
    *

modules/salesforce_pull/tests/src/Unit/QueueHandlerTest.php

@@ -87,7 +87,7 @@ class QueueHandlerTest extends UnitTestCase {
 
     // Mock mapping ConfigEntityStorage object.
     $prophecy = $this->prophesize(SalesforceMappingStorage::CLASS);
-    $prophecy->loadPullMappings(Argument::any())->willReturn([$this->mapping]);
+    $prophecy->loadCronPullMappings(Argument::any())->willReturn([$this->mapping]);
     $this->mappingStorage = $prophecy->reveal();
 
     // Mock EntityTypeManagerInterface.

src/Form/SettingsForm.php

@@ -151,7 +151,7 @@ class SettingsForm extends ConfigFormBase {
       ];
     }
 
-    if (\Drupal::moduleHandler()->moduleExists('salesforce_push')) {
+    if (\Drupal::moduleHandler()->moduleExists('salesforce_push') || \Drupal::moduleHandler()->moduleExists('salesforce_pull')) {
       $form['standalone'] = [
         '#title' => $this->t($definition['standalone']['label']),
         '#description' => $this->t($definition['standalone']['description']),
@@ -159,20 +159,38 @@ class SettingsForm extends ConfigFormBase {
         '#default_value' => $config->get('standalone'),
       ];
 
-      $standalone_url = Url::fromRoute(
+      if (\Drupal::moduleHandler()->moduleExists('salesforce_push')) {
+        $standalone_push_url = Url::fromRoute(
           'salesforce_push.endpoint',
           ['key' => \Drupal::state()->get('system.cron_key')],
           ['absolute' => TRUE]);
-      $form['standalone_url'] = [
-        '#type' => 'item',
-        '#title' => $this->t('Standalone URL'),
-        '#markup' => $this->t('<a href="@url">@url</a>', ['@url' => $standalone_url->toString()]),
-        '#states' => [
-          'visible' => [
-            ':input#edit-standalone' => ['checked' => TRUE],
+        $form['standalone_push_url'] = [
+          '#type' => 'item',
+          '#title' => $this->t('Standalone Push URL'),
+          '#markup' => $this->t('<a href="@url">@url</a>', ['@url' => $standalone_push_url->toString()]),
+          '#states' => [
+            'visible' => [
+              ':input#edit-standalone' => ['checked' => TRUE],
+            ],
           ],
-        ],
-      ];
+        ];
+      }
+      if (\Drupal::moduleHandler()->moduleExists('salesforce_pull')) {
+        $standalone_pull_url = Url::fromRoute(
+          'salesforce_pull.endpoint',
+          ['key' => \Drupal::state()->get('system.cron_key')],
+          ['absolute' => TRUE]);
+        $form['standalone_pull_url'] = [
+          '#type' => 'item',
+          '#title' => $this->t('Standalone Pull URL'),
+          '#markup' => $this->t('<a href="@url">@url</a>', ['@url' => $standalone_pull_url->toString()]),
+          '#states' => [
+            'visible' => [
+              ':input#edit-standalone' => ['checked' => TRUE],
+            ],
+          ],
+        ];
+      }
     }
 
     $form = parent::buildForm($form, $form_state);

src/SelectQueryResult.php

@@ -34,6 +34,22 @@ class SelectQueryResult {
     }
   }
 
+  /**
+   * Create a SelectQueryResult from a single SObject record.
+   *
+   * @param \Drupal\salesforce\SObject $record
+   */
+  public static function createSingle(SObject $record) {
+    $results = [
+      'totalSize' => 1,
+      'done' => TRUE,
+      'records' => []
+    ];
+    $result = new static($results);
+    $result->records[(string)$record->id()] = $record;
+    return $result;
+  }
+
   /**
    * Getter.
    *