SegmentDataBuilder::buildData

public function buildData($perPage = null): bool
{
	if (!$this->connectorIterable())
	{
		return true;
	}

	$groupState = $this->getCurrentGroupState();

	if (!$groupState)
	{
		return true;
	}

	if ($this->isBuildingCompleted())
	{
		return true;
	}

	$connector = ConnectorManager::getConnector($this->endpoint);
	$connector->setDataTypeId(null);
	$connector->setCheckAccessRights(false);
	$connector->setFieldValues($this->endpoint['FIELDS']);

	$lastId = $connector->getEntityLimitInfo()['lastId'];

	/** @var AbstractThreadStrategy $threadStrategy */
	$threadStrategy = RuntimeEnv::getGroupThreadContext();

	$threadStrategy->setGroupStateId($groupState['ID']);

	$threadState = $threadStrategy->checkThreads();
	if ($threadState === AbstractThreadStrategy::THREAD_LOCKED)
	{
		return false;
	}

	if ($threadState === AbstractThreadStrategy::THREAD_NEEDED)
	{
		$threadStrategy->fillThreads();
	}

	$threadStrategy->setPerPage(self::PER_PAGE);

	if (
		$threadStrategy->lockThread() === AbstractThreadStrategy::THREAD_UNAVAILABLE
		|| $threadStrategy->isProcessLimited()
	)
	{
		return false;
	}

	$offset = $threadStrategy->getOffset();
	if (!Locker::lock(self::SEGMENT_DATA_LOCK_KEY, $this->groupId))
	{
		return false;
	}

	if ($offset < $lastId)
	{
		$limit = $offset + self::PER_PAGE;

		$this->addToDB(
			$connector->getLimitedData($offset, $limit)
		);

		Locker::unlock(self::SEGMENT_DATA_LOCK_KEY, $this->groupId);
		$threadStrategy->updateStatus(GroupThreadTable::STATUS_NEW);
		return false;
	}
	Locker::unlock(self::SEGMENT_DATA_LOCK_KEY, $this->groupId);

	if ($threadStrategy->getThreadId() < $threadStrategy->lastThreadId())
	{
		$threadStrategy->updateStatus(GroupThreadTable::STATUS_DONE);
		return false;
	}

	$threadStrategy->updateStatus(GroupThreadTable::STATUS_NEW);

	if (!$threadStrategy->finalize())
	{
		return false;
	}

	$this->completeBuilding();

	return true;
}