Chain: цепочки вызовов, переиспользуемые шаги и конкурентный HTTP

Одинаковая цепочка методов не означает одинаковое выполнение

При повторном использовании цепочки важно знать, что она сохранила. Это может быть уже изменённый объект, определения ещё не выполненных шагов или настройки сетевых запросов. Ошибка тоже имеет разный масштаб: один вызов, оставшаяся последовательность или отдельное задание среди независимых запросов. Эти различия должны быть видны в контракте инструмента.

У трёх утилит Chain 2.0 для PHP 8.5 разные обязанности. Chain оборачивает объект и выполняет вызовы сразу. Flow хранит переиспользуемый план прикладных шагов. Batch управляет независимыми HTTP-запросами через cURL multi. Общий fluent-синтаксис оставляет различия выполнения явными.

Инструменты можно сочетать, но один не требует другого. Batch не запускает запросы через Chain, а Flow не использует HTTP-планировщик для выполнения своих шагов. Ниже — решения, которые задают поведение каждого из них.

Chain: значение изменилось, получатель остался

У обёртки два состояния: объект, на котором вызываются методы, и значение последней операции. Когда метод возвращает число, строку или null, объект-получатель сохраняется. Когда возвращается объект, он становится и новым получателем, и текущим значением. Это позволяет переходить между связанными объектами, не теряя возможность получить скалярный результат.

В примере next() возвращает int. Следующий pipe() получает именно это число, а reset() по-прежнему вызывается на исходном счётчике. Результат void-метода — null; get() не подменяет его объектом.

use tommyknocker\chain\Chain;

$counter = new class {
    private int $value = 0;

    public function next(): int { return ++$this->value; }
    public function reset(): void { $this->value = 0; }
};

$chain = Chain::of($counter);
$value = $chain->next()
    ->pipe(static fn (int $n): int => $n * 10)
    ->get();                         // 10

$receiver = $chain->instance();      // $counter
$result = $chain->reset()->get();    // null

Это различие важно и для callbacks: tap() наблюдает текущий объект, inspect() — текущее значение. Если имя метода объекта совпадает с API обёртки, например get() или map(), явный invoke() вызывает именно метод объекта. Порядок разрешения имён не должен зависеть от догадки читателя.

Chain::clone() создаёт копию обёртки и текущего объекта: по умолчанию поверхностную, с enableDeepCloning — глубокую. Это ветвление уже существующего состояния. Flow::then(), напротив, копирует план шагов; payload передаётся только при запуске.

Плата за такую гибкость — динамические переходы. PHPStan не выводит автоматически тип каждого получателя через всю цепочку. requireInstance(Foo::class) даёт проверенный объект нужного типа после перехода; fluent-синтаксис сам по себе не делает произвольную цепочку статически типизированной.

Flow: план можно расширить, не меняя базовый

Для последовательности прикладных действий важен другой контракт. План нужен отдельно от конкретного запуска: его можно сохранить, добавить шаги для одного сценария и выполнить исходный вариант для другого. Поэтому then() возвращает новый Flow, а результат запуска хранится в FlowExecution.

Шагом может быть StepInterface, имя класса с разрешением через PSR-11 контейнер, callable или вложенный Flow. Именованные классы полезны для прикладных действий с зависимостями; короткие callbacks подходят для локальных преобразований. Небольшой пример ниже показывает только механику плана и остановки.

use tommyknocker\chain\Flow;
use tommyknocker\chain\FlowStop;

$base = Flow::of(static fn (string $s): string => trim($s));
$normalized = $base
    ->then(static fn (string $s): string|FlowStop =>
        $s === '' ? Flow::stop('') : $s)
    ->then(static fn (string $s): string => strtolower($s));

$original = $base->run('  PHP  ');          // 'PHP'
$execution = $normalized->execute(' PHP ');
$result = $execution->unwrap();            // 'php'
$stopped = $normalized->execute('   ');
$didStop = $stopped->stopped();             // true

Flow::stop() — успешное завершение с результатом, а не ошибка. Оно прекращает и вложенный, и внешний Flow. Исключение шага даёт другой исход: execute() возвращает failure, run() выбрасывает FlowFailedException с индексом, именем шага и исходной причиной. Ошибка вложенного шага сохраняет его сведения; полный путь вложенности отдельным стеком шагов не формируется.

Исключения наблюдателей Flow подавляются, как и у Batch: сбой в observer не меняет результат выполнения шагов.

Контекст наследуется вложенными Flow: собственные атрибуты и resolver дочернего плана имеют приоритет. При этом неизменяемость относится к структуре плана. Объекты шагов, захваченные замыканиями значения и payload не клонируются. Для повторного использования нужны шаги без изменяемого состояния либо намеренно разделяемое состояние.

Batch: ограничить активные запросы, а не остановить весь пакет

Для независимых URL последовательный цикл складывает время ожидания сети. Запуск всех запросов сразу создаёт другую проблему: размер входа начинает определять количество соединений и нагрузку на удалённую сторону. Batch держит не больше заданного числа активных transfers и запускает следующие задания по мере освобождения мест.

Неудачная попытка может перейти в отложенный повтор. До назначенного времени она не занимает активное соединение; остальные запросы продолжают выполняться. Если активных transfers нет, планировщик ждёт, а не проверяет время в непрерывном цикле. Управление cURL multi и расписание повторов находятся в одном цикле выполнения.

use tommyknocker\chain\Batch;
use tommyknocker\chain\HttpRetryPolicy;

// $urls: an id => URL map supplied by the caller.
$report = Batch::fetch($urls)
    ->concurrency(8)
    ->maxJobs(500)
    ->maxBodySize(2 * 1024 * 1024)
    ->timeout(10)
    ->retry(HttpRetryPolicy::transient(maxAttempts: 3))
    ->run();

$responses = $report->responses(); // Successful responses, keyed by id.
$failures = $report->failures();   // Final failures, keyed by id.

Конкурентность здесь относится к сетевому вводу-выводу. run() блокирует вызывающий код до завершения или остановки пакета; это не фоновый исполнитель и не параллельный запуск PHP callbacks.

Лимиты отвечают на разные вопросы. concurrency() ограничивает активные transfers, maxJobs() — число входных заданий, maxBodySize() — размер одного ответа в памяти. Завершённые ответы сохраняются в Report, а вход предварительно собирается в список. Поэтому ограничение конкурентности не ограничивает всю память пакета. Для больших тел предназначен download() с записью в файл.

Повтор попытки, провал задания и отмена — разные события

После транспортной ошибки клиент не всегда знает, обработал ли сервер запрос. Безусловный retry может повторить уже выполненное действие. Поэтому стандартная HTTP-политика повторяет только GET и HEAD при транспортных ошибках, timeout и статусах 408, 429, 502, 503, 504. Для остальных методов требуется явное retryUnsafeMethods; одно это разрешение не обеспечивает идемпотентность прикладной операции.

Политика редиректов также зависит от запроса. При наличии заголовка Authorization или basic auth автоматическое следование редиректам по умолчанию отключено. followRedirects(true) разрешает его явно; допустимость переходов для запроса с учётными данными определяет вызывающий код.

onAttemptFailed() сообщает о неудачной попытке, которую предстоит повторить. onFailed() — об окончательном провале задания. Успешный повтор убирает предыдущую ошибку из итогового отчёта. Ошибки observers подавляются: исключение из логирования не должно превратить успешный HTTP-запрос в провал пакета.

По умолчанию одно неуспешное задание не мешает другим. failFast() прекращает дальнейшее планирование после окончательной ошибки, включая валидацию URL и подготовку файла. При ошибке предварительной проверки запросы ещё не запущены. cancel() позволяет остановить оставшуюся работу из прикладного кода или observer; активные transfers освобождаются.

Отмена не откатывает уже совершённые действия на сервере. Отчёт содержит известные результаты и ошибки, но не добавляет искусственную ошибку для каждого незапущенного задания. Отсутствие failures после отмены поэтому не доказывает, что обработан весь вход.

throwIfFailed() выбрасывает исключение с сохранённым отчётом, если есть ошибки. Успешные HTTP-операции и опубликованные файлы остаются успешными. Этот метод позволяет вызывающему коду потребовать отсутствие зарегистрированных ошибок, но не превращает пакет в транзакцию.

Скачанный файл становится доступен после проверки

При записи сразу в целевой путь другой процесс может прочитать неполный файл, а неудачная загрузка — оставить его под окончательным именем. DownloadBatch пишет тело во временный файл в целевом каталоге. Проверка HTTP-статуса и ожидаемого Content-Type происходит до публикации; неуспешная попытка удаляет временный файл.

Здесь нужны две разные файловые операции. С overwrite() готовый файл публикуется через rename(). Без разрешения на замену используется link(): создание целевого имени должно атомарно завершиться ошибкой, если оно уже занято. После успешного создания ссылки временное имя удаляется.

Проверки file_exists() до загрузки недостаточно: другой процесс может создать файл между проверкой и публикацией. Атомарный запрет замены закрывает именно эту гонку. Внутри одного пакета также резервируются целевые пути, чтобы два задания не публиковались под одним именем.

Этот способ требует файловой системы с поддержкой hard links; при невозможности создать ссылку задание завершается ошибкой, без небезопасного перехода к перезаписи. Атомарная публикация одного файла не означает атомарность всего пакета или гарантию сохранения данных при сбое питания.

Проверять последствия, а не только возвращаемое тело

Успешный запрос доказывает мало о поведении планировщика. В тестах важны ситуации на границах: сервер ответил 503, повтор затем удался, observer выбросил исключение, файл появился во время загрузки, подготовка первого задания завершилась ошибкой. Проверяются результат, события и отсутствие нежелательных побочных эффектов.

Регрессия файловой гонки создаёт целевой файл после подготовки sink и проверяет, что его содержимое сохраняется. Проверка повторов различает ошибку попытки и окончательную ошибку. Проверка ожидания фиксирует повторное засыпание, когда активных запросов нет. Для конкурентного выполнения используется отдельный HTTP fixture, способный обслуживать несколько соединений одновременно.

У Flow отдельно проверяются наследование контекста, приоритет собственного resolver, сохранение базового плана, остановка вложенного выполнения и исходная причина ошибки. PHPStan level 8 и исполняемые примеры дополняют эти проверки; они не заменяют проверку семантики выполнения.

Когда эти инструменты оправданы

Chain полезен, когда fluent-обёртка делает работу с объектами понятнее и команда принимает её динамическую семантику. Для нескольких прямых вызовов обычный PHP-код может читаться проще. Flow подходит для повторяющихся последовательностей прикладных шагов, которым нужны композиция, контекст и явный результат выполнения.

Batch закрывает ограниченный набор задач конкурентного HTTP в одном процессе: получить ответы, скачать файлы, повторить допустимые ошибки и разобрать частичный результат. В нём нет сохранения выполнения между перезапусками или распределённого планирования. Если такие требования появляются, нужен другой уровень исполнения.

Три утилиты остаются одним Composer-пакетом: composer require tommyknocker/chain. Поэтому ext-curl требуется при установке, даже если приложение использует только Flow. Пакет интерфейсов psr/container также обязателен, даже когда resolver не задан. Разделение моделей выполнения здесь не означает независимые пакеты. Исходники и тесты ниже закреплены на конкретном снимке реализации.