Merge branch '5.x' into 6.x

angrybrad · angrybrad · commit c2f8ae268b71 · 2025-03-14T21:02:55.000-07:00
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -2,6 +2,7 @@
 
 ## Unreleased
 
+- Added support for Craft’s batched queue jobs, so that long-running feeds can be processed in batches. ([#1598](https://github.com/craftcms/feed-me/pull/1598))
 - Improved support for importing into Time fields. ([#1595](https://github.com/craftcms/feed-me/pull/1595))
 - Fixed a PHP error that could occur with some plugins that were installed, but Feed Me doesn't support. ([#1596](https://github.com/craftcms/feed-me/pull/1596))
 - Fixed a bug where that could occur that would prevent importing Commerce Variants. ([#1605](https://github.com/craftcms/feed-me/pull/1605))
diff --git a/src/base/DataType.php b/src/base/DataType.php
@@ -4,6 +4,7 @@
 
 use Cake\Utility\Hash;
 use Craft;
+use craft\base\Batchable;
 use craft\base\Component;
 use craft\helpers\UrlHelper;
 
@@ -12,8 +13,16 @@
  * @property-read mixed $name
  * @property-read mixed $class
  */
-abstract class DataType extends Component
+abstract class DataType extends Component implements Batchable
 {
+    // Properties
+    // =========================================================================
+
+    /**
+     * @var array
+     */
+    protected array $feedData = [];
+
     // Public
     // =========================================================================
 
@@ -59,4 +68,30 @@ public function setupPaginationUrl($array, $feed): void
         // Replace the mapping value with the actual URL
         $feed->paginationUrl = $url;
     }
+
+    /**
+     * @inheritdoc
+     */
+    public function getSlice(int $offset, int $limit): iterable
+    {
+        $feedData = $this->feedData;
+
+        if ($offset) {
+            $feedData = array_slice($feedData, $offset);
+        }
+
+        if ($limit) {
+            $feedData = array_slice($feedData, 0, $limit);
+        }
+
+        return $feedData;
+    }
+
+    /**
+     * @inheritdoc
+     */
+    public function count(): int
+    {
+        return count($this->feedData);
+    }
 }
diff --git a/src/console/controllers/FeedsController.php b/src/console/controllers/FeedsController.php
@@ -5,6 +5,7 @@
 use craft\feedme\Plugin;
 use craft\feedme\queue\jobs\FeedImport;
 use craft\helpers\Console;
+use craft\helpers\Queue;
 use yii\console\Controller;
 use yii\console\ExitCode;
 
@@ -108,7 +109,7 @@ protected function queueFeed($feed, $limit = null, $offset = null, bool $continu
         $this->stdout($feed->name, Console::FG_CYAN);
         $this->stdout(' ... ');
 
-        $this->module->queue->push(new FeedImport([
+        Queue::push(new FeedImport([
             'feed' => $feed,
             'limit' => $limit,
             'offset' => $offset,
diff --git a/src/controllers/FeedsController.php b/src/controllers/FeedsController.php
@@ -9,6 +9,7 @@
 use craft\feedme\Plugin;
 use craft\feedme\queue\jobs\FeedImport;
 use craft\helpers\Json;
+use craft\helpers\Queue;
 use craft\helpers\StringHelper;
 use craft\web\Controller;
 use Exception;
@@ -344,7 +345,7 @@ private function _runImportTask($feed): ?bool
             Craft::$app->getSession()->setNotice(Craft::t('feed-me', 'Feed processing started.'));
 
             // Create the import task
-            $this->module->queue->push(new FeedImport([
+            Queue::push(new FeedImport([
                 'feed' => $feed,
                 'limit' => $limit,
                 'offset' => $offset,
@@ -363,7 +364,7 @@ private function _runImportTask($feed): ?bool
 
             // Create the import task only if provided the correct passkey
             if ($proceed) {
-                $this->module->queue->push(new FeedImport([
+                Queue::push(new FeedImport([
                     'feed' => $feed,
                     'limit' => $limit,
                     'offset' => $offset,
diff --git a/src/datatypes/Csv.php b/src/datatypes/Csv.php
@@ -112,6 +112,7 @@ public function getFeed($url, $settings, bool $usePrimaryElement = true): array
             $array = Plugin::$plugin->data->findPrimaryElement($primaryElement, $array);
         }
 
+        $this->feedData = $array;
         return ['success' => true, 'data' => $array];
     }
 
diff --git a/src/datatypes/DataBatcher.php b/src/datatypes/DataBatcher.php
@@ -0,0 +1,40 @@
+<?php
+
+
+namespace craft\feedme\datatypes;
+
+use craft\base\Batchable;
+
+class DataBatcher implements Batchable
+{
+    public function __construct(
+        private array $data,
+    ) {
+    }
+
+    /**
+     * @inheritdoc
+     */
+    public function count(): int
+    {
+        return count($this->data);
+    }
+
+    /**
+     * @inheritdoc
+     */
+    public function getSlice(int $offset, int $limit): iterable
+    {
+        $slice = $this->data;
+
+        if ($offset) {
+            $slice = array_slice($slice, $offset);
+        }
+
+        if ($limit) {
+            $slice = array_slice($slice, 0, $limit);
+        }
+
+        return $slice;
+    }
+}
diff --git a/src/datatypes/GoogleSheet.php b/src/datatypes/GoogleSheet.php
@@ -87,6 +87,7 @@ public function getFeed($url, $settings, bool $usePrimaryElement = true): array
             $array = Plugin::$plugin->data->findPrimaryElement($primaryElement, $array);
         }
 
+        $this->feedData = $array;
         return ['success' => true, 'data' => $array];
     }
 }
diff --git a/src/datatypes/Json.php b/src/datatypes/Json.php
@@ -85,6 +85,7 @@ public function getFeed($url, $settings, bool $usePrimaryElement = true): array
             $array = Plugin::$plugin->data->findPrimaryElement($primaryElement, $array);
         }
 
+        $this->feedData = $array;
         return ['success' => true, 'data' => $array];
     }
 }
diff --git a/src/datatypes/Xml.php b/src/datatypes/Xml.php
@@ -83,6 +83,7 @@ public function getFeed($url, $settings, bool $usePrimaryElement = true): array
             $array = Plugin::$plugin->data->findPrimaryElement($primaryElement, $array);
         }
 
+        $this->feedData = $array;
         return ['success' => true, 'data' => $array];
     }
 }
diff --git a/src/queue/jobs/FeedImport.php b/src/queue/jobs/FeedImport.php
@@ -2,18 +2,24 @@
 
 namespace craft\feedme\queue\jobs;
 
+use Cake\Utility\Hash;
 use Craft;
+use craft\base\Batchable;
+use craft\feedme\datatypes\DataBatcher;
+use craft\feedme\events\FeedProcessEvent;
 use craft\feedme\models\FeedModel;
 use craft\feedme\Plugin;
-use craft\queue\BaseJob;
+use craft\feedme\services\Process;
+use craft\helpers\Queue;
+use craft\queue\BaseBatchedJob;
 use Throwable;
 use yii\queue\RetryableJobInterface;
 
 /**
  *
  * @property-read mixed $ttr
  */
-class FeedImport extends BaseJob implements RetryableJobInterface
+class FeedImport extends BaseBatchedJob implements RetryableJobInterface
 {
     // Properties
     // =========================================================================
@@ -44,6 +50,22 @@ class FeedImport extends BaseJob implements RetryableJobInterface
      */
     public bool $continueOnError = true;
 
+    /**
+     * @var mixed The Unix timestamp with microseconds of when the feed import started being processed
+     * @since 5.11.0
+     */
+    public mixed $startTime = null;
+
+    /**
+     * @var array The Feed's settings as prepared by beforeProcessFeed()
+     */
+    private array $_feedSettings = [];
+
+    /**
+     * @var int The index of currently processed item in current batch
+     */
+    private int $_index = 0;
+
     // Public Methods
     // =========================================================================
 
@@ -65,73 +87,108 @@ public function canRetry($attempt, $error): bool
     }
 
     /**
-     * @inheritDoc
+     * @inheritdoc
      */
-    public function execute($queue): void
+    protected function loadData(): Batchable
     {
-        try {
-            $feedData = $this->feed->getFeedData();
+        $feedData = $this->feed->getFeedData();
 
-            if ($this->offset) {
-                $feedData = array_slice($feedData, $this->offset);
-            }
+        if ($this->offset) {
+            $feedData = array_slice($feedData, $this->offset);
+        }
 
-            if ($this->limit) {
-                $feedData = array_slice($feedData, 0, $this->limit);
-            }
+        if ($this->limit) {
+            $feedData = array_slice($feedData, 0, $this->limit);
+        }
+
+        $data = $feedData;
 
-            // Do we even have any data to process?
-            if (!$feedData) {
-                Plugin::info('No feed items to process.');
-                return;
+        // Our main data-parsing function. Handles the actual data values, defaults and field options
+        foreach ($feedData as $key => $nodeData) {
+            if (!is_array($nodeData)) {
+                $nodeData = [$nodeData];
             }
 
-            $feedSettings = Plugin::$plugin->process->beforeProcessFeed($this->feed, $feedData);
+            $data[$key] = Hash::flatten($nodeData, '/');
+        }
+
+        $data = array_values($data);
 
-            $feedData = $feedSettings['feedData'];
+        // Fire an 'onBeforeProcessFeed' event
+        $event = new FeedProcessEvent([
+            'feed' => $this->feed,
+            'feedData' => $data,
+        ]);
 
-            $totalSteps = count($feedData);
+        Plugin::$plugin->process->trigger(Process::EVENT_BEFORE_PROCESS_FEED, $event);
 
-            $index = 0;
+        if (!$event->isValid) {
+            return new DataBatcher([]);
+        }
 
-            foreach ($feedData as $data) {
-                try {
-                    Plugin::$plugin->process->processFeed($index, $feedSettings, $this->processedElementIds);
-                } catch (Throwable $e) {
-                    if (!$this->continueOnError) {
-                        throw $e;
-                    }
+        // Allow event to modify the feed data
+        $data = $event->feedData;
 
-                    // We want to catch any issues in each iteration of the loop (and log them), but this allows the
-                    // rest of the feed to continue processing.
-                    Plugin::error('`{e} - {f}: {l}`.', ['e' => $e->getMessage(), 'f' => basename($e->getFile()), 'l' => $e->getLine()]);
-                    Craft::$app->getErrorHandler()->logException($e);
-                }
+        return new DataBatcher($data);
+    }
 
-                $this->setProgress($queue, $index++ / $totalSteps);
+    /**
+     * @inheritdoc
+     */
+    protected function processItem(mixed $item): void
+    {
+        try {
+            Plugin::$plugin->process->processFeed($this->_index, $this->_feedSettings, $this->processedElementIds, $item, $this->batchIndex);
+        } catch (Throwable $e) {
+            if (!$this->continueOnError) {
+                throw $e;
             }
 
-            // Check if we need to paginate the feed to run again
+            // We want to catch any issues in each iteration of the loop (and log them), but this allows the
+            // rest of the feed to continue processing.
+            Plugin::error('`{e} - {f}: {l}`.', ['e' => $e->getMessage(), 'f' => basename($e->getFile()), 'l' => $e->getLine()]);
+            Craft::$app->getErrorHandler()->logException($e);
+        }
+
+        $this->_index++;
+    }
+    /**
+     * @inheritDoc
+     */
+    public function execute($queue): void
+    {
+        $processService = Plugin::$plugin->getProcess();
+        if ($this->itemOffset == 0) {
+            $processService->beforeProcessFeed($this->feed, (array)$this->data());
+        }
+
+        if (!$this->startTime) {
+            $this->startTime = $processService->time_start;
+        }
+
+        if (empty($this->_feedSettings)) {
+            $this->_feedSettings = $processService->getFeedSettings($this->feed, (array)$this->data());
+        }
+
+        parent::execute($queue);
+
+        // Check if we need to paginate the feed to run again
+        if ($this->itemOffset == $this->totalItems()) {
             if ($this->feed->getNextPagination()) {
-                Plugin::getInstance()->queue->push(new self([
+                Queue::push(new self([
                     'feed' => $this->feed,
                     'limit' => $this->limit,
                     'offset' => $this->offset,
                     'processedElementIds' => $this->processedElementIds,
+                    'startTime' => $this->startTime,
                 ]));
             } else {
                 // Only perform the afterProcessFeed function after any/all pagination is done
-                Plugin::$plugin->process->afterProcessFeed($feedSettings, $this->feed, $this->processedElementIds);
+                $processService->afterProcessFeed($this->_feedSettings, $this->feed, $this->processedElementIds, $this->startTime);
             }
-        } catch (Throwable $e) {
-            // Even though we catch errors on each step of the loop, make sure to catch errors that can be anywhere
-            // else in this function, just to be super-safe and not cause the queue job to die.
-            Plugin::error('`{e} - {f}: {l}`.', ['e' => $e->getMessage(), 'f' => basename($e->getFile()), 'l' => $e->getLine()]);
-            Craft::$app->getErrorHandler()->logException($e);
         }
     }
 
-
     // Protected Methods
     // =========================================================================
 
diff --git a/src/services/Process.php b/src/services/Process.php

Original file line number	Diff line number	Diff line change
`@@ -112,6 +112,7 @@ public function getFeed($url, $settings, bool $usePrimaryElement = true): array`
`112`	`112`	`$array = Plugin::$plugin->data->findPrimaryElement($primaryElement, $array);`
`113`	`113`	`}`
`114`	`114`
	`115`	`+ $this->feedData = $array;`
`115`	`116`	`return ['success' => true, 'data' => $array];`
`116`	`117`	`}`
`117`	`118`
Original file line number	Diff line number	Diff line change
`@@ -87,6 +87,7 @@ public function getFeed($url, $settings, bool $usePrimaryElement = true): array`
`87`	`87`	`$array = Plugin::$plugin->data->findPrimaryElement($primaryElement, $array);`
`88`	`88`	`}`
`89`	`89`
	`90`	`+ $this->feedData = $array;`
`90`	`91`	`return ['success' => true, 'data' => $array];`
`91`	`92`	`}`
`92`	`93`	`}`
Original file line number	Diff line number	Diff line change
`@@ -85,6 +85,7 @@ public function getFeed($url, $settings, bool $usePrimaryElement = true): array`
`85`	`85`	`$array = Plugin::$plugin->data->findPrimaryElement($primaryElement, $array);`
`86`	`86`	`}`
`87`	`87`
	`88`	`+ $this->feedData = $array;`
`88`	`89`	`return ['success' => true, 'data' => $array];`
`89`	`90`	`}`
`90`	`91`	`}`
Original file line number	Diff line number	Diff line change
`@@ -83,6 +83,7 @@ public function getFeed($url, $settings, bool $usePrimaryElement = true): array`
`83`	`83`	`$array = Plugin::$plugin->data->findPrimaryElement($primaryElement, $array);`
`84`	`84`	`}`
`85`	`85`
	`86`	`+ $this->feedData = $array;`
`86`	`87`	`return ['success' => true, 'data' => $array];`
`87`	`88`	`}`
`88`	`89`	`}`