Consumer.php

<?php
 
declare(strict_types=1);
/**
 * This file is part of friendsofhyperf/components.
 *
 * @link https://github.com/friendsofhyperf/components
 * @document https://github.com/friendsofhyperf/components/blob/main/README.md
 * @contact huangdijia@gmail.com
 */
 
namespace FriendsOfHyperf\Trigger;
 
use FriendsOfHyperf\Trigger\Monitor\HealthMonitor;
use FriendsOfHyperf\Trigger\Mutex\ServerMutexInterface;
use FriendsOfHyperf\Trigger\Snapshot\BinLogCurrentSnapshotInterface;
use Hyperf\Coordinator\Constants;
use Hyperf\Coordinator\CoordinatorManager;
use Hyperf\Coroutine\Coroutine;
use Hyperf\Stringable\Str;
use MySQLReplication\Config\ConfigBuilder;
use MySQLReplication\MySQLReplicationFactory;
use Psr\Log\LoggerInterface;
use Throwable;
 
use function Hyperf\Support\make;
use function Hyperf\Tappable\tap;
 
class Consumer
{
 public Config $config;
 
 public string $name;
 
 public string $identifier;
 
 public ?HealthMonitor $healthMonitor = null;
 
 public ?ServerMutexInterface $serverMutex = null;
 
 public BinLogCurrentSnapshotInterface $binLogCurrentSnapshot;
 
 private bool $stopped = false;
 
 public function __construct(
 protected SubscriberManager $subscriberManager,
 protected TriggerManager $triggerManager,
 public string $connection = 'default',
 array $options = [],
 public ?LoggerInterface $logger = null
 ) {
 $this->name = $options['name'] ?? sprintf('trigger.%s', $this->connection);
 $this->identifier = $options['identifier'] ?? sprintf('trigger.%s', $this->connection);
 $this->config = new Config($options);
 
 $this->binLogCurrentSnapshot = make(BinLogCurrentSnapshotInterface::class, ['consumer' => $this]);
 
 if ($this->config->get('health_monitor.enable', true)) {
 $this->healthMonitor = make(HealthMonitor::class, ['consumer' => $this]);
 }
 
 if ($this->config->get('server_mutex.enable', true)) {
 $this->serverMutex = make(ServerMutexInterface::class, [
 'connection' => $this->connection,
 'options' => (array) $this->config->get('server_mutex', []),
 'owner' => Util::getInternalIp(),
 'logger' => $this->logger,
 ]);
 }
 }
 
 public function start(): void
 {
 $context = ['connection' => $this->connection];
 $callback = function () use ($context) {
 // Reset stopped status
 $this->stopped = false;
 
 // Health monitor start
 $this->healthMonitor?->process();
 
 $replication = $this->makeReplication();
 
 // Replication start
 CoordinatorManager::until($this->identifier)->resume();
 
 $this->logger?->debug('[{connection}] Consumer started.', $context);
 
 // Worker exit
 Coroutine::create(function () use ($context) {
 CoordinatorManager::until(Constants::WORKER_EXIT)->yield();
 $this->stop();
 $this->logger?->warning('[{connection}] Consumer exit.', $context);
 });
 
 while (1) {
 if ($this->isStopped()) {
 $this->logger?->warning('[{connection}] Consumer stopped.', $context);
 break;
 }
 
 try {
 $replication->consume();
 } catch (Throwable $e) {
 $this->logger?->warning('[{connection}] Error occurred, will retry later.', $context + ['message' => $e->getMessage()]);
 $this->logger?->error((string) $e);
 $this->stop();
 }
 }
 };
 
 if ($this->serverMutex) {
 $this->serverMutex->attempt($callback);
 } else {
 $callback();
 }
 }
 
 /**
 * @deprecated since v3.1, use `$this->binLogCurrentSnapshot` instead, will be removed in v3.2.
 */
 public function getBinLogCurrentSnapshot(): BinLogCurrentSnapshotInterface
 {
 return $this->binLogCurrentSnapshot;
 }
 
 /**
 * @deprecated since v3.1, use `$this->healthMonitor` instead, will be removed in v3.2.
 */
 public function getHealthMonitor(): ?HealthMonitor
 {
 return $this->healthMonitor;
 }
 
 /**
 * @deprecated since v3.1, use `$this->name` instead, will be removed in v3.2.
 */
 public function getName(): string
 {
 return $this->name;
 }
 
 /**
 * @deprecated since v3.1, use `$this->config->get($key, $default)` instead, will be removed in v3.2.
 */
 public function getOption(?string $key = null, mixed $default = null): mixed
 {
 if (is_null($key)) {
 return (fn () => $this->configs ?? [])->call($this->config);
 }
 
 return $this->config->get($key, $default);
 }
 
 /**
 * @deprecated since v3.1, use `$this->connection` instead, will be removed in v3.2.
 */
 public function getConnection(): string
 {
 return $this->connection;
 }
 
 /**
 * @deprecated since v3.1, use `$this->identifier` instead, will be removed in v3.2.
 */
 public function getIdentifier(): string
 {
 return $this->identifier;
 }
 
 public function stop(): void
 {
 $this->stopped = true;
 $this->serverMutex?->release();
 }
 
 public function isStopped(): bool
 {
 return $this->stopped;
 }
 
 protected function makeReplication(): MySQLReplicationFactory
 {
 $connection = $this->connection;
 // Get databases of replication
 $databasesOnly = array_merge(
 $this->config->get('databases_only', []),
 $this->triggerManager->getDatabases($connection)
 );
 // Get tables of replication
 $tablesOnly = array_merge(
 $this->config->get('tables_only', []),
 $this->triggerManager->getTables($connection)
 );
 
 $configBuilder = (new ConfigBuilder())
 ->withUser($this->config->get('user', 'root'))
 ->withHost($this->config->get('host', '127.0.0.1'))
 ->withPassword($this->config->get('password', 'root'))
 ->withPort((int) $this->config->get('port', 3306))
 ->withSlaveId(random_int(10000, 9999999))
 ->withHeartbeatPeriod((float) $this->config->get('heartbeat_period', 3))
 ->withDatabasesOnly($databasesOnly)
 ->withTablesOnly($tablesOnly);
 
 if (method_exists($configBuilder, 'withSlaveUuid')) { // php-mysql-replication >= 8.0
 $configBuilder->withSlaveUuid(Str::uuid()->toString());
 }
 
 if ($binLogCurrent = $this->binLogCurrentSnapshot->get()) {
 $configBuilder->withBinLogFileName($binLogCurrent->getBinFileName())
 ->withBinLogPosition($binLogCurrent->getBinLogPosition());
 
 $this->logger?->debug(
 '[{connection}] Continue with position, binLogCurrent: {binlog_current}',
 compact('connection') + ['binlog_current' => json_encode($binLogCurrent->jsonSerialize())]
 );
 }
 
 $eventDispatcher = make(EventDispatcher::class);
 
 return tap(
 make(MySQLReplicationFactory::class, [
 'config' => $configBuilder->build(),
 'eventDispatcher' => $eventDispatcher,
 'logger' => $this->logger,
 ]),
 function (MySQLReplicationFactory $factory) use ($connection) {
 $subscribers = $this->subscriberManager->get($connection);
 
 foreach ($subscribers as $subscriber) {
 $factory->registerSubscriber(make($subscriber, [
 'consumer' => $this,
 'logger' => $this->logger,
 ]));
 }
 }
 );
 }
}