helper = $helper; $this->logger = $logger; $this->serializer = $serializer; $this->netSuiteAPI = $netSuiteAPI; } /** * Fetch data from NetSuite and add it to the queue. * * @return array */ public function fetchAndEnqueue($fetchAll = false) { $this->logger->info("[$this->name" . "QueueManager] fetchAndEnqueue(" . ($fetchAll ? "true" : "false") . ") started."); // max page to fetch and loop to fetch all pages $maxPage = 1; $page = 1; $pageSize = $this->getPageSize(); $importSize = 0; // Fetch data from NetSuite while ($maxPage >= $page) { $result = $this->fetch($pageSize, $page); if (isset($result['error'])) { $this->logger->error($result['message']); return [ 'error' => true, 'message' => $result['message'] ]; } $data = $this->convertToMagentoFormat($result['data']); try { $this->enqueue($data); } catch (\Exception $e) { $this->logger->error("[$this->name" . "QueueManager] " . $e->getMessage()); throw $e; } $importSize += count($data); if ($fetchAll) { $maxPage = $result['totalPages']; } $page++; sleep(30); } $this->logger->info("[$this->name" . "QueueManager] fetchAndEnqueue(" . ($fetchAll ? "true" : "false") . ") completes with " . $maxPage . " page(s), pageSize = " . $pageSize . ". " . $this->name); return [ 'error' => false, 'message' => 'Imported ' . $importSize . ' ' . $this->name . ' records.' ]; } /** * Fetch data from NetSuite * * @param int $pageSize * @param int $page * @return array */ abstract public function fetch($pageSize = 50, $page = 1); /** * get the page size for each fetch. * * @return int */ abstract public function getPageSize(); /** * Convert NetSuite data to Magento format * * @param array $data * @return array */ abstract public function convertToMagentoFormat($data); /** * adds data to the queue. * * @param array $data * @return void */ abstract public function enqueue($data); /** * Process batch of queue items * * @param int $pageSize * @param int $page * @return */ public function processBatch($batchSize = 50) { $this->logger->info("[$this->name" . "QueueManager] processBatch: batch size -> " . $batchSize); $queueItems = $this->dequeue($batchSize); $this->logger->info("[$this->name" . "QueueManager] processBatch: queue item size -> " . count($queueItems)); $totalProcessed = 0; if (count($queueItems) == 0) { return $totalProcessed; } $memoryStart = memory_get_usage(); $startTime = microtime(true); $queueKey = $this->helper->nDigitRandom(5); $this->logger->info("# " . $queueKey . " | (Starting " . $this->name . " Queue) | Batch : " . $batchSize); foreach ($queueItems as $item) { $i = 50; while ($i-- && $item->getStatus() == 'pending') { // if ($item->getStatus() != 'pending') { // continue; // we do this to make sure this isn't being consumed by another parallel cron processes // } $error = false; $status = 'success'; $errorMessage = ''; try { $item->setStatus('processing')->save(); $result = $this->processQueueItem($item); sleep(1); $totalProcessed++; } catch (\Throwable $e) { $error = true; $errorMessage = $e->getMessage(); $this->logger->error('------------------------' . $this->name); $this->logger->error('Queue ID# ' . $item->getId()); $this->logger->error($e->getMessage()); $status = 'error'; if ($e->getMessage() == 'The requested qty is not available') { $status = 'pending'; } else if ($e->getMessage() == 'The requested sku is not assigned to given stock') { $status = 'pending'; } else if (str_starts_with($e->getMessage(), 'No such entity with email')){ $status = 'pending'; } else if (str_starts_with($e->getMessage(), 'SQLSTATE[40001]: Serialization failure: 1213 Deadlock found when trying to get lock;')){ $status = 'pending'; } else if ($e->getMessage() == 'Product that you are trying to add is not available.') { $status = 'pending'; } if ($status == 'error' && ($e->getMessage() != 'Email is required.')) { $this->logger->error($e->getTraceAsString()); } } try { if ($error) { $item->setMessages($errorMessage); } else if (isset($result['data'])) { $item->setMessages(($result['is_new'] ?? false ? '' : 'Existing ') . 'ID: ' . $result['data']->getId() ?? ''); } $item->setStatus($status); $item->save(); } catch (\Throwable $e) { $this->logger->error('Failed update queue item status: ' . $e->getTraceAsString()); } } } $memoryEnd = memory_get_usage(); $endTime = microtime(true); $timeSpent = $this->helper->secondsToTime($endTime - $startTime); $memoryUsed = $this->helper->formatBytes($memoryEnd - $memoryStart); $this->logger->info("[$this->name" . "QueueManager] de-queue, # " . $queueKey . " | (Ending " . $this->name . " Queue) | Batched: " . $batchSize . " | # Processed: " . $totalProcessed . " | Memory Used: " . $memoryUsed . ' | Time: ' . $timeSpent); return $totalProcessed; } /** * Get next queue items to process * * @param int $batchSize * @return array */ abstract public function dequeue($batchSize = 50); /** * Process queue item * * @param $queueItem * @return mixed */ abstract public function processQueueItem($queueItem); }