wip formatting
This commit is contained in:
parent
908f147cb3
commit
19ca49a4a8
|
|
@ -176,8 +176,7 @@ abstract class Controller implements HttpServerInterface
|
|||
|
||||
$laravelRequest = Request::createFromBase((new HttpFoundationFactory)->createRequest($serverRequest));
|
||||
|
||||
$this
|
||||
->ensureValidAppId($laravelRequest->appId)
|
||||
$this->ensureValidAppId($laravelRequest->appId)
|
||||
->ensureValidSignature($laravelRequest);
|
||||
|
||||
// Invoke the controller action
|
||||
|
|
|
|||
|
|
@ -30,11 +30,9 @@ class ConfigAppManager implements AppManager
|
|||
*/
|
||||
public function all(): array
|
||||
{
|
||||
return $this->apps
|
||||
->map(function (array $appAttributes) {
|
||||
return $this->convertIntoApp($appAttributes);
|
||||
})
|
||||
->toArray();
|
||||
return $this->apps->map(function (array $appAttributes) {
|
||||
return $this->convertIntoApp($appAttributes);
|
||||
})->toArray();
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -106,8 +104,7 @@ class ConfigAppManager implements AppManager
|
|||
$app->setPath($appAttributes['path']);
|
||||
}
|
||||
|
||||
$app
|
||||
->enableClientMessages($appAttributes['enable_client_messages'])
|
||||
$app->enableClientMessages($appAttributes['enable_client_messages'])
|
||||
->enableStatistics($appAttributes['enable_statistics'])
|
||||
->setCapacity($appAttributes['capacity'] ?? null)
|
||||
->setAllowedOrigins($appAttributes['allowed_origins'] ?? []);
|
||||
|
|
|
|||
|
|
@ -111,16 +111,12 @@ class LocalChannelManager implements ChannelManager
|
|||
*/
|
||||
public function getLocalConnections(): PromiseInterface
|
||||
{
|
||||
$connections = collect($this->channels)
|
||||
->map(function ($channelsWithConnections, $appId) {
|
||||
return collect($channelsWithConnections)->values();
|
||||
})
|
||||
->values()->collapse()
|
||||
->map(function ($channel) {
|
||||
return collect($channel->getConnections());
|
||||
})
|
||||
->values()->collapse()
|
||||
->toArray();
|
||||
$connections = collect($this->channels)->map(function ($channelsWithConnections, $appId) {
|
||||
return collect($channelsWithConnections)->values();
|
||||
})->values()->collapse()
|
||||
->map(function ($channel) {
|
||||
return collect($channel->getConnections());
|
||||
})->values()->collapse()->toArray();
|
||||
|
||||
return Helpers::createFulfilledPromise($connections);
|
||||
}
|
||||
|
|
@ -166,11 +162,9 @@ class LocalChannelManager implements ChannelManager
|
|||
$this->getLocalChannels($connection->app->id)->then(function ($channels) use ($connection) {
|
||||
collect($channels)->each->unsubscribe($connection);
|
||||
|
||||
collect($channels)
|
||||
->reject->hasConnections()
|
||||
->each(function (Channel $channel, string $channelName) use ($connection) {
|
||||
unset($this->channels[$connection->app->id][$channelName]);
|
||||
});
|
||||
collect($channels)->reject->hasConnections()->each(function (Channel $channel, string $channelName) use ($connection) {
|
||||
unset($this->channels[$connection->app->id][$channelName]);
|
||||
});
|
||||
});
|
||||
|
||||
$this->getLocalChannels($connection->app->id)->then(function ($channels) use ($connection) {
|
||||
|
|
@ -255,11 +249,9 @@ class LocalChannelManager implements ChannelManager
|
|||
return $collection->filter(function (Channel $channel) use ($channelName) {
|
||||
return $channel->getName() === $channelName;
|
||||
});
|
||||
})
|
||||
->flatMap(function (Channel $channel) {
|
||||
})->flatMap(function (Channel $channel) {
|
||||
return collect($channel->getConnections())->pluck('socketId');
|
||||
})
|
||||
->unique()->count();
|
||||
})->unique()->count();
|
||||
});
|
||||
}
|
||||
|
||||
|
|
@ -378,14 +370,13 @@ class LocalChannelManager implements ChannelManager
|
|||
*/
|
||||
public function getChannelsMembersCount($appId, array $channelNames): PromiseInterface
|
||||
{
|
||||
$results = collect($channelNames)
|
||||
->reduce(function ($results, $channel) use ($appId) {
|
||||
$results[$channel] = isset($this->users["{$appId}:{$channel}"])
|
||||
? count($this->users["{$appId}:{$channel}"])
|
||||
: 0;
|
||||
$results = collect($channelNames)->reduce(function ($results, $channel) use ($appId) {
|
||||
$results[$channel] = isset($this->users["{$appId}:{$channel}"])
|
||||
? count($this->users["{$appId}:{$channel}"])
|
||||
: 0;
|
||||
|
||||
return $results;
|
||||
}, []);
|
||||
return $results;
|
||||
}, []);
|
||||
|
||||
return Helpers::createFulfilledPromise($results);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -412,14 +412,13 @@ class RedisChannelManager extends LocalChannelManager
|
|||
public function removeObsoleteConnections(): PromiseInterface
|
||||
{
|
||||
$this->lock()->get(function () {
|
||||
$this->getConnectionsFromSet(0, now()->subMinutes(2)->format('U'))
|
||||
->then(function ($connections) {
|
||||
foreach ($connections as $socketId => $appId) {
|
||||
$connection = $this->fakeConnectionForApp($appId, $socketId);
|
||||
$this->getConnectionsFromSet(0, now()->subMinutes(2)->format('U'))->then(function ($connections) {
|
||||
foreach ($connections as $socketId => $appId) {
|
||||
$connection = $this->fakeConnectionForApp($appId, $socketId);
|
||||
|
||||
$this->unsubscribeFromAllChannels($connection);
|
||||
}
|
||||
});
|
||||
$this->unsubscribeFromAllChannels($connection);
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
return parent::removeObsoleteConnections();
|
||||
|
|
|
|||
|
|
@ -155,8 +155,7 @@ class Channel
|
|||
*/
|
||||
public function broadcast($appId, stdClass $payload, bool $replicate = true): bool
|
||||
{
|
||||
collect($this->getConnections())
|
||||
->each->send(json_encode($payload));
|
||||
collect($this->getConnections())->each->send(json_encode($payload));
|
||||
|
||||
if ($replicate) {
|
||||
$this->channelManager->broadcastAcrossServers($appId, null, $this->getName(), $payload);
|
||||
|
|
|
|||
|
|
@ -43,8 +43,7 @@ class MemoryCollector implements StatisticsCollector
|
|||
*/
|
||||
public function webSocketMessage($appId)
|
||||
{
|
||||
$this->findOrMake($appId)
|
||||
->webSocketMessage();
|
||||
$this->findOrMake($appId)->webSocketMessage();
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -55,8 +54,7 @@ class MemoryCollector implements StatisticsCollector
|
|||
*/
|
||||
public function apiMessage($appId)
|
||||
{
|
||||
$this->findOrMake($appId)
|
||||
->apiMessage();
|
||||
$this->findOrMake($appId)->apiMessage();
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -67,8 +65,7 @@ class MemoryCollector implements StatisticsCollector
|
|||
*/
|
||||
public function connection($appId)
|
||||
{
|
||||
$this->findOrMake($appId)
|
||||
->connection();
|
||||
$this->findOrMake($appId)->connection();
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -79,8 +76,7 @@ class MemoryCollector implements StatisticsCollector
|
|||
*/
|
||||
public function disconnection($appId)
|
||||
{
|
||||
$this->findOrMake($appId)
|
||||
->disconnection();
|
||||
$this->findOrMake($appId)->disconnection();
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -55,8 +55,9 @@ class RedisCollector extends MemoryCollector
|
|||
*/
|
||||
public function webSocketMessage($appId)
|
||||
{
|
||||
$this->ensureAppIsInSet($appId)
|
||||
->hincrby($this->channelManager->getRedisKey($appId, null, ['stats']), 'websocket_messages_count', 1);
|
||||
$this->ensureAppIsInSet($appId)->hincrby(
|
||||
$this->channelManager->getRedisKey($appId, null, ['stats']), 'websocket_messages_count', 1
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -67,8 +68,9 @@ class RedisCollector extends MemoryCollector
|
|||
*/
|
||||
public function apiMessage($appId)
|
||||
{
|
||||
$this->ensureAppIsInSet($appId)
|
||||
->hincrby($this->channelManager->getRedisKey($appId, null, ['stats']), 'api_messages_count', 1);
|
||||
$this->ensureAppIsInSet($appId)->hincrby(
|
||||
$this->channelManager->getRedisKey($appId, null, ['stats']), 'api_messages_count', 1
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -42,8 +42,7 @@ class DatabaseStore implements StatisticsStore
|
|||
return static::$model::where('created_at', '<', $moment->toDateTimeString())
|
||||
->when(! is_null($appId), function ($query) use ($appId) {
|
||||
return $query->whereAppId($appId);
|
||||
})
|
||||
->delete();
|
||||
})->delete();
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -54,12 +53,11 @@ class DatabaseStore implements StatisticsStore
|
|||
*/
|
||||
public function getRawRecords(callable $processQuery = null)
|
||||
{
|
||||
return static::$model::query()
|
||||
->when(! is_null($processQuery), function ($query) use ($processQuery) {
|
||||
return call_user_func($processQuery, $query);
|
||||
}, function ($query) {
|
||||
return $query->latest()->limit(120);
|
||||
})->get();
|
||||
return static::$model::query()->when(! is_null($processQuery), function ($query) use ($processQuery) {
|
||||
return call_user_func($processQuery, $query);
|
||||
}, function ($query) {
|
||||
return $query->latest()->limit(120);
|
||||
})->get();
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -74,11 +72,9 @@ class DatabaseStore implements StatisticsStore
|
|||
return $this->getRawRecords($processQuery)
|
||||
->when(! is_null($processCollection), function ($collection) use ($processCollection) {
|
||||
return call_user_func($processCollection, $collection);
|
||||
})
|
||||
->map(function (Model $statistic) {
|
||||
})->map(function (Model $statistic) {
|
||||
return $this->statisticToArray($statistic);
|
||||
})
|
||||
->toArray();
|
||||
})->toArray();
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -371,8 +371,7 @@ class PresenceChannelTest extends TestCase
|
|||
|
||||
$receiver->assertSentEvent('some-event', $message->getPayloadAsArray());
|
||||
|
||||
$this->getSubscribeClient()
|
||||
->assertNothingDispatched();
|
||||
$this->getSubscribeClient()->assertNothingDispatched();
|
||||
|
||||
$this->getPublishClient()->assertCalledWithArgs('publish', [
|
||||
$this->channelManager->getRedisKey('1234', 'presence-channel'),
|
||||
|
|
|
|||
|
|
@ -205,8 +205,7 @@ class PrivateChannelTest extends TestCase
|
|||
|
||||
$receiver->assertSentEvent('some-event', $message->getPayloadAsArray());
|
||||
|
||||
$this->getSubscribeClient()
|
||||
->assertNothingDispatched();
|
||||
$this->getSubscribeClient()->assertNothingDispatched();
|
||||
|
||||
$this->getPublishClient()->assertCalledWithArgs('publish', [
|
||||
$this->channelManager->getRedisKey('1234', 'private-channel'),
|
||||
|
|
|
|||
|
|
@ -186,8 +186,7 @@ class PublicChannelTest extends TestCase
|
|||
|
||||
$receiver->assertSentEvent('some-event', $message->getPayloadAsArray());
|
||||
|
||||
$this->getSubscribeClient()
|
||||
->assertNothingDispatched();
|
||||
$this->getSubscribeClient()->assertNothingDispatched();
|
||||
|
||||
$this->getPublishClient()->assertCalledWithArgs('publish', [
|
||||
$this->channelManager->getRedisKey('1234', 'public-channel'),
|
||||
|
|
|
|||
Loading…
Reference in New Issue