1 | <?php |
||
2 | |||
3 | declare(strict_types=1); |
||
4 | |||
5 | namespace Umbrellio\TableSync\Rabbit; |
||
6 | |||
7 | use ErrorException; |
||
8 | use PhpAmqpLib\Connection\AbstractConnection; |
||
9 | use PhpAmqpLib\Connection\AMQPSSLConnection; |
||
10 | |||
11 | class ConnectionContainer |
||
12 | { |
||
13 | private const DEFAULT_WAIT_BEFORE_RECONNECT_MICROSECONDS = 1000000; |
||
14 | |||
15 | private $host; |
||
16 | private $port; |
||
17 | private $user; |
||
18 | private $pass; |
||
19 | private $vhost; |
||
20 | private $sslOptions; |
||
21 | private $options; |
||
22 | private $waitBeforeReconnectMicroseconds; |
||
23 | |||
24 | /** |
||
25 | * @var AbstractConnection|null |
||
26 | */ |
||
27 | private $connection; |
||
28 | |||
29 | public function __construct( |
||
30 | string $host, |
||
31 | string $port, |
||
32 | string $user, |
||
33 | string $pass, |
||
34 | string $vhost = '/', |
||
35 | array $sslOptions = [], |
||
36 | array $options = [] |
||
37 | ) { |
||
38 | $this->host = $host; |
||
39 | $this->port = $port; |
||
40 | $this->user = $user; |
||
41 | $this->pass = $pass; |
||
42 | $this->vhost = $vhost; |
||
43 | $this->sslOptions = $sslOptions; |
||
44 | $this->options = $options; |
||
45 | |||
46 | $this->waitBeforeReconnectMicroseconds = $options['wait_before_reconnect_microseconds'] |
||
47 | ?? self::DEFAULT_WAIT_BEFORE_RECONNECT_MICROSECONDS; |
||
48 | } |
||
49 | |||
50 | public function __destruct() |
||
51 | { |
||
52 | $this->close(); |
||
53 | } |
||
54 | |||
55 | public function connection(): AbstractConnection |
||
56 | { |
||
57 | if ($this->connection === null) { |
||
58 | $this->reconnect(false); |
||
59 | } |
||
60 | |||
61 | return $this->connection; |
||
62 | } |
||
63 | |||
64 | public function reconnect(bool $wait = true): void |
||
65 | { |
||
66 | $this->close(); |
||
67 | |||
68 | if ($wait) { |
||
69 | usleep($this->waitBeforeReconnectMicroseconds); |
||
70 | } |
||
71 | |||
72 | $this->connection = new AMQPSSLConnection( |
||
73 | $this->host, |
||
74 | $this->port, |
||
75 | $this->user, |
||
76 | $this->pass, |
||
77 | $this->vhost, |
||
78 | $this->sslOptions, |
||
79 | $this->options |
||
80 | ); |
||
81 | } |
||
82 | |||
83 | public function close(): void |
||
84 | { |
||
85 | try { |
||
86 | if ($this->connection === null) { |
||
87 | return; |
||
88 | } |
||
89 | |||
90 | $this->connection->close(); |
||
91 | $this->connection = null; |
||
92 | } catch (ErrorException $errorException) { |
||
1 ignored issue
–
show
Coding Style
Comprehensibility
introduced
by
![]() |
|||
93 | } |
||
94 | } |
||
95 | } |
||
96 |