Файл IndexAction.php в Cotonti CMF

Этот метод — конструктор класса, который инициализирует необходимые сервисы и репозитории для работы с событиями на сервере. В нем используются паттерны "одиночка" (Singleton) для получения единственного экземпляра каждого сервиса и репозитория.

public function run(): void // Основной метод, выполняющий логику обработки событий
{
    if (Cot::$usr['id'] === 0) { // Проверяем, не является ли пользователь гостем
        throw new NotFoundHttpException(); // Если пользователь не авторизован, генерируем исключение
    }

    $this->clearOldObservers(); // Очищаем старых наблюдателей

    $this->token = $this->observerService->register(Cot::$usr['id']); // Регистрируем нового наблюдателя для текущего пользователя и получаем токен

    // Make session read-only
    session_write_close(); // Закрываем сессию для предотвращения дальнейших изменений

    // Disable the time limit
    set_time_limit(0); // Отключаем ограничение по времени для выполнения скрипта (он будет работать бесконечно)

    // Disable output buffering
    for ($i = 0; $i <= ob_get_level(); $i++) { // Очищаем все уровни буферизации вывода
        ob_end_clean(); // Очистка текущего уровня буфера
    }
    ob_end_flush(); // Завершаем работу с буфером
    ob_implicit_flush(); // Включаем немедленную отправку данных в браузер

    // Set headers for stream
    header('Content-Type: text/event-stream'); // Устанавливаем заголовок для передачи событий в формате Server-Sent Events
    header('Cache-Control: no-cache'); // Отключаем кэширование
    header('Connection: keep-alive'); // Поддерживаем постоянное соединение
    header('Transfer-Encoding: chunked'); // Используем поблочную передачу данных

    while (true) { // Входим в бесконечный цикл для непрерывной отправки событий
        $currentTime = time(); // Получаем текущее время
        if (connection_aborted()) { // Проверяем, было ли разорвано соединение
            $this->observerService->remove(Cot::$usr['id'], $this->token); // Убираем наблюдателя
            break; // Выходим из цикла
        }

        if (
            $currentTime - $this->getGetLastOldObserversClearedTime()
            > ServerEventsDictionary::CLEAR_OLD_OBSERVERS_PERIOD
        ) { // Если прошло достаточно времени с последней очистки старых наблюдателей
            $this->clearOldObservers(); // Очищаем старых наблюдателей
        }

        if (
            $currentTime - $this->getGetLastCheckConnectionTime() > ServerEventsDictionary::CHECK_CONNECTION_PERIOD
            && !$this->isStillConnected()
        ) { // Если прошло достаточно времени с последней проверки соединения, и соединение разорвано
            break; // Прерываем цикл
        }

        if ($currentTime - $this->getLastClearSentEventsRegistryTime() > ServerEventsDictionary::EVENT_EXPIRE_IN) { // Если прошло достаточно времени с последней очистки регистра отправленных событий
            $this->clearSentEventsRegistry(); // Очищаем реестр отправленных событий
        }

        $events = $this->eventsRepository->getForObserver(Cot::$usr['id'], $this->token); // Получаем события для наблюдателя
        if ($events !== []) { // Если события найдены
            $lastId = 0; // Инициализируем переменную для отслеживания последнего ID события
            foreach ($events as $event) { // Для каждого события
                if (isset($this->sentEvents[$event->id])) { // Проверяем, было ли уже отправлено это событие
                    continue; // Если да, пропускаем его
                }

                echo $event; // Отправляем событие
                if (ob_get_level() > 0) { // Если есть уровни буфера
                    ob_flush(); // Принудительно очищаем буфер
                }
                flush(); // Принудительно очищаем вывод

                $this->sentEvents[$event->id] = $currentTime; // Регистрируем отправленное событие
                if ($event->id > $lastId) { // Обновляем последний отправленный ID события
                    $lastId = $event->id;
                }
            }
            $this->observerService->setLastEventId($this->token, $lastId); // Устанавливаем ID последнего отправленного события для текущего наблюдателя
            $this->eventService->deleteByUserId(Cot::$usr['id']); // Удаляем события для пользователя
        }

        sleep($this->interval); // Засыпаем на заданный интервал перед следующей итерацией
    }
}

Этот метод выполняет ключевую логику для обработки событий на сервере в режиме реального времени, используя механизм Server-Sent Events (SSE). Он:

• Регистрирует пользователя как наблюдателя.
• Устанавливает необходимые заголовки для стриминга данных.
• Периодически очищает старых наблюдателей, проверяет соединение с клиентом и очищает реестр отправленных событий.
• Отправляет события клиенту в реальном времени и обновляет состояние на сервере.

Цикл продолжается до тех пор, пока соединение не будет разорвано или пока не будут выполнены все необходимые условия.

/**
 * @return Temporary_cache_driver|Db_cache_driver|null
 */
private function getCache() // Метод для получения объекта кеша
{
    if (!empty(Cot::$cache)) { // Проверяем, что объект кеша не пустой
        return Cot::$cache->mem ?: Cot::$cache->db; // Возвращаем мемори-кеш, если он доступен, иначе возвращаем DB-кеш
    }
    return null; // Если кеш не был найден, возвращаем null
}

private function getGetLastOldObserversClearedTime(): int // Метод для получения времени последней очистки старых наблюдателей
{
    if ($this->lastOldObserversClearedTime === 0) { // Если время последней очистки старых наблюдателей равно 0
        $this->lastOldObserversClearedTime = time(); // Устанавливаем время последней очистки равным текущему времени
    }

    $cache = $this->getCache(); // Получаем объект кеша
    if (!empty($cache)) { // Если кеш существует и не пуст
        $result = $cache->get(self::OLD_OBSERVERS_CHECK_KEY); // Пытаемся получить время последней очистки из кеша
        if (!$result) { // Если в кеше нет данных
            $result = time(); // Устанавливаем время как текущее
        }
        $this->lastOldObserversClearedTime = $result; // Обновляем время последней очистки
    }

    return $this->lastOldObserversClearedTime; // Возвращаем время последней очистки старых наблюдателей
}

private function getGetLastCheckConnectionTime(): int // Метод для получения времени последней проверки соединения
{
    if ($this->lastCheckConnectionTime === 0) { // Если время последней проверки соединения равно 0
        $this->lastCheckConnectionTime = time(); // Устанавливаем время последней проверки равным текущему времени
    }

    $cache = $this->getCache(); // Получаем объект кеша
    if (!empty($cache)) { // Если кеш существует и не пуст
        $result = $cache->get(self::CHECK_CONNECTION_KEY); // Пытаемся получить время последней проверки из кеша
        if (!$result) { // Если в кеше нет данных
            $result = time(); // Устанавливаем время как текущее
        }
        $this->lastCheckConnectionTime = $result; // Обновляем время последней проверки соединения
    }

    return $this->lastCheckConnectionTime; // Возвращаем время последней проверки соединения
}

private function getLastClearSentEventsRegistryTime(): int // Метод для получения времени последней очистки реестра отправленных событий
{
    if ($this->sentEventsRegistryClearTime === 0) { // Если время последней очистки реестра равно 0
        $this->sentEventsRegistryClearTime = time(); // Устанавливаем время последней очистки реестра равным текущему времени
    }
    return $this->sentEventsRegistryClearTime; // Возвращаем время последней очистки реестра отправленных событий
}

private function clearOldObservers(): void // Метод для очистки старых наблюдателей
{
    $this->observerService->clearOld(); // Вызываем метод для очистки старых наблюдателей в сервисе наблюдателей

    $this->lastCheckConnectionTime = time(); // Обновляем время последней проверки соединения
    $cache = $this->getCache(); // Получаем кэш

    if (!empty($cache)) { // Если кэш не пуст
        if (method_exists($cache, 'store_now')) { // Проверяем, существует ли метод 'store_now' у кэша
            $cache->store_now(self::OLD_OBSERVERS_CHECK_KEY, $this->lastCheckConnectionTime); // Сохраняем время в кэш с методом 'store_now'
        } else { // Если метод 'store_now' не существует
            $cache->store(self::OLD_OBSERVERS_CHECK_KEY, $this->lastCheckConnectionTime); // Сохраняем время в кэш обычным методом 'store'
        }
    }
}

private function isStillConnected(): bool // Метод для проверки, остается ли соединение активным
{
    $this->lastOldObserversClearedTime = time(); // Обновляем время последней очистки старых наблюдателей
    $cache = $this->getCache(); // Получаем кэш

    if (!empty($cache)) { // Если кэш не пуст
        if (method_exists($cache, 'store_now')) { // Проверяем, существует ли метод 'store_now' у кэша
            $cache->store_now(self::OLD_OBSERVERS_CHECK_KEY, $this->lastOldObserversClearedTime); // Сохраняем время в кэш с методом 'store_now'
        } else { // Если метод 'store_now' не существует
            $cache->store(self::OLD_OBSERVERS_CHECK_KEY, $this->lastOldObserversClearedTime); // Сохраняем время в кэш обычным методом 'store'
        }
    }

    return $this->observerService->isConnected(Cot::$usr['id'], $this->token); // Проверяем, подключен ли пользователь, и возвращаем результат
}

private function clearSentEventsRegistry(): void // Метод для очистки реестра отправленных событий
{
    $expireTime = time() - ServerEventsDictionary::EVENT_EXPIRE_IN + 30; // Рассчитываем время, после которого события считаются устаревшими
    foreach ($this->sentEvents as $eventId => $eventSentTime) { // Перебираем все отправленные события
        if ($eventSentTime <= $expireTime) { // Если время отправки события меньше или равно времени истечения
            unset($this->sentEvents[$eventId]); // Удаляем событие из реестра отправленных
        }
    }
}