Using separate connection counts for global & local.

This commit is contained in:
Alex Renoki 2020-09-04 11:34:33 +03:00
parent 4e11355172
commit b9dfecab68
11 changed files with 64 additions and 17 deletions

View File

@ -164,9 +164,20 @@ class LocalClient implements ReplicationInterface
* Get the amount of unique connections.
*
* @param mixed $appId
* @return null|int
*/
public function getLocalConnectionsCount($appId)
{
return null;
}
/**
* Get the amount of connections aggregated on multiple instances.
*
* @param mixed $appId
* @return null|int|\React\Promise\PromiseInterface
*/
public function appConnectionsCount($appId)
public function getGlobalConnectionsCount($appId)
{
return null;
}

View File

@ -292,13 +292,22 @@ class RedisClient extends LocalClient
* Get the amount of unique connections.
*
* @param mixed $appId
* @return null|int
*/
public function getLocalConnectionsCount($appId)
{
return null;
}
/**
* Get the amount of connections aggregated on multiple instances.
*
* @param mixed $appId
* @return null|int|\React\Promise\PromiseInterface
*/
public function appConnectionsCount($appId)
public function getGlobalConnectionsCount($appId)
{
// Use the in-built Redis manager to avoid async run.
return $this->publishClient->hget($this->getTopicName($appId), 'connections') ?: 0;
return $this->publishClient->hget($this->getTopicName($appId), 'connections');
}
/**

View File

@ -106,7 +106,15 @@ interface ReplicationInterface
* Get the amount of unique connections.
*
* @param mixed $appId
* @return null|int
*/
public function getLocalConnectionsCount($appId);
/**
* Get the amount of connections aggregated on multiple instances.
*
* @param mixed $appId
* @return null|int|\React\Promise\PromiseInterface
*/
public function appConnectionsCount($appId);
public function getGlobalConnectionsCount($appId);
}

View File

@ -105,7 +105,7 @@ class MemoryStatisticsLogger implements StatisticsLogger
$this->createRecord($statistic, $appId);
$currentConnectionCount = $this->channelManager->getConnectionCount($appId);
$currentConnectionCount = $this->channelManager->getGlobalConnectionsCount($appId);
$statistic->reset($currentConnectionCount);
}

View File

@ -124,7 +124,7 @@ class RedisStatisticsLogger implements StatisticsLogger
$this->createRecord($statistic, $appId);
$currentConnectionCount = $this->channelManager->getConnectionCount($appId);
$currentConnectionCount = $this->channelManager->getGlobalConnectionsCount($appId);
$currentConnectionCount === 0
? $this->resetAppTraces($appId)

View File

@ -38,7 +38,15 @@ interface ChannelManager
* @param mixed $appId
* @return int
*/
public function getConnectionCount($appId): int;
public function getLocalConnectionsCount($appId): int;
/**
* Get the connections count across multiple servers.
*
* @param mixed $appId
* @return int
*/
public function getGlobalConnectionsCount($appId): int;
/**
* Remove connection from all channels.

View File

@ -73,7 +73,7 @@ class ArrayChannelManager implements ChannelManager
* @param mixed $appId
* @return int
*/
public function getConnectionCount($appId): int
public function getLocalConnectionsCount($appId): int
{
return collect($this->getChannels($appId))
->flatMap(function (Channel $channel) {
@ -83,6 +83,17 @@ class ArrayChannelManager implements ChannelManager
->count();
}
/**
* Get the connections count across multiple servers.
*
* @param mixed $appId
* @return int
*/
public function getGlobalConnectionsCount($appId): int
{
return $this->getLocalConnectionsCount($appId);
}
/**
* Remove connection from all channels.
*

View File

@ -24,13 +24,13 @@ class RedisChannelManager extends ArrayChannelManager
}
/**
* Get the connections count on the app.
* Get the connections count across multiple servers.
*
* @param mixed $appId
* @return int
*/
public function getConnectionCount($appId): int
public function getGlobalConnectionsCount($appId): int
{
return $this->replicator->appConnectionsCount($appId);
return $this->replicator->getGlobalConnectionsCount($appId);
}
}

View File

@ -165,7 +165,7 @@ class WebSocketHandler implements MessageComponentInterface
protected function limitConcurrentConnections(ConnectionInterface $connection)
{
if (! is_null($capacity = $connection->app->capacity)) {
$connectionsCount = $this->channelManager->getConnectionCount($connection->app->id);
$connectionsCount = $this->channelManager->getGlobalConnectionsCount($connection->app->id);
if ($connectionsCount >= $capacity) {
throw new ConnectionsOverCapacity();

View File

@ -12,7 +12,7 @@ class FakeMemoryStatisticsLogger extends MemoryStatisticsLogger
public function save()
{
foreach ($this->statistics as $appId => $statistic) {
$currentConnectionCount = $this->channelManager->getConnectionCount($appId);
$currentConnectionCount = $this->channelManager->getLocalConnectionsCount($appId);
$statistic->reset($currentConnectionCount);
}
}

View File

@ -49,7 +49,7 @@ class StatisticsLoggerTest extends TestCase
StatisticsLogger::save();
$this->assertEquals(3, StatisticsLogger::getForAppId(1234)['peak_connection_count']);
$this->assertEquals(2, StatisticsLogger::getForAppId(1234)['peak_connection_count']);
}
/** @test */
@ -93,7 +93,7 @@ class StatisticsLoggerTest extends TestCase
StatisticsLogger::save();
$this->assertEquals(3, StatisticsLogger::getForAppId(1234)['peak_connection_count']);
$this->assertEquals(1, StatisticsLogger::getForAppId(1234)['peak_connection_count']);
}
/** @test */