Lines 83.33% 30 / 36
Functions and Methods 60.00% 6 / 10
Classes and Traits 0.00% 0 / 1
Name Lines Functions and Methods CRAP Classes and Traits
ControlClient 83.33% 30 / 36 60.00% 6 / 10 19.50 0.00% 0 / 1
 __construct 100.00% 1 / 1 100.00% 1 / 1 1
 fetchSnapshot 100.00% 1 / 1 100.00% 1 / 1 1
 pauseApp 100.00% 1 / 1 100.00% 1 / 1 1
 resumeApp 100.00% 1 / 1 100.00% 1 / 1 1
 resetApp 100.00% 1 / 1 100.00% 1 / 1 1
 restartWorker 0.00% 0 / 1 0.00% 0 / 1 2
 sendRequest 75.00% 3 / 4 0.00% 0 / 1 2.06
 exchangeFrames 100.00% 17 / 17 100.00% 1 / 1 4
 readFrame 33.33% 1 / 3 0.00% 0 / 1 3.19
 expectWelcome 66.66% 4 / 6 0.00% 0 / 1 4.59
1<?php
2
3declare(strict_types=1);
4
5namespace Stewart\Runtime\Control\Client;
6
7use Amp\ByteStream\BufferedReader;
8use Amp\ByteStream\BufferException;
9use Amp\Cancellation;
10use Amp\CancelledException;
11use Amp\Socket;
12use Stewart\Contracts\App\AppId;
13use Stewart\Contracts\Time\Duration;
14use Stewart\Runtime\Config\ProjectRoot;
15use Stewart\Runtime\Control\Protocol\Codec\FrameCodec;
16use Stewart\Runtime\Control\Protocol\ControlProtocol;
17use Stewart\Runtime\Control\Protocol\Frame\Bye;
18use Stewart\Runtime\Control\Protocol\Frame\ClientFrame;
19use Stewart\Runtime\Control\Protocol\Frame\CommandResult;
20use Stewart\Runtime\Control\Protocol\Frame\Hello;
21use Stewart\Runtime\Control\Protocol\Frame\PauseAppRequest;
22use Stewart\Runtime\Control\Protocol\Frame\Rejected;
23use Stewart\Runtime\Control\Protocol\Frame\RequestFailed;
24use Stewart\Runtime\Control\Protocol\Frame\ResetAppRequest;
25use Stewart\Runtime\Control\Protocol\Frame\RestartWorkerRequest;
26use Stewart\Runtime\Control\Protocol\Frame\ResumeAppRequest;
27use Stewart\Runtime\Control\Protocol\Frame\ServerFrame;
28use Stewart\Runtime\Control\Protocol\Frame\SnapshotFrame;
29use Stewart\Runtime\Control\Protocol\Frame\SnapshotRequest;
30use Stewart\Runtime\Control\Protocol\Frame\Welcome;
31use Stewart\Runtime\Control\Protocol\Status\RuntimeSnapshot;
32use Stewart\Runtime\Exception\ControlError;
33use Stewart\Runtime\Exception\ControlException;
34use Stewart\Support\Time\Deadlines;
35use Throwable;
36
37final readonly class ControlClient
38{
39    private const int LINE_LIMIT = 4 * 1024 * 1024;
40
41    private const string CLIENT_NAME = 'stewart cli';
42
43    public function __construct(
44        private FrameCodec $codec,
45        private ProjectRoot $projectRoot,
46        private Deadlines $deadlines,
47    ) {}
48
49    /** @throws ControlException|Throwable */
50    public function fetchSnapshot(ControlTarget $target, Duration $timeout): RuntimeSnapshot
51    {
52        return $this->sendRequest($target, new SnapshotRequest(), SnapshotFrame::class, $timeout)->snapshot;
53    }
54
55    /** @throws ControlException|Throwable */
56    public function pauseApp(ControlTarget $target, AppId $appId, Duration $timeout): CommandResult
57    {
58        return $this->sendRequest($target, new PauseAppRequest($appId->value), CommandResult::class, $timeout);
59    }
60
61    /** @throws ControlException|Throwable */
62    public function resumeApp(ControlTarget $target, AppId $appId, Duration $timeout): CommandResult
63    {
64        return $this->sendRequest($target, new ResumeAppRequest($appId->value), CommandResult::class, $timeout);
65    }
66
67    /** @throws ControlException|Throwable */
68    public function resetApp(ControlTarget $target, AppId $appId, Duration $timeout): CommandResult
69    {
70        return $this->sendRequest($target, new ResetAppRequest($appId->value), CommandResult::class, $timeout);
71    }
72
73    /** @throws ControlException|Throwable */
74    public function restartWorker(ControlTarget $target, string $workerId, Duration $timeout): CommandResult
75    {
76        return $this->sendRequest($target, new RestartWorkerRequest($workerId), CommandResult::class, $timeout);
77    }
78
79    /**
80     * @template T of ServerFrame
81     * @param class-string<T> $responseClass
82     * @return T
83     * @throws ControlException|Throwable
84     */
85    private function sendRequest(ControlTarget $target, ClientFrame $request, string $responseClass, Duration $timeout): ServerFrame
86    {
87        $deadline = $this->deadlines->timeout($timeout);
88
89        try {
90            return $this->exchangeFrames($target, $request, $responseClass, $deadline);
91        } catch (CancelledException $e) {
92            throw ControlException::createForReason(ControlError::RequestTimedOut, $e, timeout: (string) $timeout);
93        }
94    }
95
96    /**
97     * @template T of ServerFrame
98     * @param class-string<T> $responseClass
99     * @return T
100     * @throws ControlException|Throwable
101     */
102    private function exchangeFrames(ControlTarget $target, ClientFrame $request, string $responseClass, Cancellation $deadline): ServerFrame
103    {
104        $socket = Socket\connect($target->address->resolveSocketAddress($this->projectRoot), null, $deadline);
105
106        try {
107            $socket->write($this->codec->encodeFrame(new Hello($target->token, self::CLIENT_NAME)));
108            $reader = new BufferedReader($socket);
109            $this->expectWelcome($this->readFrame($reader, $deadline));
110            $socket->write($this->codec->encodeFrame($request));
111            $response = $this->readFrame($reader, $deadline);
112
113            if ($response instanceof RequestFailed) {
114                throw ControlException::createForReason(ControlError::RequestFailed, reason: $response->reason, message: $response->message);
115            }
116
117            if (!$response instanceof $responseClass) {
118                throw ControlException::createForReason(ControlError::UnexpectedFrame, frameClass: $response::class, expectedFrameClass: $responseClass);
119            }
120
121            $bye = $this->readFrame($reader, $deadline);
122
123            return $bye instanceof Bye ? $response : throw ControlException::createForReason(
124                ControlError::UnexpectedFrame,
125                frameClass: $bye::class,
126                expectedFrameClass: Bye::class,
127            );
128        } finally {
129            $socket->close();
130        }
131    }
132
133    /** @throws ControlException|Throwable */
134    private function readFrame(BufferedReader $reader, Cancellation $deadline): ServerFrame
135    {
136        try {
137            return $this->codec->decodeServerFrame($reader->readUntil("\n", $deadline, self::LINE_LIMIT));
138        } catch (BufferException) {
139            throw ControlException::createForReason(ControlError::BrokerHungUp);
140        }
141    }
142
143    /** @throws ControlException */
144    private function expectWelcome(object $frame): void
145    {
146        if ($frame instanceof Rejected) {
147            throw ControlException::createForReason(ControlError::ConnectionRejected, reason: $frame->reason);
148        }
149
150        if (!$frame instanceof Welcome) {
151            throw ControlException::createForReason(ControlError::UnexpectedGreeting, frameClass: $frame::class);
152        }
153
154        if ($frame->protocol !== ControlProtocol::VERSION) {
155            throw ControlException::createForReason(ControlError::ProtocolMismatch, brokerProtocol: $frame->protocol, clientProtocol: ControlProtocol::VERSION);
156        }
157    }
158}