nsq/nsqphp
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Lookup: graceful shutdown

Browse Source
This commit is contained in:
Konstantin Grachev
2021-09-19 01:57:53 +03:00
parent be696f17b5
commit 45048320a8
1 changed files with 20 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
src/Lookup.php
View File

@@ -23,10 +23,12 @@ use function Amp\delay;
final class Lookup final class Lookup
{ {
/** /**
* @var array<string, array<string, \Nsq\Lookup\Producer[]>> * @var array<string, array<string, Lookup\Producer[]>>
*/ */
private array $producers = []; private array $producers = [];
private array $consumers = [];
private array $running = []; private array $running = [];
private array $topicWatchers = []; private array $topicWatchers = [];
@@ -91,11 +93,10 @@ final class Lookup
public function stop(): void public function stop(): void
{ {
foreach ($this->running as $topic => $channels) { $this->producers = [];
foreach (array_keys($channels) as $channel) { $this->consumers = [];
$this->unsubscribe($topic, $channel); $this->running = [];
} $this->topicWatchers = [];
}
$this->logger->info('Lookup stopped.'); $this->logger->info('Lookup stopped.');
} }
@@ -108,16 +109,9 @@ final class Lookup
$this->running[$topic][$channel] = true; $this->running[$topic][$channel] = true;
/** @var Consumer[] $consumers */ asyncCall(function () use ($topic, $channel, $onMessage, $config) {
$consumers = [];
asyncCall(function () use ($topic, $channel, $onMessage, $config, &$consumers) {
while (true) { while (true) {
if (null === ($this->running[$topic][$channel] ?? null)) { if (null === ($this->running[$topic][$channel] ?? null)) {
foreach ($consumers as $consumer) {
$consumer->close();
}
return; return;
} }
@@ -128,12 +122,12 @@ final class Lookup
$producers = yield $producers->promise(); $producers = yield $producers->promise();
} }
foreach (array_diff_key($consumers, $producers) as $address => $producer) { foreach (array_diff_key($this->consumers, $producers) as $address => $producer) {
unset($consumers[$address]); unset($this->consumers[$address]);
} }
foreach ($producers as $address => $producer) { foreach ($producers as $address => $producer) {
if (\array_key_exists($address, $consumers)) { if (null !== ($this->consumers[$address][$topic][$channel] ?? null)) {
continue; continue;
} }
@@ -146,7 +140,6 @@ final class Lookup
$config, $config,
$this->logger, $this->logger,
), ),
$consumers,
); );
} }
@@ -176,18 +169,21 @@ final class Lookup
$this->logger->info('Unsubscribed.', compact('topic', 'channel')); $this->logger->info('Unsubscribed.', compact('topic', 'channel'));
} }
private function keepConnection(Consumer $consumer, &$consumers): void private function keepConnection(Consumer $consumer): void
{ {
$consumers[$consumer->address] = $consumer; $this->consumers[$consumer->address][$consumer->topic][$consumer->channel] = $consumer;
asyncCall(function () use ($consumer, &$consumers) { asyncCall(function () use ($consumer) {
while (\array_key_exists($consumer->address, $consumers)) { while (null !== ($this->consumers[$consumer->address][$consumer->topic][$consumer->channel] ?? null)) {
try { try {
yield $consumer->connect(); yield $consumer->connect();
} catch (DnsException $e) { } catch (DnsException $e) {
$this->logger->error($e->getMessage(), ['exception' => $e]); $this->logger->error($e->getMessage(), ['exception' => $e]);
unset($consumers[$consumer->address], $this->producers[$consumer->topic][$consumer->address]); unset(
$this->consumers[$consumer->address],
$this->producers[$consumer->topic][$consumer->address],
);
return; return;
} catch (\Throwable $e) { } catch (\Throwable $e) {
@@ -199,7 +195,7 @@ final class Lookup
} }
while (true) { while (true) {
if (!\array_key_exists($consumer->address, $consumers)) { if (null === ($this->consumers[$consumer->address][$consumer->topic][$consumer->channel] ?? null)) {
$consumer->close(); $consumer->close();
return; return;