1: | <?php |
2: | |
3: | declare(strict_types=1); |
4: | |
5: | |
6: | |
7: | |
8: | |
9: | |
10: | |
11: | |
12: | |
13: | |
14: | |
15: | |
16: | |
17: | |
18: | |
19: | |
20: | |
21: | |
22: | namespace OpenSearch\ConnectionPool; |
23: | |
24: | use OpenSearch\Common\Exceptions\NoNodesAvailableException; |
25: | use OpenSearch\ConnectionPool\Selectors\SelectorInterface; |
26: | use OpenSearch\Connections\Connection; |
27: | use OpenSearch\Connections\ConnectionInterface; |
28: | use OpenSearch\Connections\ConnectionFactoryInterface; |
29: | |
30: | class StaticConnectionPool extends AbstractConnectionPool implements ConnectionPoolInterface |
31: | { |
32: | |
33: | |
34: | |
35: | private $pingTimeout = 60; |
36: | |
37: | |
38: | |
39: | |
40: | private $maxPingTimeout = 3600; |
41: | |
42: | |
43: | |
44: | |
45: | |
46: | public function __construct($connections, SelectorInterface $selector, ConnectionFactoryInterface $factory, $connectionPoolParams) |
47: | { |
48: | parent::__construct($connections, $selector, $factory, $connectionPoolParams); |
49: | $this->scheduleCheck(); |
50: | } |
51: | |
52: | public function nextConnection(bool $force = false): ConnectionInterface |
53: | { |
54: | $skipped = []; |
55: | |
56: | $total = count($this->connections); |
57: | while ($total--) { |
58: | |
59: | |
60: | |
61: | $connection = $this->selector->select($this->connections); |
62: | if ($connection->isAlive() === true) { |
63: | return $connection; |
64: | } |
65: | |
66: | if ($this->readyToRevive($connection) === true) { |
67: | if ($connection->ping() === true) { |
68: | return $connection; |
69: | } |
70: | } else { |
71: | $skipped[] = $connection; |
72: | } |
73: | } |
74: | |
75: | |
76: | foreach ($skipped as $connection) { |
77: | if ($connection->ping() === true) { |
78: | return $connection; |
79: | } |
80: | } |
81: | |
82: | throw new NoNodesAvailableException("No alive nodes found in your cluster"); |
83: | } |
84: | |
85: | public function scheduleCheck(): void |
86: | { |
87: | foreach ($this->connections as $connection) { |
88: | $connection->markDead(); |
89: | } |
90: | } |
91: | |
92: | private function readyToRevive(Connection $connection): bool |
93: | { |
94: | $timeout = min( |
95: | $this->pingTimeout * pow(2, $connection->getPingFailures()), |
96: | $this->maxPingTimeout |
97: | ); |
98: | |
99: | if ($connection->getLastPing() + $timeout < time()) { |
100: | return true; |
101: | } else { |
102: | return false; |
103: | } |
104: | } |
105: | } |
106: | |