SegmentDataBuilder::updateCounters

private function updateCounters(array $rowsDataCounter)
{
	if (!Locker::lock(self::SEGMENT_LOCK_KEY, $this->groupId))
	{
		return;
	}

	$counter = GroupCounterTable::getList([
		'select' => [
			'GROUP_ID', 'TYPE_ID', 'CNT'
		],
		'filter' => [
			'=GROUP_ID' => $this->groupId
		],
	]);

	while ($item = $counter->fetch())
	{
		if (!isset($rowsDataCounter[$item['GROUP_ID']][$item['TYPE_ID']]))
		{
			$rowsDataCounter[$item['GROUP_ID']][$item['TYPE_ID']] = $item['CNT'];
		}

		$rowsDataCounter[$item['GROUP_ID']][$item['TYPE_ID']] += $item['CNT'];
	}

	GroupCounterTable::deleteByGroupId($this->groupId);
	foreach ($rowsDataCounter as $groupId => $dataCounter)
	{
		foreach ($dataCounter as $typeId => $count)
		{
			GroupCounterTable::add(array(
				'GROUP_ID' => $groupId,
				'TYPE_ID' => $typeId,
				'CNT' => $count,
			));
		}
	}
	Locker::unlock(self::SEGMENT_LOCK_KEY, $this->groupId);

	$groupState = $this->getCurrentGroupState();
	if (!$groupState)
	{
		return;
	}

	$counter = new ConnectorDataCounter($rowsDataCounter);

	if (CModule::IncludeModule('pull'))
	{
		CPullWatch::AddToStack(
			self::FILTER_COUNTER_TAG,
			[
				'module_id' => 'sender',
				'command' => 'updateFilterCounter',
				'params' => [
					'groupId' => $this->groupId,
					'filterId' => $this->filterId,
					'count' => $counter->getArray(),
					'state' => $groupState['STATE'],
					'completed' => (int)$groupState['STATE'] === GroupStateTable::STATES['COMPLETED']
				],
			]
		);
	}
}