Skip to content

Commit e3572a1

Browse files
huangdijiahuangdijia
authored
refactor: improve server mutex implementation and configuration (#1019)
* refactor: improve server mutex implementation and configuration - Change server mutex prefix from APP_ENV to APP_NAME for better identification - Remove readonly properties from Consumer class to allow dependency injection flexibility - Improve RedisServerMutex constructor with better name construction and default owner handling - Add getName() method to ServerMutexInterface for consistency - Create ServerMutexCommand for mutex management operations This refactor improves the server mutex functionality by: - Using application name instead of environment for mutex prefix - Making Consumer properties mutable for better DI container compatibility - Enhancing RedisServerMutex with proper name formatting and owner handling - Providing interface consistency with getName() method * refactor: enhance ServerMutexCommand for improved action handling and configuration * refactor: update ServerMutexCommand and RedisServerMutex for improved clarity and functionality * Update * refactor: improve server mutex filtering logic to handle missing configuration * refactor: add ServerMutexCommand to the command list in ConfigProvider * refactor: improve variable naming in ServerMutexCommand for clarity * refactor: set default value for connection in RedisServerMutex constructor * refactor: update RedisServerMutex constructor to conditionally set options * refactor: move owner assignment in RedisServerMutex constructor for clarity --------- Co-authored-by: Deeka Wong <8337659+huangdijia@users.noreply.github.com>
1 parent 4d6451b commit e3572a1

6 files changed

Lines changed: 131 additions & 28 deletions

File tree

‎publish/trigger.php‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,7 @@
3131

3232
'server_mutex' => [
3333
'enable' => true,
34-
'prefix' => env('APP_ENV', 'dev') . '_',
34+
'prefix' => env('APP_NAME', 'trigger') . ':',
3535
'expires' => 30,
3636
'keepalive_interval' => 10,
3737
'retry_interval' => 10,

‎src/Command/ServerMutexCommand.php‎

Lines changed: 82 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,82 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
/**
5+
* This file is part of friendsofhyperf/components.
6+
*
7+
* @link https://github.com/friendsofhyperf/components
8+
* @document https://github.com/friendsofhyperf/components/blob/main/README.md
9+
* @contact huangdijia@gmail.com
10+
*/
11+
12+
namespace FriendsOfHyperf\Trigger\Command;
13+
14+
use FriendsOfHyperf\Trigger\Mutex\ServerMutexInterface;
15+
use Hyperf\Contract\ConfigInterface;
16+
use Hyperf\Redis\Redis;
17+
use Psr\Container\ContainerInterface;
18+
19+
use function Hyperf\Collection\collect;
20+
use function Hyperf\Support\make;
21+
22+
class ServerMutexCommand extends \Hyperf\Command\Command
23+
{
24+
protected ?string $signature = 'trigger:server-mutex {action : list|release} {--C|connection=default : The connection name}';
25+
26+
public function __construct(protected ContainerInterface $container)
27+
{
28+
parent::__construct();
29+
}
30+
31+
public function handle()
32+
{
33+
$action = $this->input->getArgument('action');
34+
$redis = $this->container->get(Redis::class);
35+
$config = $this->container->get(ConfigInterface::class);
36+
37+
if ($action === 'list') {
38+
$headers = ['Connection', 'Name', 'Owner', 'Expires At'];
39+
$mutexes = collect($config->get('trigger.connections', []))
40+
->reject(function ($config, $connection) {
41+
return ! ($config['server_mutex']['enable'] ?? true);
42+
})
43+
->transform(function ($connectionConfig, $connection) use ($redis) {
44+
$mutex = make(ServerMutexInterface::class, [
45+
'connection' => $connection,
46+
'options' => $connectionConfig['server_mutex'] ?? [],
47+
]);
48+
49+
$name = $mutex->getName();
50+
$ttl = $redis->ttl($name);
51+
$owner = $redis->get($name) ?? 'none';
52+
53+
return [$connection, $name, $owner, $ttl > 0 ? date('Y-m-d H:i:s', time() + $ttl) : 'expired'];
54+
})
55+
->toArray();
56+
57+
$this->table($headers, $mutexes);
58+
return;
59+
}
60+
61+
if ($action === 'release') {
62+
$connection = $this->input->getOption('connection');
63+
64+
if (! $connection) {
65+
$this->error('Please specify the connection name using --connection option.');
66+
return;
67+
}
68+
69+
$options = $config->get("trigger.connections.{$connection}.server_mutex", []);
70+
$mutex = make(ServerMutexInterface::class, [
71+
'connection' => $connection,
72+
'options' => $options,
73+
]);
74+
$mutex->release(true);
75+
76+
$this->line("Released mutex for connection: {$connection}");
77+
return;
78+
}
79+
80+
$this->error('Invalid action. Use "list" or "release".');
81+
}
82+
}

‎src/ConfigProvider.php‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@ public function __invoke(): array
2828
],
2929
'commands' => [
3030
Command\ConsumeCommand::class,
31+
Command\ServerMutexCommand::class,
3132
Command\SubscribersCommand::class,
3233
Command\TriggersCommand::class,
3334
],

‎src/Consumer.php‎

Lines changed: 23 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -28,39 +28,45 @@
2828

2929
class Consumer
3030
{
31-
public readonly Config $config;
31+
public Config $config;
3232

33-
public readonly string $name;
33+
public string $name;
3434

35-
public readonly string $identifier;
35+
public string $identifier;
3636

37-
public readonly ?HealthMonitor $healthMonitor;
37+
public ?HealthMonitor $healthMonitor = null;
3838

39-
public readonly ?ServerMutexInterface $serverMutex;
39+
public ?ServerMutexInterface $serverMutex = null;
4040

41-
public readonly BinLogCurrentSnapshotInterface $binLogCurrentSnapshot;
41+
public BinLogCurrentSnapshotInterface $binLogCurrentSnapshot;
4242

4343
private bool $stopped = false;
4444

4545
public function __construct(
46-
protected readonly SubscriberManager $subscriberManager,
47-
protected readonly TriggerManager $triggerManager,
48-
public readonly string $connection = 'default',
46+
protected SubscriberManager $subscriberManager,
47+
protected TriggerManager $triggerManager,
48+
public string $connection = 'default',
4949
array $options = [],
50-
public readonly ?LoggerInterface $logger = null
50+
public ?LoggerInterface $logger = null
5151
) {
5252
$this->name = $options['name'] ?? sprintf('trigger.%s', $this->connection);
5353
$this->identifier = $options['identifier'] ?? sprintf('trigger.%s', $this->connection);
5454
$this->config = new Config($options);
5555

5656
$this->binLogCurrentSnapshot = make(BinLogCurrentSnapshotInterface::class, ['consumer' => $this]);
57-
$this->healthMonitor = $this->config->get('health_monitor.enable', true) ? make(HealthMonitor::class, ['consumer' => $this]) : null;
58-
$this->serverMutex = $this->config->get('server_mutex.enable', true) ? make(ServerMutexInterface::class, [
59-
'name' => 'trigger:mutex:' . $this->connection,
60-
'owner' => Util::getInternalIp(),
61-
'options' => $this->config->get('server_mutex', []) + ['connection' => $this->connection],
62-
'logger' => $this->logger,
63-
]) : null;
57+
58+
if ($this->config->get('health_monitor.enable', true)) {
59+
$this->healthMonitor = make(HealthMonitor::class, ['consumer' => $this]);
60+
}
61+
62+
if ($this->config->get('server_mutex.enable', true)) {
63+
$this->serverMutex = make(ServerMutexInterface::class, [
64+
'connection' => $this->connection,
65+
'options' => (array) $this->config->get('server_mutex', []),
66+
'owner' => Util::getInternalIp(),
67+
'logger' => $this->logger,
68+
]);
69+
}
6470
}
6571

6672
public function start(): void

‎src/Mutex/RedisServerMutex.php‎

Lines changed: 22 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -31,24 +31,36 @@ class RedisServerMutex implements ServerMutexInterface
3131

3232
protected int $retryInterval = 10;
3333

34-
protected string $connection = 'default';
34+
protected string $name = '';
3535

3636
public function __construct(
3737
protected Redis $redis,
38-
protected ?string $name = null,
39-
protected ?string $owner = null,
38+
protected string $connection = 'default',
4039
array $options = [],
40+
protected ?string $owner = null,
4141
protected ?LoggerInterface $logger = null
4242
) {
43-
$this->expires = (int) ($options['expires'] ?? 60);
44-
$this->keepaliveInterval = (int) ($options['keepalive_interval'] ?? 10);
45-
$this->name = ($options['prefix'] ?? '') . ($name ?? sprintf('trigger:server:%s', $this->connection));
46-
$this->owner = $owner ?? Util::getInternalIp();
47-
if (isset($options['connection'])) {
48-
$this->connection = $options['connection'];
43+
$this->name = sprintf(
44+
'%s%s',
45+
$options['prefix'] ?? 'trigger:mutex:',
46+
$this->connection
47+
);
48+
if (isset($options['expires'])) {
49+
$this->expires = (int) $options['expires'];
50+
}
51+
if (isset($options['keepalive_interval'])) {
52+
$this->keepaliveInterval = (int) $options['keepalive_interval'];
53+
}
54+
if (isset($options['retry_interval'])) {
55+
$this->retryInterval = (int) $options['retry_interval'];
4956
}
57+
$this->owner = $owner ?? Util::getInternalIp();
5058
$this->timer = new Timer($this->logger);
51-
$this->retryInterval = (int) ($options['retry_interval'] ?? 10);
59+
}
60+
61+
public function getName(): string
62+
{
63+
return $this->name;
5264
}
5365

5466
public function attempt(?callable $callback = null): void

‎src/Mutex/ServerMutexInterface.php‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,8 @@
1313

1414
interface ServerMutexInterface
1515
{
16+
public function getName(): string;
17+
1618
public function attempt(?callable $callback = null);
1719

1820
public function release(bool $force = false);

0 commit comments

Comments
 (0)