QueueTable::processQueue

static function processQueue($type)
{
	$hasQueue = false;
	$queryData = array();

	$audience = Service::getAudience($type);
	$maxQuantity = $audience->getMaxContactsPerPacket();
	$maxQuantity = $maxQuantity > 1000 ? 1000 : $maxQuantity;
	$queueDb = static::getList(array(
		'filter' => array(
			'=TYPE' => $type,
			'=ACTION' => array(
				self::ACTION_IMPORT,
				self::ACTION_REMOVE,
				self::ACTION_IMPORT_AND_AUTO_REMOVE,
			)
		),
		'limit' => $maxQuantity
	));
	while ($queueItem = $queueDb->fetch())
	{
		$hasQueue = true;

		$isRemove = $queueItem['ACTION'] == self::ACTION_REMOVE ? 'Y' : 'N';
		$queryId = $queueItem['TYPE'];
		$queryId .= '_' . $queueItem['PARENT_ID'];
		$queryId .= '_' . $queueItem['CLIENT_ID'];
		$queryId .= '_' . $queueItem['ACCOUNT_ID'];
		$queryId .= '_' . $queueItem['AUDIENCE_ID'];
		$queryId .= '_' . $isRemove;

		if (!isset($queryData[$queryId]))
		{
			$queryData[$queryId] = array(
				'CLIENT_ID' => $queueItem['CLIENT_ID'],
				'ACCOUNT_ID' => $queueItem['ACCOUNT_ID'],
				'AUDIENCE_ID' => $queueItem['AUDIENCE_ID'],
				'PARENT_ID' => $queueItem['PARENT_ID'],
				'IS_REMOVE' => $isRemove,
				'CONTACTS' => array(),
				'DELETE_ID_LIST' => array(),
				'AUTO_REMOVE_ID_LIST' => array(),
			);
		}
		$contactType = $queueItem['CONTACT_TYPE'];
		if (!isset($queryData[$queryId]['CONTACTS'][$contactType]))
		{
			$queryData[$queryId]['CONTACTS'][$contactType] = array();
		}

		$queryData[$queryId]['CONTACTS'][$contactType][] = $queueItem['VALUE'];

		if ($queueItem['ACTION'] == self::ACTION_IMPORT_AND_AUTO_REMOVE)
		{
			$queryData[$queryId]['AUTO_REMOVE_ID_LIST'][] = $queueItem['ID'];
		}
		else
		{
			$queryData[$queryId]['DELETE_ID_LIST'][] = $queueItem['ID'];
		}
	}

	$lastClientId = null;
	$service = null;
	$authAdapter = Service::getAuthAdapter($type);
	foreach ($queryData as $queryId => $query)
	{
		foreach ($query['CONTACTS'] as $contactType => $contacts)
		{
			$query['CONTACTS'][$contactType] = array_unique($contacts);
		}

		if ($lastClientId != $query['CLIENT_ID'] || !$service || !$authAdapter)
		{
			$lastClientId = $query['CLIENT_ID'];

			$service = new Service();
			$service->setClientId($lastClientId);
			$authAdapter->setService($service);
		}

		$audience->setService($service);
		$audience->getRequest()->setAuthAdapter($authAdapter);

		$audience->disableQueueMode();
		$audience->setAccountId($query['ACCOUNT_ID']);

		$contactTypes = $audience->isSupportMultiTypeContacts() ? array('') : array_keys($query['CONTACTS']);
		foreach ($contactTypes as $contactType)
		{
			if ($query['IS_REMOVE'] != 'Y')
			{
				$audienceImportResult = $audience->addContacts(
					$query['AUDIENCE_ID'],
					$query['CONTACTS'],
					array(
						'type' => $contactType
					)
				);
			}
			else
			{
				$audienceImportResult = $audience->deleteContacts(
					$query['AUDIENCE_ID'],
					$query['CONTACTS'],
					array(
						'type' => $contactType
					)
				);
			}

			if ($audienceImportResult->isSuccess())
			{
				if (!empty($query['DELETE_ID_LIST']))
				{
					$portions = self::divideListIntoPortions($query['DELETE_ID_LIST']);
					foreach ($portions as $portion)
					{
						Application::getConnection()->query(
							"DELETE FROM " . self::getTableName() . " WHERE ID IN (" . implode(',', $portion) . ")"
						);
					}
				}

				if (!empty($query['AUTO_REMOVE_ID_LIST']))
				{
					$portions = self::divideListIntoPortions($query['AUTO_REMOVE_ID_LIST']);
					foreach ($portions as $portion)
					{
						Application::getConnection()->query(
							"UPDATE " . self::getTableName() . " SET ACTION='" . self::ACTION_AUTO_REMOVE . "'" .
							"WHERE ID IN (" . implode(',', $portion) . ")"
						);
					}
					static::addQueueAutoRemoveAgent();
				}
			}
			else
			{
				Application::getConnection()->query(
					"DELETE FROM " . self::getTableName() .
					" WHERE TYPE = '" . Application::getConnection()->getSqlHelper()->forSql($type) . "'" .
					" AND ACTION in ('" . implode("', '", [self::ACTION_IMPORT, self::ACTION_IMPORT_AND_AUTO_REMOVE, self::ACTION_REMOVE]) . "')" .
					" AND DATE_INSERT < '" . (new DateTime())->add('-1 day')->format("Y-m-d H:i:s") . "'"
				);
			}
		}
	}

	return $hasQueue;
}