|
1
|
|
|
<?php |
|
2
|
|
|
|
|
3
|
|
|
namespace BenTools\MercurePHP\Transport\Redis; |
|
4
|
|
|
|
|
5
|
|
|
use BenTools\MercurePHP\Helpers\LoggerAwareTrait; |
|
6
|
|
|
use BenTools\MercurePHP\Helpers\RedisHelper; |
|
7
|
|
|
use BenTools\MercurePHP\Transport\TransportFactoryInterface; |
|
8
|
|
|
use Clue\React\Redis\Client as AsynchronousClient; |
|
9
|
|
|
use Clue\React\Redis\Factory; |
|
10
|
|
|
use Psr\Log\LoggerInterface; |
|
11
|
|
|
use React\EventLoop\LoopInterface; |
|
12
|
|
|
use React\Promise\PromiseInterface; |
|
13
|
|
|
|
|
14
|
|
|
use function React\Promise\all; |
|
15
|
|
|
|
|
16
|
|
|
final class RedisTransportFactory implements TransportFactoryInterface |
|
17
|
|
|
{ |
|
18
|
|
|
use LoggerAwareTrait; |
|
19
|
|
|
|
|
20
|
6 |
|
private LoopInterface $loop; |
|
21
|
|
|
|
|
22
|
6 |
|
public function __construct(LoopInterface $loop, ?LoggerInterface $logger = null) |
|
23
|
6 |
|
{ |
|
24
|
|
|
$this->loop = $loop; |
|
25
|
5 |
|
$this->logger = $logger; |
|
|
|
|
|
|
26
|
|
|
} |
|
27
|
5 |
|
|
|
28
|
|
|
public function supports(string $dsn): bool |
|
29
|
|
|
{ |
|
30
|
1 |
|
return RedisHelper::isRedisDSN($dsn); |
|
31
|
|
|
} |
|
32
|
1 |
|
|
|
33
|
|
|
public function create(string $dsn): PromiseInterface |
|
34
|
1 |
|
{ |
|
35
|
1 |
|
$factory = new Factory($this->loop); |
|
36
|
|
|
$promises = [ |
|
37
|
1 |
|
'subscriber' => $factory->createClient($dsn) |
|
38
|
1 |
|
->then( |
|
39
|
|
|
function (AsynchronousClient $client) { |
|
40
|
|
|
$client->on( |
|
41
|
|
|
'close', |
|
42
|
1 |
|
function () { |
|
43
|
|
|
$this->logger()->error('Connection closed.'); |
|
44
|
|
|
$this->loop->stop(); |
|
45
|
1 |
|
} |
|
46
|
1 |
|
); |
|
47
|
|
|
|
|
48
|
|
|
return $client; |
|
49
|
|
|
}, |
|
50
|
1 |
|
function (\Exception $exception) { |
|
51
|
|
|
$this->loop->stop(); |
|
52
|
1 |
|
$this->logger->error($exception->getMessage()); |
|
|
|
|
|
|
53
|
1 |
|
} |
|
54
|
|
|
), |
|
55
|
1 |
|
'publisher' => $factory->createClient($dsn) |
|
56
|
1 |
|
->then( |
|
57
|
|
|
function (AsynchronousClient $client) { |
|
58
|
|
|
$client->on( |
|
59
|
|
|
'close', |
|
60
|
1 |
|
function () { |
|
61
|
|
|
$this->logger()->error('Connection closed.'); |
|
62
|
|
|
$this->loop->stop(); |
|
63
|
1 |
|
} |
|
64
|
1 |
|
); |
|
65
|
|
|
|
|
66
|
|
|
return $client; |
|
67
|
|
|
}, |
|
68
|
1 |
|
function (\Exception $exception) { |
|
69
|
|
|
$this->loop->stop(); |
|
70
|
|
|
$this->logger()->error($exception->getMessage()); |
|
71
|
|
|
} |
|
72
|
1 |
|
), |
|
73
|
1 |
|
]; |
|
74
|
|
|
|
|
75
|
1 |
|
return all($promises) |
|
76
|
1 |
|
->then( |
|
77
|
1 |
|
function (iterable $results): array { |
|
78
|
|
|
$clients = []; |
|
79
|
|
|
foreach ($results as $key => $client) { |
|
80
|
|
|
$clients[$key] = $client; |
|
81
|
1 |
|
} |
|
82
|
1 |
|
|
|
83
|
|
|
RedisHelper::testAsynchronousClient($clients['subscriber'], $this->loop, $this->logger()); |
|
84
|
1 |
|
RedisHelper::testAsynchronousClient($clients['publisher'], $this->loop, $this->logger()); |
|
85
|
1 |
|
|
|
86
|
|
|
return $clients; |
|
87
|
1 |
|
} |
|
88
|
1 |
|
) |
|
89
|
1 |
|
->then( |
|
90
|
1 |
|
fn (array $clients) => new RedisTransport($clients['subscriber'], $clients['publisher']) |
|
91
|
|
|
); |
|
92
|
1 |
|
} |
|
93
|
|
|
} |
|
94
|
|
|
|
bpolaszek /
mercure-php-hub
Loading history...
This property has been deprecated. The supplier of the class has supplied an explanatory message.
The explanatory message should give you some clue as to whether and when the property will be removed from the class and what other property to use instead.