1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
|
<?php
namespace PhpAmqpLib\Connection\Heartbeat;
use PhpAmqpLib\Connection\AbstractConnection;
use PhpAmqpLib\Exception\AMQPRuntimeException;
/**
* Manages pcntl-based heartbeat sending for a {@link AbstractConnection}.
*/
abstract class AbstractSignalHeartbeatSender
{
/**
* @var AbstractConnection|null
*/
protected $connection;
/**
* @var bool
*/
protected $wasActive = false;
/**
* @param AbstractConnection $connection
* @throws AMQPRuntimeException
*/
public function __construct(AbstractConnection $connection)
{
if (!$this->isSupported()) {
throw new AMQPRuntimeException('Signal-based heartbeat sender is unsupported');
}
$this->connection = $connection;
}
public function __destruct()
{
$this->unregister();
}
/**
* @return bool
*/
protected function isSupported(): bool
{
return extension_loaded('pcntl')
&& function_exists('pcntl_async_signals')
&& (defined('AMQP_WITHOUT_SIGNALS') ? !AMQP_WITHOUT_SIGNALS : true);
}
/**
* Starts the heartbeats
*/
abstract public function register(): void;
/**
* Stops the heartbeats.
*/
abstract public function unregister(): void;
/**
* Handles the heartbeat when a signal interrupt is received
*
* @param int $interval
*/
protected function handleSignal(int $interval): void
{
if (!$this->connection) {
return;
}
// Support for lazy connections
if (!$this->wasActive && $this->connection->isConnected()) {
$this->wasActive = true;
}
if (!$this->wasActive) {
return;
}
if (!$this->connection->isConnected()) {
$this->unregister();
return;
}
if ($this->connection->isWriting()) {
return;
}
if (time() > ($this->connection->getLastActivity() + $interval)) {
$this->connection->checkHeartBeat();
}
}
}
|
