Lines 92.30% 24 / 26
Methods 75.00% 3 / 4
Classes 0.00% 0 / 1
Name Lines Methods CRAP
 __construct 100.00% 1 / 1 100.00% 1 / 1 1
 run 100.00% 7 / 7 100.00% 1 / 1 1
 startApps 100.00% 4 / 4 100.00% 1 / 1 2
 constructAndInitializeApps 85.71% 12 / 14 0.00% 0 / 1 4.05
19final readonly class WorkerSession
20{
21    public function __construct(
22        private WorkerId $workerId,
23        private Transport $transport,
24        private Bootstrap $bootstrap,
25        private AppLifecycle $apps,
26        private PausedAppsSync $pausedApps,
27        private StateCacheSync $stateCacheSync,
28        private PendingRequests $pending,
29        private BrokerMessageReader $reader,
30        private WorkerShutdown $shutdown,
31        private LoopErrorReporter $loopErrors,
32        private LoggerInterface $logger,
33    ) {}
34
35    public function run(): string
36    {
37        $this->loopErrors->install();
38
39        try {
40            $this->shutdown->trackStartup(async($this->startApps(...)));
41
42            async($this->reader->readUntilChannelCloses(...))->ignore();
43
44            $this->shutdown->awaitStopped();
45
46            $abandoned = $this->pending->failAll('the worker is shutting down');
47        } finally {
48            $this->loopErrors->restorePrevious();
49        }
50
51        return \sprintf('worker %d stopped (%s, %d calls abandoned)', $this->workerId->value, $this->shutdown->reason, $abandoned);
52    }
53
54    private function startApps(): void
55    {
56        try {
57            $this->constructAndInitializeApps();
58        } catch (Throwable $e) {
59            $this->logger->error('Worker failed starting its apps', ['exception' => $e]);
60            $this->shutdown->stopWithConfiguredGrace('startup failed');
61        }
62    }
63
64    private function constructAndInitializeApps(): void
65    {
66        $this->pausedApps->applyPausedAppIds($this->bootstrap->pausedAppIds->collection);
67
68        // Constructors may do I/O because the reader runs; initialize() waits for the state cache.
69        $this->apps->constructApps();
70
71        if (!$this->stateCacheSync->awaitFirstSeed($this->shutdown->startupCancellation())) {
72            return;
73        }
74
75        $this->apps->initializeApps();
76
77        if ($this->shutdown->isStopping()) {
78            return;
79        }
80
81        try {
82            $this->transport->send(new WorkerReady(
83                appIds: AppIdsFragment::fromCollection($this->apps->listAppIdsInState(AppState::Running)),
84                failedAppIds: AppIdsFragment::fromCollection($this->apps->listAppIdsInState(AppState::Failed)),
85                memoryBytes: memory_get_usage(true),
86            ));
87        } catch (TransportException $e) {
88            $this->logger->debug('Could not report ready; the broker channel is closed', ['exception' => $e]);
89        }
90    }
91}