[0.1.2] add store and website dimensions for parallel process

Author: Benjamin Calef

.gitignore

@@ -1,2 +1,3 @@
 .idea
-/vendor
+composer.lock
+/vendor

Model/Dimension/ParallelStoreProcessor.php

@@ -0,0 +1,69 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Zepgram\MultiThreading\Model\Dimension;
+
+use Magento\Store\Api\Data\StoreInterface;
+use Magento\Store\Api\StoreRepositoryInterface;
+use Zepgram\MultiThreading\Model\ItemProvider\ArrayWrapper;
+use Zepgram\MultiThreading\Model\ItemProvider\ArrayWrapperFactory;
+use Zepgram\MultiThreading\Model\Processor\ForkedProcessorRunner;
+
+class ParallelStoreProcessor
+{
+    /** @var ForkedProcessorRunner */
+    private $forkedProcessorRunner;
+
+    /** @var ArrayWrapperFactory */
+    private $arrayWrapperFactory;
+
+    /** @var StoreRepositoryInterface */
+    private $storeRepository;
+
+    public function __construct(
+        ForkedProcessorRunner $forkedProcessorRunner,
+        ArrayWrapperFactory $arrayWrapperFactory,
+        StoreRepositoryInterface $storeRepository
+    ) {
+        $this->forkedProcessorRunner = $forkedProcessorRunner;
+        $this->arrayWrapperFactory = $arrayWrapperFactory;
+        $this->storeRepository = $storeRepository;
+    }
+
+    /**
+     * @param callable $callback
+     * @param int|null $maxChildrenProcess
+     * @param bool $onlyActiveStores
+     * @param bool $withDefaultStore
+     * @return void
+     */
+    public function process(
+        callable $callback,
+        int $maxChildrenProcess = null,
+        bool $onlyActiveStores = true,
+        bool $withDefaultStore = false
+    ): void {
+        $stores = array_filter($this->storeRepository->getList(),
+            function (StoreInterface $store) use ($onlyActiveStores, $withDefaultStore) {
+                if (!$withDefaultStore && (int)$store->getId() === 0) {
+                    return false;
+                }
+                if ($onlyActiveStores && !$store->getIsActive()) {
+                    return false;
+                }
+                return true;
+            });
+
+        /** @var ArrayWrapper $itemProvider */
+        $itemProvider = $this->arrayWrapperFactory->create([
+            'items' => $stores,
+            'pageSize' => 1
+        ]);
+        $storeCount = $itemProvider->getSize();
+        $maxChildrenProcess = ($maxChildrenProcess >= $storeCount || $maxChildrenProcess === null)
+            ? $storeCount : $maxChildrenProcess;
+
+        $this->forkedProcessorRunner->run($itemProvider, $callback, $maxChildrenProcess);
+    }
+}

Model/Dimension/ParallelWebsiteProcessor.php

@@ -0,0 +1,64 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Zepgram\MultiThreading\Model;
+
+use Magento\Store\Api\Data\WebsiteInterface;
+use Magento\Store\Api\WebsiteRepositoryInterface;
+use Zepgram\MultiThreading\Model\ItemProvider\ArrayWrapper;
+use Zepgram\MultiThreading\Model\ItemProvider\ArrayWrapperFactory;
+use Zepgram\MultiThreading\Model\Processor\ForkedProcessorRunner;
+
+class ParallelWebsiteProcessor
+{
+    /** @var ForkedProcessorRunner */
+    private $forkedProcessorRunner;
+
+    /** @var ArrayWrapperFactory */
+    private $arrayWrapperFactory;
+
+    /** @var WebsiteRepositoryInterface */
+    private $websiteRepository;
+
+    public function __construct(
+        ForkedProcessorRunner $forkedProcessorRunner,
+        ArrayWrapperFactory $arrayWrapperFactory,
+        WebsiteRepositoryInterface $websiteRepository
+    ) {
+        $this->forkedProcessorRunner = $forkedProcessorRunner;
+        $this->arrayWrapperFactory = $arrayWrapperFactory;
+        $this->websiteRepository = $websiteRepository;
+    }
+
+    /**
+     * @param callable $callback
+     * @param int|null $maxChildrenProcess
+     * @param bool $withDefaultWebsite
+     * @return void
+     */
+    public function process(
+        callable $callback,
+        int $maxChildrenProcess = null,
+        bool $withDefaultWebsite = false
+    ): void {
+        $websites = array_filter($this->websiteRepository->getList(),
+            function (WebsiteInterface $website) use ($withDefaultWebsite) {
+                if (!$withDefaultWebsite && (int)$website->getId() === 0) {
+                    return false;
+                }
+                return true;
+            });
+
+        /** @var ArrayWrapper $itemProvider */
+        $itemProvider = $this->arrayWrapperFactory->create([
+            'items' => $websites,
+            'pageSize' => 1
+        ]);
+        $websiteCount = $itemProvider->getSize();
+        $maxChildrenProcess = ($maxChildrenProcess >= $websiteCount || $maxChildrenProcess === null)
+            ? $websiteCount : $maxChildrenProcess;
+
+        $this->forkedProcessorRunner->run($itemProvider, $callback, $maxChildrenProcess);
+    }
+}

README.md

@@ -109,7 +109,7 @@ class MyAwesomeClass
         $this->forkedArrayProcessor = $forkedArrayProcessor;
     }
 
-    $array = [1,2,3,4,5];
+    $array = [1,2,3,4,5,...];
     $callback = function ($item) {
         echo $item;
         // do your business logic here
@@ -124,6 +124,63 @@ class MyAwesomeClass
 }
 ```
 
+### ParallelStoreProcessor or ParallelWebsiteProcessor
+
+```php
+use Zepgram\MultiThreading\Model\Dimension\ParallelStoreProcessor;
+use Magento\Catalog\Model\ResourceModel\Product\Collection;
+use Magento\Catalog\Model\ResourceModel\Product\CollectionFactory;
+
+class MyAwesomeClass
+{
+    /** @var ParallelStoreProcessor */
+    private $parallelStoreProcessor;
+    
+    /** @var CollectionFactory */
+    private $collectionFactory;
+    
+    public function __construct(
+        ParallelStoreProcessor $parallelStoreProcessor,
+        CollectionFactory $collectionFactory
+    ) {
+        $this->parallelStoreProcessor = $parallelStoreProcessor;
+        $this->collectionFactory = $collectionFactory;
+    }
+    
+    $array = [1,2,3,4,5,...];
+    $callback = function (StoreInterface $store) {
+        // retrieve data from database foreach stores (do not load the collection !)
+        $collection = $this->collectionFactory->create();
+        $collection->addFieldToFilter('type_id', 'simple')
+            ->addFieldToSelect(['sku', 'description', 'created_at'])
+            ->setStoreId($store->getId())
+            ->addStoreFilter($store->getId())
+            ->distinct(true);
+            
+        // handle pagination system to avoid memory leak
+        $currentPage = 1;
+        $pageSize = 1000;
+        $totalPages = (int)ceil($collection->getSize() / $pageSize);
+        while ($currentPage <= $totalPages) {
+            $collection->setCurPage($currentPage);
+            $collection->setPageSize($pageSize);
+            foreach ($collection->getItems() as $product) {
+                // do your business logic here
+            }
+            $currentPage++;
+        }
+    };
+    
+    // your collection will be processed foreach store by a dedicated child process
+    $this->parallelStoreProcessor->process(
+        $callback,
+        $maxChildrenProcess = null,
+        $onlyActiveStores = true,
+        $withDefaultStore = false
+    );
+}
+```
+
 ### bin/magento thread:processor command
 
 This command allows running a command indefinitely in a dedicated thread using 
@@ -150,6 +207,15 @@ use a similar approach to process a search criteria or a collection. The process
 into several pages, and for each page, a child process is created to run the callback
 function specified by the user on each item of that page.
 
+The `ParallelStoreProcessor` and `ParallelWebsiteProcessor` classes are designed to make it easier 
+to process a list of stores or websites in parallel. To use either of these classes, you'll need to
+provide a callback function that will be called for each store or website in the list. The callback 
+function should take one parameter, which will be a single store or website object.<br>
+Each store or website will be passed to the callback function in a separate process, 
+allowing faster processing times.
+The number of children process cannot exceed the number of stores or websites: for example,
+if you have 10 stores, the maximum number of child processes that can be created in parallel is 10.
+
 #### Here is a breakdown of the parameters:
 - `$collection`/`$searchCriteria`/`$array`: The first parameter is the data source,
   either a `Magento\Framework\Api\SearchCriteriaInterface` for ForkedSearchResultProcessor or

composer.json

@@ -2,7 +2,7 @@
     "name": "zepgram/module-multi-threading",
     "description": "This module is a powerful tool for developers who want to process large data sets in a short amount of time",
     "type": "magento2-module",
-    "version": "0.1.1",
+    "version": "0.1.2",
     "authors": [
         {
             "name": "Benjamin Calef",
@@ -11,6 +11,7 @@
     ],
     "require": {
         "magento/framework": "^101.0.0|^102.0.0|^103.0.0",
+        "magento/module-store": "^101",
         "ext-pcntl": "*"
     },
     "autoload": {
@@ -27,5 +28,10 @@
             "type": "composer",
             "url": "https://repo.magento.com/"
         }
+    },
+    "config": {
+        "allow-plugins": {
+            "magento/composer-dependency-version-audit-plugin": true
+        }
     }
 }