103 lines
2.6 KiB
PHP
103 lines
2.6 KiB
PHP
<?php
|
|
|
|
declare(strict_types=1);
|
|
|
|
require dirname(__DIR__).'/../vendor/autoload.php';
|
|
|
|
use Amp\ByteStream;
|
|
use Amp\Log\ConsoleFormatter;
|
|
use Amp\Log\StreamHandler;
|
|
use Amp\Loop;
|
|
use Monolog\Logger;
|
|
use Monolog\Processor\PsrLogMessageProcessor;
|
|
use Nsq\Config\ClientConfig;
|
|
use Nsq\Config\LookupConfig;
|
|
use Nsq\Lookup;
|
|
use Nsq\Producer;
|
|
|
|
use function Amp\asyncCall;
|
|
use function Amp\delay;
|
|
|
|
Loop::run(static function () {
|
|
$handler = new StreamHandler(ByteStream\getStdout());
|
|
$handler->setFormatter(new ConsoleFormatter());
|
|
$logger = new Logger('publisher', [$handler], [new PsrLogMessageProcessor()]);
|
|
|
|
$clientConfig = new ClientConfig();
|
|
|
|
/** @var Producer[] $producers */
|
|
$producers = [];
|
|
|
|
$lookupConfig = new LookupConfig();
|
|
|
|
$lookup = Lookup::create(
|
|
['http://nsqlookupd0:4161', 'http://nsqlookupd1:4161', 'http://nsqlookupd2:4161'],
|
|
$lookupConfig,
|
|
$logger,
|
|
);
|
|
|
|
$isRunning = true;
|
|
|
|
asyncCall(static function () use ($lookup, $clientConfig, $logger, &$producers, &$isRunning) {
|
|
while ($isRunning) {
|
|
/** @var Lookup\Producer[] $nodes */
|
|
$nodes = yield $lookup->nodes();
|
|
|
|
foreach ($nodes as $node) {
|
|
$address = $node->toTcpUri();
|
|
|
|
if (array_key_exists($address, $producers)) {
|
|
continue;
|
|
}
|
|
|
|
asyncCall(function () use ($address, $clientConfig, $logger, &$producers) {
|
|
yield ($producers[$address] = Producer::create($address, $clientConfig, $logger))
|
|
->onClose(function () use (&$producers, $address) {
|
|
unset($producers[$address]);
|
|
})
|
|
->connect()
|
|
;
|
|
});
|
|
}
|
|
|
|
yield delay(5000);
|
|
|
|
yield Amp\Dns\resolver()->reloadConfig(); // for reload /etc/hosts
|
|
}
|
|
});
|
|
|
|
Loop::delay(5000, function () use (&$isRunning, $logger) {
|
|
$logger->info('Stopping producer.');
|
|
|
|
$isRunning = false;
|
|
});
|
|
|
|
$counter = 0;
|
|
while (true) {
|
|
if (!$isRunning) {
|
|
foreach ($producers as $producer) {
|
|
$producer->close();
|
|
}
|
|
|
|
break;
|
|
}
|
|
|
|
if ([] === $producers) {
|
|
yield delay(200);
|
|
|
|
continue;
|
|
}
|
|
|
|
$index = array_rand($producers);
|
|
$producer = $producers[$index];
|
|
|
|
if (!$producer->isConnected()) {
|
|
yield delay(100);
|
|
|
|
continue;
|
|
}
|
|
|
|
yield $producer->publish('local', 'This is message of count '.$counter++);
|
|
}
|
|
});
|