_init(CustomerQueue::class, CustomerQueueResource::class); } /** * Get next items in the queue based on batch size * * @param int $type_id * @param int $batchSize * @return \Magento\Framework\Data\Collection\AbstractDb */ public function getNext($batchSize = null) { $collection = $this ->addFieldToSelect('*') ->addFieldToFilter('status', ['eq' => 'pending']) ->setOrder('id', 'asc'); if ($batchSize) { $collection->getSelect()->limit($batchSize); } return $collection; } /** * Insert or update based on UNIQUE combo 'key-processing' * @param $key * @param $value */ public function insertOrUpdate($item) { $connection = $this->getConnection(); try { // Start a transaction // $connection->beginTransaction(); // echo json_encode($item, JSON_PRETTY_PRINT) . "\n"; // update the existing item with same netsuite_customer_id $select = $connection->select() ->from($this->getMainTable()) ->where('status = ?', 'pending') ->where('netsuite_customer_id = ?', $item['netsuite_customer_id']); $existingItem = $connection->fetchRow($select); if (!$existingItem) { $existingItem = []; } else { $existingItem['data'] = json_decode($existingItem['data'], true); } $data = [ 'netsuite_customer_id' => $item['netsuite_customer_id'], 'netsuite_entity_id' => $item['netsuite_entity_id'], 'data' => json_encode($item), 'messages' => '', // 'created_at' => (new \DateTime())->format(\Magento\Framework\Stdlib\DateTime::DATETIME_PHP_FORMAT), 'status' => 'pending', ]; // echo json_encode($data, JSON_PRETTY_PRINT) . "\n"; // update the existing item with default shipping or billing if (isset($existingItem['id'])) { $data['id'] = $existingItem['id']; $item['isdefaultshipping'] = $item['isdefaultshipping'] || ($existingItem['data']['isdefaultshipping'] ?? false); $item['isdefaultbilling'] = $item['isdefaultbilling'] || ($existingItem['data']['isdefaultbilling'] ?? false); $item['shipping_address'] = $item['shipping_address'] ?? $existingItem['data']['shipping_address'] ?? []; $item['billing_address'] = $item['billing_address'] ?? $existingItem['data']['billing_address'] ?? []; $data['data'] = json_encode($item); } // Insert new item or update existing one $connection->insertOnDuplicate($this->getMainTable(), $data, ['data', 'status', 'messages']); // Commit the transaction // $connection->commit(); } catch (\Exception $e) { // Rollback the transaction in case of an error $connection->rollBack(); throw new \Magento\Framework\Exception\LocalizedException( __('Error in insertOrUpdate: %1', $e->getMessage()), $e ); } } }