Skip to content

Commit 00579d0

Browse files
kumykumy
committed
refactor: introduce a base worker class
1 parent 066db22 commit 00579d0

3 files changed

Lines changed: 228 additions & 319 deletions

File tree

‎website/workers/WorkerBase.php‎

Lines changed: 195 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,195 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
use PhpAmqpLib\Connection\AMQPStreamConnection;
6+
use PhpAmqpLib\Exception\AMQPConnectionException;
7+
use PhpAmqpLib\Message\AMQPMessage;
8+
9+
abstract class WorkerBase {
10+
protected $connection;
11+
protected $channel;
12+
protected int $retryDelay = 1;
13+
protected int $maxRetryDelay = 60;
14+
protected int $logLevel;
15+
16+
public const LOG_DEBUG = 100;
17+
public const LOG_INFO = 200;
18+
public const LOG_WARNING = 300;
19+
public const LOG_ERROR = 400;
20+
public const LOG_CRITICAL = 500;
21+
22+
public function __construct() {
23+
$this->logLevel = $this->getLogLevelFromEnv(getenv('GK_NOTIFICATION_LOG_LEVEL') ?: 'INFO');
24+
}
25+
26+
abstract protected function getWorkerName(): string;
27+
28+
abstract protected function processMessage(AMQPMessage $msg): void;
29+
30+
abstract protected function getQueueName(): string;
31+
32+
protected function getQueueArguments(): array {
33+
return [];
34+
}
35+
36+
protected function getPrefetchCount(): int {
37+
return 1;
38+
}
39+
40+
protected function getWaitTimeout(): ?int {
41+
return null;
42+
}
43+
44+
protected function getLogLevelFromEnv(string $level): int {
45+
$levels = [
46+
'DEBUG' => self::LOG_DEBUG,
47+
'INFO' => self::LOG_INFO,
48+
'WARNING' => self::LOG_WARNING,
49+
'ERROR' => self::LOG_ERROR,
50+
'CRITICAL' => self::LOG_CRITICAL,
51+
];
52+
53+
return $levels[strtoupper($level)] ?? self::LOG_INFO;
54+
}
55+
56+
protected function log(int $level, string $message, array $context = []): void {
57+
if ($level < $this->logLevel) {
58+
return;
59+
}
60+
61+
$levelNames = [
62+
self::LOG_DEBUG => 'DEBUG',
63+
self::LOG_INFO => 'INFO',
64+
self::LOG_WARNING => 'WARNING',
65+
self::LOG_ERROR => 'ERROR',
66+
self::LOG_CRITICAL => 'CRITICAL',
67+
];
68+
69+
$timestamp = date('Y-m-d H:i:s');
70+
$levelName = $levelNames[$level] ?? 'UNKNOWN';
71+
$contextStr = !empty($context) ? ' '.json_encode($context) : '';
72+
73+
echo "[{$timestamp}] [{$levelName}] {$message}{$contextStr}\n";
74+
flush();
75+
}
76+
77+
protected function connect(): void {
78+
if (!GK_RABBITMQ_HOST || !GK_RABBITMQ_PORT) {
79+
throw new RuntimeException('RabbitMQ configuration not found. Check GK_RABBITMQ_* environment variables.');
80+
}
81+
82+
$this->connection = new AMQPStreamConnection(
83+
GK_RABBITMQ_HOST,
84+
GK_RABBITMQ_PORT,
85+
GK_RABBITMQ_USER,
86+
GK_RABBITMQ_PASS,
87+
GK_RABBITMQ_VHOST ?: '/'
88+
);
89+
90+
$this->channel = $this->connection->channel();
91+
92+
// Declare exchange (idempotent - won't recreate if exists)
93+
$this->channel->exchange_declare('geokrety', 'fanout', false, true, false);
94+
95+
$queueName = $this->getQueueName();
96+
$this->channel->queue_declare(
97+
$queueName,
98+
false,
99+
true,
100+
false,
101+
false,
102+
false,
103+
$this->getQueueArguments()
104+
);
105+
$this->channel->queue_bind($queueName, 'geokrety');
106+
107+
$this->channel->basic_qos(0, $this->getPrefetchCount(), false);
108+
$this->channel->basic_consume(
109+
$queueName,
110+
'',
111+
false,
112+
false,
113+
false,
114+
false,
115+
Closure::fromCallable([$this, 'processMessage'])
116+
);
117+
118+
$this->log(self::LOG_INFO, 'Connected to RabbitMQ', [
119+
'host' => GK_RABBITMQ_HOST,
120+
'port' => GK_RABBITMQ_PORT,
121+
'queue' => $queueName,
122+
]);
123+
}
124+
125+
protected function waitForMessage(): void {
126+
$timeout = $this->getWaitTimeout();
127+
if ($timeout === null) {
128+
$this->channel->wait();
129+
130+
return;
131+
}
132+
133+
$this->channel->wait(timeout: $timeout);
134+
}
135+
136+
public function run(): void {
137+
$this->log(self::LOG_INFO, $this->getWorkerName().' starting...');
138+
139+
while (true) {
140+
try {
141+
$this->connect();
142+
$this->retryDelay = 1; // Reset on successful connection
143+
144+
while ($this->channel->is_consuming()) {
145+
$this->waitForMessage();
146+
}
147+
} catch (AMQPConnectionException $e) {
148+
$this->log(self::LOG_ERROR, 'Connection lost: '.$e->getMessage());
149+
$this->cleanup();
150+
$this->log(self::LOG_INFO, "Reconnecting in {$this->retryDelay}s...");
151+
sleep($this->retryDelay);
152+
$this->retryDelay = min($this->retryDelay * 2, $this->maxRetryDelay);
153+
} catch (Exception $e) {
154+
$this->log(self::LOG_CRITICAL, 'Unexpected error: '.$e->getMessage(), [
155+
'exception' => get_class($e),
156+
'file' => $e->getFile(),
157+
'line' => $e->getLine(),
158+
'trace' => $e->getTraceAsString(),
159+
]);
160+
$this->cleanup();
161+
sleep($this->retryDelay);
162+
$this->retryDelay = min($this->retryDelay * 2, $this->maxRetryDelay);
163+
}
164+
}
165+
}
166+
167+
protected function cleanup(): void {
168+
try {
169+
if ($this->channel) {
170+
$this->channel->close();
171+
}
172+
if ($this->connection) {
173+
$this->connection->close();
174+
}
175+
} catch (Exception $e) {
176+
$this->log(self::LOG_WARNING, 'Error during cleanup: '.$e->getMessage());
177+
}
178+
}
179+
180+
public function __destruct() {
181+
$this->cleanup();
182+
}
183+
184+
public static function registerSignalHandlers(): void {
185+
pcntl_async_signals(true);
186+
pcntl_signal(SIGTERM, function () {
187+
echo "Received SIGTERM, shutting down gracefully...\n";
188+
exit(0);
189+
});
190+
pcntl_signal(SIGINT, function () {
191+
echo "Received SIGINT, shutting down gracefully...\n";
192+
exit(0);
193+
});
194+
}
195+
}

0 commit comments

Comments
 (0)