blax-software
/
laravel-websockets
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

formatting

This commit is contained in:
Alex Renoki 2020-12-07 20:36:39 +02:00
parent 6be62b149d
commit 8308a7d16d
15 changed files with 557 additions and 756 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

86
src/API/FetchChannels.php
View File

@ -28,50 +28,48 @@ class FetchChannels extends Controller
} }
} }
return $this->channelManager return $this->channelManager->getGlobalChannels($request->appId)->then(function ($channels) use ($request, $attributes) {
->getGlobalChannels($request->appId) $channels = collect($channels)->keyBy(function ($channel) {
->then(function ($channels) use ($request, $attributes) { return $channel instanceof Channel
$channels = collect($channels)->keyBy(function ($channel) { ? $channel->getName()
return $channel instanceof Channel : $channel;
? $channel->getName()
: $channel;
});
if ($request->has('filter_by_prefix')) {
$channels = $channels->filter(function ($channel, $channelName) use ($request) {
return Str::startsWith($channelName, $request->filter_by_prefix);
});
}
$channelNames = $channels->map(function ($channel) {
return $channel instanceof Channel
? $channel->getName()
: $channel;
})->toArray();
return $this->channelManager
->getChannelsMembersCount($request->appId, $channelNames)
->then(function ($counts) use ($channels, $attributes) {
$channels = $channels->map(function ($channel) use ($counts, $attributes) {
$info = new stdClass;
$channelName = $channel instanceof Channel
? $channel->getName()
: $channel;
if (in_array('user_count', $attributes)) {
$info->user_count = $counts[$channelName];
}
return $info;
})->sortBy(function ($content, $name) {
return $name;
})->all();
return [
'channels' => $channels ?: new stdClass,
];
});
}); });
if ($request->has('filter_by_prefix')) {
$channels = $channels->filter(function ($channel, $channelName) use ($request) {
return Str::startsWith($channelName, $request->filter_by_prefix);
});
}
$channelNames = $channels->map(function ($channel) {
return $channel instanceof Channel
? $channel->getName()
: $channel;
})->toArray();
return $this->channelManager
->getChannelsMembersCount($request->appId, $channelNames)
->then(function ($counts) use ($channels, $attributes) {
$channels = $channels->map(function ($channel) use ($counts, $attributes) {
$info = new stdClass;
$channelName = $channel instanceof Channel
? $channel->getName()
: $channel;
if (in_array('user_count', $attributes)) {
$info->user_count = $counts[$channelName];
}
return $info;
})->sortBy(function ($content, $name) {
return $name;
})->all();
return [
'channels' => $channels ?: new stdClass,
];
});
});
} }
} }

66
src/ChannelManagers/LocalChannelManager.php
View File

@ -163,23 +163,21 @@ class LocalChannelManager implements ChannelManager
return Helpers::createFulfilledPromise(false); return Helpers::createFulfilledPromise(false);
} }
$this->getLocalChannels($connection->app->id) $this->getLocalChannels($connection->app->id)->then(function ($channels) use ($connection) {
->then(function ($channels) use ($connection) { collect($channels)->each->unsubscribe($connection);
collect($channels)->each->unsubscribe($connection);
collect($channels) collect($channels)
->reject->hasConnections() ->reject->hasConnections()
->each(function (Channel $channel, string $channelName) use ($connection) { ->each(function (Channel $channel, string $channelName) use ($connection) {
unset($this->channels[$connection->app->id][$channelName]); unset($this->channels[$connection->app->id][$channelName]);
}); });
}); });
$this->getLocalChannels($connection->app->id) $this->getLocalChannels($connection->app->id)->then(function ($channels) use ($connection) {
->then(function ($channels) use ($connection) { if (count($channels) === 0) {
if (count($channels) === 0) { unset($this->channels[$connection->app->id]);
unset($this->channels[$connection->app->id]); }
} });
});
return Helpers::createFulfilledPromise(true); return Helpers::createFulfilledPromise(true);
} }
@ -252,18 +250,17 @@ class LocalChannelManager implements ChannelManager
*/ */
public function getLocalConnectionsCount($appId, string $channelName = null): PromiseInterface public function getLocalConnectionsCount($appId, string $channelName = null): PromiseInterface
{ {
return $this->getLocalChannels($appId) return $this->getLocalChannels($appId)->then(function ($channels) use ($channelName) {
->then(function ($channels) use ($channelName) { return collect($channels)->when(! is_null($channelName), function ($collection) use ($channelName) {
return collect($channels)->when(! is_null($channelName), function ($collection) use ($channelName) { return $collection->filter(function (Channel $channel) use ($channelName) {
return $collection->filter(function (Channel $channel) use ($channelName) { return $channel->getName() === $channelName;
return $channel->getName() === $channelName; });
}); })
}) ->flatMap(function (Channel $channel) {
->flatMap(function (Channel $channel) { return collect($channel->getConnections())->pluck('socketId');
return collect($channel->getConnections())->pluck('socketId'); })
}) ->unique()->count();
->unique()->count(); });
});
} }
/** /**
@ -455,16 +452,15 @@ class LocalChannelManager implements ChannelManager
*/ */
public function updateConnectionInChannels($connection): PromiseInterface public function updateConnectionInChannels($connection): PromiseInterface
{ {
return $this->getLocalChannels($connection->app->id) return $this->getLocalChannels($connection->app->id)->then(function ($channels) use ($connection) {
->then(function ($channels) use ($connection) { foreach ($channels as $channel) {
foreach ($channels as $channel) { if ($channel->hasConnection($connection)) {
if ($channel->hasConnection($connection)) { $channel->saveConnection($connection);
$channel->saveConnection($connection);
}
} }
}
return true; return true;
}); });
} }
/** /**

81
src/ChannelManagers/RedisChannelManager.php
View File

@ -137,15 +137,13 @@ class RedisChannelManager extends LocalChannelManager
*/ */
public function unsubscribeFromAllChannels(ConnectionInterface $connection): PromiseInterface public function unsubscribeFromAllChannels(ConnectionInterface $connection): PromiseInterface
{ {
return $this->getGlobalChannels($connection->app->id) return $this->getGlobalChannels($connection->app->id)->then(function ($channels) use ($connection) {
->then(function ($channels) use ($connection) { foreach ($channels as $channel) {
foreach ($channels as $channel) { $this->unsubscribeFromChannel($connection, $channel, new stdClass);
$this->unsubscribeFromChannel($connection, $channel, new stdClass); }
} })->then(function () use ($connection) {
}) return parent::unsubscribeFromAllChannels($connection);
->then(function () use ($connection) { });
return parent::unsubscribeFromAllChannels($connection);
});
} }
/** /**
@ -158,19 +156,15 @@ class RedisChannelManager extends LocalChannelManager
*/ */
public function subscribeToChannel(ConnectionInterface $connection, string $channelName, stdClass $payload): PromiseInterface public function subscribeToChannel(ConnectionInterface $connection, string $channelName, stdClass $payload): PromiseInterface
{ {
return $this->subscribeToTopic($connection->app->id, $channelName) return $this->subscribeToTopic($connection->app->id, $channelName)->then(function () use ($connection) {
->then(function () use ($connection) { return $this->addConnectionToSet($connection, Carbon::now());
return $this->addConnectionToSet($connection, Carbon::now()); })->then(function () use ($connection, $channelName) {
}) return $this->addChannelToSet($connection->app->id, $channelName);
->then(function () use ($connection, $channelName) { })->then(function () use ($connection, $channelName) {
return $this->addChannelToSet($connection->app->id, $channelName); return $this->incrementSubscriptionsCount($connection->app->id, $channelName, 1);
}) })->then(function () use ($connection, $channelName, $payload) {
->then(function () use ($connection, $channelName) { return parent::subscribeToChannel($connection, $channelName, $payload);
return $this->incrementSubscriptionsCount($connection->app->id, $channelName, 1); });
})
->then(function () use ($connection, $channelName, $payload) {
return parent::subscribeToChannel($connection, $channelName, $payload);
});
} }
/** /**
@ -199,14 +193,11 @@ class RedisChannelManager extends LocalChannelManager
$this->unsubscribeFromTopic($connection->app->id, $channelName); $this->unsubscribeFromTopic($connection->app->id, $channelName);
} }
}); });
}) })->then(function () use ($connection, $channelName) {
->then(function () use ($connection, $channelName) {
return $this->removeChannelFromSet($connection->app->id, $channelName); return $this->removeChannelFromSet($connection->app->id, $channelName);
}) })->then(function () use ($connection) {
->then(function () use ($connection) {
return $this->removeConnectionFromSet($connection); return $this->removeConnectionFromSet($connection);
}) })->then(function () use ($connection, $channelName, $payload) {
->then(function () use ($connection, $channelName, $payload) {
return parent::unsubscribeFromChannel($connection, $channelName, $payload); return parent::unsubscribeFromChannel($connection, $channelName, $payload);
}); });
} }
@ -220,10 +211,9 @@ class RedisChannelManager extends LocalChannelManager
*/ */
public function subscribeToApp($appId): PromiseInterface public function subscribeToApp($appId): PromiseInterface
{ {
return $this->subscribeToTopic($appId) return $this->subscribeToTopic($appId)->then(function () use ($appId) {
->then(function () use ($appId) { return $this->incrementSubscriptionsCount($appId);
return $this->incrementSubscriptionsCount($appId); });
});
} }
/** /**
@ -235,10 +225,9 @@ class RedisChannelManager extends LocalChannelManager
*/ */
public function unsubscribeFromApp($appId): PromiseInterface public function unsubscribeFromApp($appId): PromiseInterface
{ {
return $this->unsubscribeFromTopic($appId) return $this->unsubscribeFromTopic($appId)->then(function () use ($appId) {
->then(function () use ($appId) { return $this->decrementSubscriptionsCount($appId);
return $this->decrementSubscriptionsCount($appId); });
});
} }
/** /**
@ -308,8 +297,7 @@ class RedisChannelManager extends LocalChannelManager
return $this->storeUserData($connection->app->id, $channel, $connection->socketId, json_encode($user)) return $this->storeUserData($connection->app->id, $channel, $connection->socketId, json_encode($user))
->then(function () use ($connection, $channel, $user) { ->then(function () use ($connection, $channel, $user) {
return $this->addUserSocket($connection->app->id, $channel, $user, $connection->socketId); return $this->addUserSocket($connection->app->id, $channel, $user, $connection->socketId);
}) })->then(function () use ($connection, $user, $channel, $payload) {
->then(function () use ($connection, $user, $channel, $payload) {
return parent::userJoinedPresenceChannel($connection, $user, $channel, $payload); return parent::userJoinedPresenceChannel($connection, $user, $channel, $payload);
}); });
} }
@ -328,8 +316,7 @@ class RedisChannelManager extends LocalChannelManager
return $this->removeUserData($connection->app->id, $channel, $connection->socketId) return $this->removeUserData($connection->app->id, $channel, $connection->socketId)
->then(function () use ($connection, $channel, $user) { ->then(function () use ($connection, $channel, $user) {
return $this->removeUserSocket($connection->app->id, $channel, $user, $connection->socketId); return $this->removeUserSocket($connection->app->id, $channel, $user, $connection->socketId);
}) })->then(function () use ($connection, $user, $channel) {
->then(function () use ($connection, $user, $channel) {
return parent::userLeftPresenceChannel($connection, $user, $channel); return parent::userLeftPresenceChannel($connection, $user, $channel);
}); });
} }
@ -383,10 +370,9 @@ class RedisChannelManager extends LocalChannelManager
); );
} }
return $this->publishClient->exec() return $this->publishClient->exec()->then(function ($data) use ($channelNames) {
->then(function ($data) use ($channelNames) { return array_combine($channelNames, $data);
return array_combine($channelNames, $data); });
});
} }
/** /**
@ -413,10 +399,9 @@ class RedisChannelManager extends LocalChannelManager
public function connectionPonged(ConnectionInterface $connection): PromiseInterface public function connectionPonged(ConnectionInterface $connection): PromiseInterface
{ {
// This will update the score with the current timestamp. // This will update the score with the current timestamp.
return $this->addConnectionToSet($connection, Carbon::now()) return $this->addConnectionToSet($connection, Carbon::now())->then(function () use ($connection) {
->then(function () use ($connection) { return parent::connectionPonged($connection);
return parent::connectionPonged($connection); });
});
} }
/** /**

82
src/Channels/PresenceChannel.php
View File

@ -54,8 +54,7 @@ class PresenceChannel extends PrivateChannel
]), ]),
])); ]));
}); });
}) })->then(function () use ($connection, $user, $payload) {
->then(function () use ($connection, $user, $payload) {
// The `pusher_internal:member_added` event is triggered when a user joins a channel. // The `pusher_internal:member_added` event is triggered when a user joins a channel.
// It's quite possible that a user can have multiple connections to the same channel // It's quite possible that a user can have multiple connections to the same channel
// (for example by having multiple browser tabs open) // (for example by having multiple browser tabs open)
@ -104,50 +103,47 @@ class PresenceChannel extends PrivateChannel
{ {
$truth = parent::unsubscribe($connection); $truth = parent::unsubscribe($connection);
$this->channelManager $this->channelManager->getChannelMember($connection, $this->getName())->then(function ($user) {
->getChannelMember($connection, $this->getName()) return @json_decode($user);
->then(function ($user) { })->then(function ($user) use ($connection) {
return @json_decode($user); if (! $user) {
}) return;
->then(function ($user) use ($connection) { }
if (! $user) {
return;
}
$this->channelManager $this->channelManager
->userLeftPresenceChannel($connection, $user, $this->getName()) ->userLeftPresenceChannel($connection, $user, $this->getName())
->then(function () use ($connection, $user) { ->then(function () use ($connection, $user) {
// The `pusher_internal:member_removed` is triggered when a user leaves a channel. // The `pusher_internal:member_removed` is triggered when a user leaves a channel.
// It's quite possible that a user can have multiple connections to the same channel // It's quite possible that a user can have multiple connections to the same channel
// (for example by having multiple browser tabs open) // (for example by having multiple browser tabs open)
// and in this case the events will only be triggered when the last one is closed. // and in this case the events will only be triggered when the last one is closed.
$this->channelManager $this->channelManager
->getMemberSockets($user->user_id, $connection->app->id, $this->getName()) ->getMemberSockets($user->user_id, $connection->app->id, $this->getName())
->then(function ($sockets) use ($connection, $user) { ->then(function ($sockets) use ($connection, $user) {
if (count($sockets) === 0) { if (count($sockets) === 0) {
$memberRemovedPayload = [ $memberRemovedPayload = [
'event' => 'pusher_internal:member_removed', 'event' => 'pusher_internal:member_removed',
'channel' => $this->getName(), 'channel' => $this->getName(),
'data' => json_encode([ 'data' => json_encode([
'user_id' => $user->user_id, 'user_id' => $user->user_id,
]), ]),
]; ];
$this->broadcastToEveryoneExcept( $this->broadcastToEveryoneExcept(
(object) $memberRemovedPayload, $connection->socketId, (object) $memberRemovedPayload, $connection->socketId,
$connection->app->id $connection->app->id
); );
UnsubscribedFromChannel::dispatch( UnsubscribedFromChannel::dispatch(
$connection->app->id, $connection->app->id,
$connection->socketId, $connection->socketId,
$this->getName(), $this->getName(),
$user $user
); );
} }
}); });
}); });
}); });
return $truth; return $truth;
} }

16
src/Console/Commands/StartServer.php
View File

@ -304,14 +304,12 @@ class StartServer extends Command
// Get all local connections and close them. They will // Get all local connections and close them. They will
// be automatically be unsubscribed from all channels. // be automatically be unsubscribed from all channels.
$channelManager->getLocalConnections() $channelManager->getLocalConnections()->then(function ($connections) {
->then(function ($connections) { foreach ($connections as $connection) {
foreach ($connections as $connection) { $connection->close();
$connection->close(); }
} })->then(function () {
}) $this->loop->stop();
->then(function () { });
$this->loop->stop();
});
} }
} }

10
src/Server/Messages/PusherChannelProtocolMessage.php
View File

@ -32,13 +32,11 @@ class PusherChannelProtocolMessage extends PusherClientMessage
*/ */
protected function ping(ConnectionInterface $connection) protected function ping(ConnectionInterface $connection)
{ {
$this->channelManager $this->channelManager->connectionPonged($connection)->then(function () use ($connection) {
->connectionPonged($connection) $connection->send(json_encode(['event' => 'pusher:pong']));
->then(function () use ($connection) {
$connection->send(json_encode(['event' => 'pusher:pong']));
ConnectionPonged::dispatch($connection->app->id, $connection->socketId); ConnectionPonged::dispatch($connection->app->id, $connection->socketId);
}); });
} }
/** /**

12
src/Statistics/Collectors/MemoryCollector.php
View File

@ -98,13 +98,11 @@ class MemoryCollector implements StatisticsCollector
$this->createRecord($statistic, $appId); $this->createRecord($statistic, $appId);
$this->channelManager $this->channelManager->getGlobalConnectionsCount($appId)->then(function ($connections) use ($statistic) {
->getGlobalConnectionsCount($appId) $statistic->reset(
->then(function ($connections) use ($statistic) { is_null($connections) ? 0 : $connections
$statistic->reset( );
is_null($connections) ? 0 : $connections });
);
});
} }
}); });
} }

216
src/Statistics/Collectors/RedisCollector.php
View File

@ -84,30 +84,24 @@ class RedisCollector extends MemoryCollector
->hincrby( ->hincrby(
$this->channelManager->getRedisKey($appId, null, ['stats']), $this->channelManager->getRedisKey($appId, null, ['stats']),
'current_connections_count', 1 'current_connections_count', 1
) )->then(function ($currentConnectionsCount) use ($appId) {
->then(function ($currentConnectionsCount) use ($appId) {
// Get the peak connections count from Redis. // Get the peak connections count from Redis.
$this->channelManager $this->channelManager->getPublishClient()->hget(
->getPublishClient() $this->channelManager->getRedisKey($appId, null, ['stats']),
->hget( 'peak_connections_count'
$this->channelManager->getRedisKey($appId, null, ['stats']), )->then(function ($currentPeakConnectionCount) use ($currentConnectionsCount, $appId) {
'peak_connections_count' // Extract the greatest number between the current peak connection count
) // and the current connection number.
->then(function ($currentPeakConnectionCount) use ($currentConnectionsCount, $appId) { $peakConnectionsCount = is_null($currentPeakConnectionCount)
// Extract the greatest number between the current peak connection count ? $currentConnectionsCount
// and the current connection number. : max($currentPeakConnectionCount, $currentConnectionsCount);
$peakConnectionsCount = is_null($currentPeakConnectionCount)
? $currentConnectionsCount
: max($currentPeakConnectionCount, $currentConnectionsCount);
// Then set it to the database. // Then set it to the database.
$this->channelManager $this->channelManager->getPublishClient()->hset(
->getPublishClient() $this->channelManager->getRedisKey($appId, null, ['stats']),
->hset( 'peak_connections_count', $peakConnectionsCount
$this->channelManager->getRedisKey($appId, null, ['stats']), );
'peak_connections_count', $peakConnectionsCount });
);
});
}); });
} }
@ -135,12 +129,10 @@ class RedisCollector extends MemoryCollector
: max($currentPeakConnectionCount, $currentConnectionsCount); : max($currentPeakConnectionCount, $currentConnectionsCount);
// Then set it to the database. // Then set it to the database.
$this->channelManager $this->channelManager->getPublishClient()->hset(
->getPublishClient() $this->channelManager->getRedisKey($appId, null, ['stats']),
->hset( 'peak_connections_count', $peakConnectionsCount
$this->channelManager->getRedisKey($appId, null, ['stats']), );
'peak_connections_count', $peakConnectionsCount
);
}); });
}); });
} }
@ -153,35 +145,32 @@ class RedisCollector extends MemoryCollector
public function save() public function save()
{ {
$this->lock()->get(function () { $this->lock()->get(function () {
$this->channelManager $this->channelManager->getPublishClient()->smembers(static::$redisSetName)->then(function ($members) {
->getPublishClient() foreach ($members as $appId) {
->smembers(static::$redisSetName) $this->channelManager
->then(function ($members) { ->getPublishClient()
foreach ($members as $appId) { ->hgetall($this->channelManager->getRedisKey($appId, null, ['stats']))
$this->channelManager ->then(function ($list) use ($appId) {
->getPublishClient() if (! $list) {
->hgetall($this->channelManager->getRedisKey($appId, null, ['stats'])) return;
->then(function ($list) use ($appId) { }
if (! $list) {
return;
}
$statistic = $this->arrayToStatisticInstance( $statistic = $this->arrayToStatisticInstance(
$appId, Helpers::redisListToArray($list) $appId, Helpers::redisListToArray($list)
); );
$this->createRecord($statistic, $appId); $this->createRecord($statistic, $appId);
$this->channelManager $this->channelManager
->getGlobalConnectionsCount($appId) ->getGlobalConnectionsCount($appId)
->then(function ($currentConnectionsCount) use ($appId) { ->then(function ($currentConnectionsCount) use ($appId) {
$currentConnectionsCount === 0 || is_null($currentConnectionsCount) $currentConnectionsCount === 0 || is_null($currentConnectionsCount)
? $this->resetAppTraces($appId) ? $this->resetAppTraces($appId)
: $this->resetStatistics($appId, $currentConnectionsCount); : $this->resetStatistics($appId, $currentConnectionsCount);
}); });
}); });
} }
}); });
}); });
} }
@ -206,25 +195,22 @@ class RedisCollector extends MemoryCollector
*/ */
public function getStatistics(): PromiseInterface public function getStatistics(): PromiseInterface
{ {
return $this->channelManager return $this->channelManager->getPublishClient()->smembers(static::$redisSetName)->then(function ($members) {
->getPublishClient() $appsWithStatistics = [];
->smembers(static::$redisSetName)
->then(function ($members) {
$appsWithStatistics = [];
foreach ($members as $appId) { foreach ($members as $appId) {
$this->channelManager $this->channelManager
->getPublishClient() ->getPublishClient()
->hgetall($this->channelManager->getRedisKey($appId, null, ['stats'])) ->hgetall($this->channelManager->getRedisKey($appId, null, ['stats']))
->then(function ($list) use ($appId, &$appsWithStatistics) { ->then(function ($list) use ($appId, &$appsWithStatistics) {
$appsWithStatistics[$appId] = $this->arrayToStatisticInstance( $appsWithStatistics[$appId] = $this->arrayToStatisticInstance(
$appId, Helpers::redisListToArray($list) $appId, Helpers::redisListToArray($list)
); );
}); });
} }
return $appsWithStatistics; return $appsWithStatistics;
}); });
} }
/** /**
@ -254,33 +240,25 @@ class RedisCollector extends MemoryCollector
*/ */
public function resetStatistics($appId, int $currentConnectionCount) public function resetStatistics($appId, int $currentConnectionCount)
{ {
$this->channelManager $this->channelManager->getPublishClient()->hset(
->getPublishClient() $this->channelManager->getRedisKey($appId, null, ['stats']),
->hset( 'current_connections_count', $currentConnectionCount
$this->channelManager->getRedisKey($appId, null, ['stats']), );
'current_connections_count', $currentConnectionCount
);
$this->channelManager $this->channelManager->getPublishClient()->hset(
->getPublishClient() $this->channelManager->getRedisKey($appId, null, ['stats']),
->hset( 'peak_connections_count', $currentConnectionCount
$this->channelManager->getRedisKey($appId, null, ['stats']), );
'peak_connections_count', $currentConnectionCount
);
$this->channelManager $this->channelManager->getPublishClient()->hset(
->getPublishClient() $this->channelManager->getRedisKey($appId, null, ['stats']),
->hset( 'websocket_messages_count', 0
$this->channelManager->getRedisKey($appId, null, ['stats']), );
'websocket_messages_count', 0
);
$this->channelManager $this->channelManager->getPublishClient()->hset(
->getPublishClient() $this->channelManager->getRedisKey($appId, null, ['stats']),
->hset( 'api_messages_count', 0
$this->channelManager->getRedisKey($appId, null, ['stats']), );
'api_messages_count', 0
);
} }
/** /**
@ -292,37 +270,27 @@ class RedisCollector extends MemoryCollector
*/ */
public function resetAppTraces($appId) public function resetAppTraces($appId)
{ {
$this->channelManager $this->channelManager->getPublishClient()->hdel(
->getPublishClient() $this->channelManager->getRedisKey($appId, null, ['stats']),
->hdel( 'current_connections_count'
$this->channelManager->getRedisKey($appId, null, ['stats']), );
'current_connections_count'
);
$this->channelManager $this->channelManager->getPublishClient()->hdel(
->getPublishClient() $this->channelManager->getRedisKey($appId, null, ['stats']),
->hdel( 'peak_connections_count'
$this->channelManager->getRedisKey($appId, null, ['stats']), );
'peak_connections_count'
);
$this->channelManager $this->channelManager->getPublishClient()->hdel(
->getPublishClient() $this->channelManager->getRedisKey($appId, null, ['stats']),
->hdel( 'websocket_messages_count'
$this->channelManager->getRedisKey($appId, null, ['stats']), );
'websocket_messages_count'
);
$this->channelManager $this->channelManager->getPublishClient()->hdel(
->getPublishClient() $this->channelManager->getRedisKey($appId, null, ['stats']),
->hdel( 'api_messages_count'
$this->channelManager->getRedisKey($appId, null, ['stats']), );
'api_messages_count'
);
$this->channelManager $this->channelManager->getPublishClient()->srem(static::$redisSetName, $appId);
->getPublishClient()
->srem(static::$redisSetName, $appId);
} }
/** /**
@ -333,9 +301,7 @@ class RedisCollector extends MemoryCollector
*/ */
protected function ensureAppIsInSet($appId) protected function ensureAppIsInSet($appId)
{ {
$this->channelManager $this->channelManager->getPublishClient()->sadd(static::$redisSetName, $appId);
->getPublishClient()
->sadd(static::$redisSetName, $appId);
return $this->channelManager->getPublishClient(); return $this->channelManager->getPublishClient();
} }

17
tests/AsyncRedisQueueTest.php
View File

@ -62,11 +62,9 @@ class AsyncRedisQueueTest extends TestCase
$this->queue->later(-300, $jobs[2]); $this->queue->later(-300, $jobs[2]);
$this->queue->later(-100, $jobs[3]); $this->queue->later(-100, $jobs[3]);
$this->getPublishClient() $this->getPublishClient()->zcard('queues:default:delayed')->then(function ($count) {
->zcard('queues:default:delayed') $this->assertEquals(4, $count);
->then(function ($count) { });
$this->assertEquals(4, $count);
});
$this->unregisterManagers(); $this->unregisterManagers();
@ -87,8 +85,7 @@ class AsyncRedisQueueTest extends TestCase
$this->unregisterManagers(); $this->unregisterManagers();
$this->getPublishClient() $this->getPublishClient()->assertCalledCount(1, 'eval');
->assertCalledCount(1, 'eval');
$redisJob = $this->queue->pop(); $redisJob = $this->queue->pop();
@ -126,8 +123,7 @@ class AsyncRedisQueueTest extends TestCase
$this->unregisterManagers(); $this->unregisterManagers();
$this->getPublishClient() $this->getPublishClient()->assertCalledCount(1, 'eval');
->assertCalledCount(1, 'eval');
$redisJob = $this->queue->pop(); $redisJob = $this->queue->pop();
@ -152,8 +148,7 @@ class AsyncRedisQueueTest extends TestCase
$this->queue->push($job1); $this->queue->push($job1);
$this->queue->push($job2); $this->queue->push($job2);
$this->getPublishClient() $this->getPublishClient()->assertCalledCount(2, 'eval');
->assertCalledCount(2, 'eval');
$this->unregisterManagers(); $this->unregisterManagers();

32
tests/ConnectionTest.php
View File

@ -54,31 +54,23 @@ class ConnectionTest extends TestCase
{ {
$connection = $this->newActiveConnection(['public-channel']); $connection = $this->newActiveConnection(['public-channel']);
$this->channelManager $this->channelManager->getGlobalChannels('1234')->then(function ($channels) {
->getGlobalChannels('1234') $this->assertCount(1, $channels);
->then(function ($channels) { });
$this->assertCount(1, $channels);
});
$this->channelManager $this->channelManager->getGlobalConnectionsCount('1234')->then(function ($total) {
->getGlobalConnectionsCount('1234') $this->assertEquals(1, $total);
->then(function ($total) { });
$this->assertEquals(1, $total);
});
$this->pusherServer->onClose($connection); $this->pusherServer->onClose($connection);
$this->channelManager $this->channelManager->getGlobalConnectionsCount('1234')->then(function ($total) {
->getGlobalConnectionsCount('1234') $this->assertEquals(0, $total);
->then(function ($total) { });
$this->assertEquals(0, $total);
});
$this->channelManager $this->channelManager->getGlobalChannels('1234')->then(function ($channels) {
->getGlobalChannels('1234') $this->assertCount(0, $channels);
->then(function ($channels) { });
$this->assertCount(0, $channels);
});
} }
public function test_websocket_exceptions_are_sent() public function test_websocket_exceptions_are_sent()

94
tests/LocalPongRemovalTest.php
View File

@ -20,27 +20,21 @@ class LocalPongRemovalTest extends TestCase
$this->channelManager->updateConnectionInChannels($activeConnection); $this->channelManager->updateConnectionInChannels($activeConnection);
$this->channelManager->updateConnectionInChannels($obsoleteConnection); $this->channelManager->updateConnectionInChannels($obsoleteConnection);
$this->channelManager $this->channelManager->getGlobalConnectionsCount('1234', 'public-channel')->then(function ($count) {
->getGlobalConnectionsCount('1234', 'public-channel') $this->assertEquals(2, $count);
->then(function ($count) { });
$this->assertEquals(2, $count);
});
$this->channelManager->removeObsoleteConnections(); $this->channelManager->removeObsoleteConnections();
$this->channelManager $this->channelManager->getGlobalConnectionsCount('1234', 'public-channel')->then(function ($count) {
->getGlobalConnectionsCount('1234', 'public-channel') $this->assertEquals(1, $count);
->then(function ($count) { });
$this->assertEquals(1, $count);
});
$this->channelManager $this->channelManager->getLocalConnections()->then(function ($connections) use ($activeConnection) {
->getLocalConnections() $connection = $connections[$activeConnection->socketId];
->then(function ($connections) use ($activeConnection) {
$connection = $connections[$activeConnection->socketId];
$this->assertEquals($activeConnection->socketId, $connection->socketId); $this->assertEquals($activeConnection->socketId, $connection->socketId);
}); });
} }
public function test_not_ponged_connections_do_get_removed_on_local_for_private_channels() public function test_not_ponged_connections_do_get_removed_on_local_for_private_channels()
@ -57,27 +51,21 @@ class LocalPongRemovalTest extends TestCase
$this->channelManager->updateConnectionInChannels($activeConnection); $this->channelManager->updateConnectionInChannels($activeConnection);
$this->channelManager->updateConnectionInChannels($obsoleteConnection); $this->channelManager->updateConnectionInChannels($obsoleteConnection);
$this->channelManager $this->channelManager->getGlobalConnectionsCount('1234', 'private-channel')->then(function ($count) {
->getGlobalConnectionsCount('1234', 'private-channel') $this->assertEquals(2, $count);
->then(function ($count) { });
$this->assertEquals(2, $count);
});
$this->channelManager->removeObsoleteConnections(); $this->channelManager->removeObsoleteConnections();
$this->channelManager $this->channelManager->getGlobalConnectionsCount('1234', 'private-channel')->then(function ($count) {
->getGlobalConnectionsCount('1234', 'private-channel') $this->assertEquals(1, $count);
->then(function ($count) { });
$this->assertEquals(1, $count);
});
$this->channelManager $this->channelManager->getLocalConnections()->then(function ($connections) use ($activeConnection) {
->getLocalConnections() $connection = $connections[$activeConnection->socketId];
->then(function ($connections) use ($activeConnection) {
$connection = $connections[$activeConnection->socketId];
$this->assertEquals($activeConnection->socketId, $connection->socketId); $this->assertEquals($activeConnection->socketId, $connection->socketId);
}); });
} }
public function test_not_ponged_connections_do_get_removed_on_local_for_presence_channels() public function test_not_ponged_connections_do_get_removed_on_local_for_presence_channels()
@ -94,38 +82,28 @@ class LocalPongRemovalTest extends TestCase
$this->channelManager->updateConnectionInChannels($activeConnection); $this->channelManager->updateConnectionInChannels($activeConnection);
$this->channelManager->updateConnectionInChannels($obsoleteConnection); $this->channelManager->updateConnectionInChannels($obsoleteConnection);
$this->channelManager $this->channelManager->getGlobalConnectionsCount('1234', 'presence-channel')->then(function ($count) {
->getGlobalConnectionsCount('1234', 'presence-channel') $this->assertEquals(2, $count);
->then(function ($count) { });
$this->assertEquals(2, $count);
});
$this->channelManager $this->channelManager->getChannelMembers('1234', 'presence-channel')->then(function ($members) {
->getChannelMembers('1234', 'presence-channel') $this->assertCount(2, $members);
->then(function ($members) { });
$this->assertCount(2, $members);
});
$this->channelManager->removeObsoleteConnections(); $this->channelManager->removeObsoleteConnections();
$this->channelManager $this->channelManager->getGlobalConnectionsCount('1234', 'presence-channel')->then(function ($count) {
->getGlobalConnectionsCount('1234', 'presence-channel') $this->assertEquals(1, $count);
->then(function ($count) { });
$this->assertEquals(1, $count);
});
$this->channelManager $this->channelManager->getLocalConnections()->then(function ($connections) use ($activeConnection) {
->getLocalConnections() $connection = $connections[$activeConnection->socketId];
->then(function ($connections) use ($activeConnection) {
$connection = $connections[$activeConnection->socketId];
$this->assertEquals($activeConnection->socketId, $connection->socketId); $this->assertEquals($activeConnection->socketId, $connection->socketId);
}); });
$this->channelManager $this->channelManager->getChannelMembers('1234', 'presence-channel')->then(function ($members) {
->getChannelMembers('1234', 'presence-channel') $this->assertCount(1, $members);
->then(function ($members) { });
$this->assertCount(1, $members);
});
} }
} }

203
tests/PresenceChannelTest.php
View File

@ -58,11 +58,9 @@ class PresenceChannelTest extends TestCase
'channel' => 'presence-channel', 'channel' => 'presence-channel',
]); ]);
$this->channelManager $this->channelManager->getGlobalConnectionsCount('1234', 'presence-channel')->then(function ($total) {
->getGlobalConnectionsCount('1234', 'presence-channel') $this->assertEquals(1, $total);
->then(function ($total) { });
$this->assertEquals(1, $total);
});
} }
public function test_connect_to_presence_channel_when_user_with_same_ids_is_already_joined() public function test_connect_to_presence_channel_when_user_with_same_ids_is_already_joined()
@ -112,17 +110,13 @@ class PresenceChannelTest extends TestCase
]), ]),
]); ]);
$this->channelManager $this->channelManager->getGlobalConnectionsCount('1234', 'presence-channel')->then(function ($total) {
->getGlobalConnectionsCount('1234', 'presence-channel') $this->assertEquals(3, $total);
->then(function ($total) { });
$this->assertEquals(3, $total);
});
$this->channelManager $this->channelManager->getChannelMembers('1234', 'presence-channel')->then(function ($members) {
->getChannelMembers('1234', 'presence-channel') $this->assertCount(2, $members);
->then(function ($members) { });
$this->assertCount(2, $members);
});
} }
public function test_presence_channel_broadcast_member_events() public function test_presence_channel_broadcast_member_events()
@ -135,11 +129,9 @@ class PresenceChannelTest extends TestCase
'data' => json_encode(['user_id' => 2]), 'data' => json_encode(['user_id' => 2]),
]); ]);
$this->channelManager $this->channelManager->getChannelMembers('1234', 'presence-channel')->then(function ($members) {
->getChannelMembers('1234', 'presence-channel') $this->assertCount(2, $members);
->then(function ($members) { });
$this->assertCount(2, $members);
});
$this->pusherServer->onClose($morty); $this->pusherServer->onClose($morty);
@ -148,29 +140,23 @@ class PresenceChannelTest extends TestCase
'data' => json_encode(['user_id' => 2]), 'data' => json_encode(['user_id' => 2]),
]); ]);
$this->channelManager $this->channelManager->getGlobalConnectionsCount('1234', 'presence-channel')->then(function ($total) {
->getGlobalConnectionsCount('1234', 'presence-channel') $this->assertEquals(1, $total);
->then(function ($total) { });
$this->assertEquals(1, $total);
});
$this->channelManager $this->channelManager->getChannelMembers('1234', 'presence-channel')->then(function ($members) use ($rick) {
->getChannelMembers('1234', 'presence-channel') $this->assertCount(1, $members);
->then(function ($members) use ($rick) { $this->assertEquals(1, $members[$rick->socketId]->user_id);
$this->assertCount(1, $members); });
$this->assertEquals(1, $members[$rick->socketId]->user_id);
});
} }
public function test_unsubscribe_from_presence_channel() public function test_unsubscribe_from_presence_channel()
{ {
$connection = $this->newPresenceConnection('presence-channel', ['user_id' => 1]); $connection = $this->newPresenceConnection('presence-channel', ['user_id' => 1]);
$this->channelManager $this->channelManager->getGlobalConnectionsCount('1234', 'presence-channel')->then(function ($total) {
->getGlobalConnectionsCount('1234', 'presence-channel') $this->assertEquals(1, $total);
->then(function ($total) { });
$this->assertEquals(1, $total);
});
$message = new Mocks\Message([ $message = new Mocks\Message([
'event' => 'pusher:unsubscribe', 'event' => 'pusher:unsubscribe',
@ -181,12 +167,10 @@ class PresenceChannelTest extends TestCase
$this->pusherServer->onMessage($connection, $message); $this->pusherServer->onMessage($connection, $message);
$this->channelManager $this->channelManager->getGlobalConnectionsCount('1234', 'presence-channel')->then(function ($total) {
->getGlobalConnectionsCount('1234', 'presence-channel') $this->assertEquals(0, $total);
->then(function ($total) { });
$this->assertEquals(0, $total); }
});
}
public function test_can_whisper_to_private_channel() public function test_can_whisper_to_private_channel()
{ {
@ -229,22 +213,18 @@ class PresenceChannelTest extends TestCase
$rick = $this->newPresenceConnection('presence-channel', ['user_id' => 1]); $rick = $this->newPresenceConnection('presence-channel', ['user_id' => 1]);
$morty = $this->newPresenceConnection('presence-channel', ['user_id' => 2]); $morty = $this->newPresenceConnection('presence-channel', ['user_id' => 2]);
$this->statisticsCollector $this->statisticsCollector->getStatistics()->then(function ($statistics) {
->getStatistics() $this->assertCount(1, $statistics);
->then(function ($statistics) { });
$this->assertCount(1, $statistics);
});
$this->statisticsCollector $this->statisticsCollector->getAppStatistics('1234')->then(function ($statistic) {
->getAppStatistics('1234') $this->assertEquals([
->then(function ($statistic) { 'peak_connections_count' => 2,
$this->assertEquals([ 'websocket_messages_count' => 2,
'peak_connections_count' => 2, 'api_messages_count' => 0,
'websocket_messages_count' => 2, 'app_id' => '1234',
'api_messages_count' => 0, ], $statistic->toArray());
'app_id' => '1234', });
], $statistic->toArray());
});
} }
public function test_local_connections_for_presence_channels() public function test_local_connections_for_presence_channels()
@ -252,17 +232,15 @@ class PresenceChannelTest extends TestCase
$this->newPresenceConnection('presence-channel', ['user_id' => 1]); $this->newPresenceConnection('presence-channel', ['user_id' => 1]);
$this->newPresenceConnection('presence-channel-2', ['user_id' => 2]); $this->newPresenceConnection('presence-channel-2', ['user_id' => 2]);
$this->channelManager $this->channelManager->getLocalConnections()->then(function ($connections) {
->getLocalConnections() $this->assertCount(2, $connections);
->then(function ($connections) {
$this->assertCount(2, $connections);
foreach ($connections as $connection) { foreach ($connections as $connection) {
$this->assertInstanceOf( $this->assertInstanceOf(
ConnectionInterface::class, $connection ConnectionInterface::class, $connection
); );
} }
}); });
} }
public function test_multiple_clients_with_same_user_id_trigger_member_added_and_removed_event_only_on_first_and_last_socket_connection() public function test_multiple_clients_with_same_user_id_trigger_member_added_and_removed_event_only_on_first_and_last_socket_connection()
@ -304,17 +282,13 @@ class PresenceChannelTest extends TestCase
$this->assertCount(0, $sockets); $this->assertCount(0, $sockets);
}); });
$this->channelManager $this->channelManager->getMemberSockets('2', '1234', 'presence-channel')->then(function ($sockets) {
->getMemberSockets('2', '1234', 'presence-channel') $this->assertCount(0, $sockets);
->then(function ($sockets) { });
$this->assertCount(0, $sockets);
});
$this->channelManager $this->channelManager->getMemberSockets('observer', '1234', 'presence-channel')->then(function ($sockets) {
->getMemberSockets('observer', '1234', 'presence-channel') $this->assertCount(1, $sockets);
->then(function ($sockets) { });
$this->assertCount(1, $sockets);
});
} }
public function test_events_are_processed_by_on_message_on_presence_channels() public function test_events_are_processed_by_on_message_on_presence_channels()
@ -400,11 +374,10 @@ class PresenceChannelTest extends TestCase
$this->getSubscribeClient() $this->getSubscribeClient()
->assertNothingDispatched(); ->assertNothingDispatched();
$this->getPublishClient() $this->getPublishClient()->assertCalledWithArgs('publish', [
->assertCalledWithArgs('publish', [ $this->channelManager->getRedisKey('1234', 'presence-channel'),
$this->channelManager->getRedisKey('1234', 'presence-channel'), $message->getPayload(),
$message->getPayload(), ]);
]);
} }
public function test_it_fires_the_event_to_presence_channel() public function test_it_fires_the_event_to_presence_channel()
@ -438,16 +411,14 @@ class PresenceChannelTest extends TestCase
$this->assertSame([], json_decode($response->getContent(), true)); $this->assertSame([], json_decode($response->getContent(), true));
$this->statisticsCollector $this->statisticsCollector->getAppStatistics('1234')->then(function ($statistic) {
->getAppStatistics('1234') $this->assertEquals([
->then(function ($statistic) { 'peak_connections_count' => 1,
$this->assertEquals([ 'websocket_messages_count' => 1,
'peak_connections_count' => 1, 'api_messages_count' => 1,
'websocket_messages_count' => 1, 'app_id' => '1234',
'api_messages_count' => 1, ], $statistic->toArray());
'app_id' => '1234', });
], $statistic->toArray());
});
} }
public function test_it_fires_event_across_servers_when_there_are_not_users_locally_for_presence_channel() public function test_it_fires_event_across_servers_when_there_are_not_users_locally_for_presence_channel()
@ -480,19 +451,17 @@ class PresenceChannelTest extends TestCase
$this->assertSame([], json_decode($response->getContent(), true)); $this->assertSame([], json_decode($response->getContent(), true));
if (method_exists($this->channelManager, 'getPublishClient')) { if (method_exists($this->channelManager, 'getPublishClient')) {
$this->channelManager $this->channelManager->getPublishClient()->assertCalledWithArgsCount(1, 'publish', [
->getPublishClient() $this->channelManager->getRedisKey('1234', 'presence-channel'),
->assertCalledWithArgsCount(1, 'publish', [ json_encode([
$this->channelManager->getRedisKey('1234', 'presence-channel'), 'event' => 'some-event',
json_encode([ 'channel' => 'presence-channel',
'event' => 'some-event', 'data' => json_encode(['some-data' => 'yes']),
'channel' => 'presence-channel', 'appId' => '1234',
'data' => json_encode(['some-data' => 'yes']), 'socketId' => null,
'appId' => '1234', 'serverId' => $this->channelManager->getServerId(),
'socketId' => null, ]),
'serverId' => $this->channelManager->getServerId(), ]);
]),
]);
} }
} }
@ -528,19 +497,17 @@ class PresenceChannelTest extends TestCase
$this->assertSame([], json_decode($response->getContent(), true)); $this->assertSame([], json_decode($response->getContent(), true));
if (method_exists($this->channelManager, 'getPublishClient')) { if (method_exists($this->channelManager, 'getPublishClient')) {
$this->channelManager $this->channelManager->getPublishClient()->assertCalledWithArgsCount(1, 'publish', [
->getPublishClient() $this->channelManager->getRedisKey('1234', 'presence-channel'),
->assertCalledWithArgsCount(1, 'publish', [ json_encode([
$this->channelManager->getRedisKey('1234', 'presence-channel'), 'event' => 'some-event',
json_encode([ 'channel' => 'presence-channel',
'event' => 'some-event', 'data' => json_encode(['some-data' => 'yes']),
'channel' => 'presence-channel', 'appId' => '1234',
'data' => json_encode(['some-data' => 'yes']), 'socketId' => null,
'appId' => '1234', 'serverId' => $this->channelManager->getServerId(),
'socketId' => null, ]),
'serverId' => $this->channelManager->getServerId(), ]);
]),
]);
} }
$wsConnection->assertSentEvent('some-event', [ $wsConnection->assertSentEvent('some-event', [

143
tests/PrivateChannelTest.php
View File

@ -48,22 +48,18 @@ class PrivateChannelTest extends TestCase
'channel' => 'private-channel', 'channel' => 'private-channel',
]); ]);
$this->channelManager $this->channelManager->getGlobalConnectionsCount('1234', 'private-channel')->then(function ($total) {
->getGlobalConnectionsCount('1234', 'private-channel') $this->assertEquals(1, $total);
->then(function ($total) { });
$this->assertEquals(1, $total);
});
} }
public function test_unsubscribe_from_private_channel() public function test_unsubscribe_from_private_channel()
{ {
$connection = $this->newPrivateConnection('private-channel'); $connection = $this->newPrivateConnection('private-channel');
$this->channelManager $this->channelManager->getGlobalConnectionsCount('1234', 'private-channel')->then(function ($total) {
->getGlobalConnectionsCount('1234', 'private-channel') $this->assertEquals(1, $total);
->then(function ($total) { });
$this->assertEquals(1, $total);
});
$message = new Mocks\Message([ $message = new Mocks\Message([
'event' => 'pusher:unsubscribe', 'event' => 'pusher:unsubscribe',
@ -74,11 +70,9 @@ class PrivateChannelTest extends TestCase
$this->pusherServer->onMessage($connection, $message); $this->pusherServer->onMessage($connection, $message);
$this->channelManager $this->channelManager->getGlobalConnectionsCount('1234', 'private-channel')->then(function ($total) {
->getGlobalConnectionsCount('1234', 'private-channel') $this->assertEquals(0, $total);
->then(function ($total) { });
$this->assertEquals(0, $total);
});
} }
public function test_can_whisper_to_private_channel() public function test_can_whisper_to_private_channel()
@ -122,22 +116,18 @@ class PrivateChannelTest extends TestCase
$rick = $this->newPrivateConnection('private-channel'); $rick = $this->newPrivateConnection('private-channel');
$morty = $this->newPrivateConnection('private-channel'); $morty = $this->newPrivateConnection('private-channel');
$this->statisticsCollector $this->statisticsCollector->getStatistics()->then(function ($statistics) {
->getStatistics() $this->assertCount(1, $statistics);
->then(function ($statistics) { });
$this->assertCount(1, $statistics);
});
$this->statisticsCollector $this->statisticsCollector->getAppStatistics('1234')->then(function ($statistic) {
->getAppStatistics('1234') $this->assertEquals([
->then(function ($statistic) { 'peak_connections_count' => 2,
$this->assertEquals([ 'websocket_messages_count' => 2,
'peak_connections_count' => 2, 'api_messages_count' => 0,
'websocket_messages_count' => 2, 'app_id' => '1234',
'api_messages_count' => 0, ], $statistic->toArray());
'app_id' => '1234', });
], $statistic->toArray());
});
} }
public function test_local_connections_for_private_channels() public function test_local_connections_for_private_channels()
@ -145,17 +135,15 @@ class PrivateChannelTest extends TestCase
$this->newPrivateConnection('private-channel'); $this->newPrivateConnection('private-channel');
$this->newPrivateConnection('private-channel-2'); $this->newPrivateConnection('private-channel-2');
$this->channelManager $this->channelManager->getLocalConnections()->then(function ($connections) {
->getLocalConnections() $this->assertCount(2, $connections);
->then(function ($connections) {
$this->assertCount(2, $connections);
foreach ($connections as $connection) { foreach ($connections as $connection) {
$this->assertInstanceOf( $this->assertInstanceOf(
ConnectionInterface::class, $connection ConnectionInterface::class, $connection
); );
} }
}); });
} }
public function test_events_are_processed_by_on_message_on_private_channels() public function test_events_are_processed_by_on_message_on_private_channels()
@ -220,11 +208,10 @@ class PrivateChannelTest extends TestCase
$this->getSubscribeClient() $this->getSubscribeClient()
->assertNothingDispatched(); ->assertNothingDispatched();
$this->getPublishClient() $this->getPublishClient()->assertCalledWithArgs('publish', [
->assertCalledWithArgs('publish', [ $this->channelManager->getRedisKey('1234', 'private-channel'),
$this->channelManager->getRedisKey('1234', 'private-channel'), $message->getPayload(),
$message->getPayload(), ]);
]);
} }
public function test_it_fires_the_event_to_private_channel() public function test_it_fires_the_event_to_private_channel()
@ -258,16 +245,14 @@ class PrivateChannelTest extends TestCase
$this->assertSame([], json_decode($response->getContent(), true)); $this->assertSame([], json_decode($response->getContent(), true));
$this->statisticsCollector $this->statisticsCollector->getAppStatistics('1234')->then(function ($statistic) {
->getAppStatistics('1234') $this->assertEquals([
->then(function ($statistic) { 'peak_connections_count' => 1,
$this->assertEquals([ 'websocket_messages_count' => 1,
'peak_connections_count' => 1, 'api_messages_count' => 1,
'websocket_messages_count' => 1, 'app_id' => '1234',
'api_messages_count' => 1, ], $statistic->toArray());
'app_id' => '1234', });
], $statistic->toArray());
});
} }
public function test_it_fires_event_across_servers_when_there_are_not_users_locally_for_private_channel() public function test_it_fires_event_across_servers_when_there_are_not_users_locally_for_private_channel()
@ -300,19 +285,17 @@ class PrivateChannelTest extends TestCase
$this->assertSame([], json_decode($response->getContent(), true)); $this->assertSame([], json_decode($response->getContent(), true));
if (method_exists($this->channelManager, 'getPublishClient')) { if (method_exists($this->channelManager, 'getPublishClient')) {
$this->channelManager $this->channelManager->getPublishClient()->assertCalledWithArgsCount(1, 'publish', [
->getPublishClient() $this->channelManager->getRedisKey('1234', 'private-channel'),
->assertCalledWithArgsCount(1, 'publish', [ json_encode([
$this->channelManager->getRedisKey('1234', 'private-channel'), 'event' => 'some-event',
json_encode([ 'channel' => 'private-channel',
'event' => 'some-event', 'data' => json_encode(['some-data' => 'yes']),
'channel' => 'private-channel', 'appId' => '1234',
'data' => json_encode(['some-data' => 'yes']), 'socketId' => null,
'appId' => '1234', 'serverId' => $this->channelManager->getServerId(),
'socketId' => null, ]),
'serverId' => $this->channelManager->getServerId(), ]);
]),
]);
} }
} }
@ -348,19 +331,17 @@ class PrivateChannelTest extends TestCase
$this->assertSame([], json_decode($response->getContent(), true)); $this->assertSame([], json_decode($response->getContent(), true));
if (method_exists($this->channelManager, 'getPublishClient')) { if (method_exists($this->channelManager, 'getPublishClient')) {
$this->channelManager $this->channelManager->getPublishClient()->assertCalledWithArgsCount(1, 'publish', [
->getPublishClient() $this->channelManager->getRedisKey('1234', 'private-channel'),
->assertCalledWithArgsCount(1, 'publish', [ json_encode([
$this->channelManager->getRedisKey('1234', 'private-channel'), 'event' => 'some-event',
json_encode([ 'channel' => 'private-channel',
'event' => 'some-event', 'data' => json_encode(['some-data' => 'yes']),
'channel' => 'private-channel', 'appId' => '1234',
'data' => json_encode(['some-data' => 'yes']), 'socketId' => null,
'appId' => '1234', 'serverId' => $this->channelManager->getServerId(),
'socketId' => null, ]),
'serverId' => $this->channelManager->getServerId(), ]);
]),
]);
} }
$wsConnection->assertSentEvent('some-event', [ $wsConnection->assertSentEvent('some-event', [

143
tests/PublicChannelTest.php
View File

@ -14,11 +14,9 @@ class PublicChannelTest extends TestCase
{ {
$connection = $this->newActiveConnection(['public-channel']); $connection = $this->newActiveConnection(['public-channel']);
$this->channelManager $this->channelManager->getGlobalConnectionsCount('1234', 'public-channel')->then(function ($total) {
->getGlobalConnectionsCount('1234', 'public-channel') $this->assertEquals(1, $total);
->then(function ($total) { });
$this->assertEquals(1, $total);
});
$connection->assertSentEvent( $connection->assertSentEvent(
'pusher:connection_established', 'pusher:connection_established',
@ -40,11 +38,9 @@ class PublicChannelTest extends TestCase
{ {
$connection = $this->newActiveConnection(['public-channel']); $connection = $this->newActiveConnection(['public-channel']);
$this->channelManager $this->channelManager->getGlobalConnectionsCount('1234', 'public-channel')->then(function ($total) {
->getGlobalConnectionsCount('1234', 'public-channel') $this->assertEquals(1, $total);
->then(function ($total) { });
$this->assertEquals(1, $total);
});
$message = new Mocks\Message([ $message = new Mocks\Message([
'event' => 'pusher:unsubscribe', 'event' => 'pusher:unsubscribe',
@ -55,11 +51,9 @@ class PublicChannelTest extends TestCase
$this->pusherServer->onMessage($connection, $message); $this->pusherServer->onMessage($connection, $message);
$this->channelManager $this->channelManager->getGlobalConnectionsCount('1234', 'public-channel')->then(function ($total) {
->getGlobalConnectionsCount('1234', 'public-channel') $this->assertEquals(0, $total);
->then(function ($total) { });
$this->assertEquals(0, $total);
});
} }
public function test_can_whisper_to_public_channel() public function test_can_whisper_to_public_channel()
@ -103,22 +97,18 @@ class PublicChannelTest extends TestCase
$rick = $this->newActiveConnection(['public-channel']); $rick = $this->newActiveConnection(['public-channel']);
$morty = $this->newActiveConnection(['public-channel']); $morty = $this->newActiveConnection(['public-channel']);
$this->statisticsCollector $this->statisticsCollector->getStatistics()->then(function ($statistics) {
->getStatistics() $this->assertCount(1, $statistics);
->then(function ($statistics) { });
$this->assertCount(1, $statistics);
});
$this->statisticsCollector $this->statisticsCollector->getAppStatistics('1234')->then(function ($statistic) {
->getAppStatistics('1234') $this->assertEquals([
->then(function ($statistic) { 'peak_connections_count' => 2,
$this->assertEquals([ 'websocket_messages_count' => 2,
'peak_connections_count' => 2, 'api_messages_count' => 0,
'websocket_messages_count' => 2, 'app_id' => '1234',
'api_messages_count' => 0, ], $statistic->toArray());
'app_id' => '1234', });
], $statistic->toArray());
});
} }
public function test_local_connections_for_public_channels() public function test_local_connections_for_public_channels()
@ -126,17 +116,15 @@ class PublicChannelTest extends TestCase
$this->newActiveConnection(['public-channel']); $this->newActiveConnection(['public-channel']);
$this->newActiveConnection(['public-channel-2']); $this->newActiveConnection(['public-channel-2']);
$this->channelManager $this->channelManager->getLocalConnections()->then(function ($connections) {
->getLocalConnections() $this->assertCount(2, $connections);
->then(function ($connections) {
$this->assertCount(2, $connections);
foreach ($connections as $connection) { foreach ($connections as $connection) {
$this->assertInstanceOf( $this->assertInstanceOf(
ConnectionInterface::class, $connection ConnectionInterface::class, $connection
); );
} }
}); });
} }
public function test_events_are_processed_by_on_message_on_public_channels() public function test_events_are_processed_by_on_message_on_public_channels()
@ -201,11 +189,10 @@ class PublicChannelTest extends TestCase
$this->getSubscribeClient() $this->getSubscribeClient()
->assertNothingDispatched(); ->assertNothingDispatched();
$this->getPublishClient() $this->getPublishClient()->assertCalledWithArgs('publish', [
->assertCalledWithArgs('publish', [ $this->channelManager->getRedisKey('1234', 'public-channel'),
$this->channelManager->getRedisKey('1234', 'public-channel'), $message->getPayload(),
$message->getPayload(), ]);
]);
} }
public function test_it_fires_the_event_to_public_channel() public function test_it_fires_the_event_to_public_channel()
@ -239,16 +226,14 @@ class PublicChannelTest extends TestCase
$this->assertSame([], json_decode($response->getContent(), true)); $this->assertSame([], json_decode($response->getContent(), true));
$this->statisticsCollector $this->statisticsCollector->getAppStatistics('1234')->then(function ($statistic) {
->getAppStatistics('1234') $this->assertEquals([
->then(function ($statistic) { 'peak_connections_count' => 1,
$this->assertEquals([ 'websocket_messages_count' => 1,
'peak_connections_count' => 1, 'api_messages_count' => 1,
'websocket_messages_count' => 1, 'app_id' => '1234',
'api_messages_count' => 1, ], $statistic->toArray());
'app_id' => '1234', });
], $statistic->toArray());
});
} }
public function test_it_fires_event_across_servers_when_there_are_not_users_locally_for_public_channel() public function test_it_fires_event_across_servers_when_there_are_not_users_locally_for_public_channel()
@ -281,19 +266,17 @@ class PublicChannelTest extends TestCase
$this->assertSame([], json_decode($response->getContent(), true)); $this->assertSame([], json_decode($response->getContent(), true));
if (method_exists($this->channelManager, 'getPublishClient')) { if (method_exists($this->channelManager, 'getPublishClient')) {
$this->channelManager $this->channelManager->getPublishClient()->assertCalledWithArgsCount(1, 'publish', [
->getPublishClient() $this->channelManager->getRedisKey('1234', 'public-channel'),
->assertCalledWithArgsCount(1, 'publish', [ json_encode([
$this->channelManager->getRedisKey('1234', 'public-channel'), 'event' => 'some-event',
json_encode([ 'channel' => 'public-channel',
'event' => 'some-event', 'data' => json_encode(['some-data' => 'yes']),
'channel' => 'public-channel', 'appId' => '1234',
'data' => json_encode(['some-data' => 'yes']), 'socketId' => null,
'appId' => '1234', 'serverId' => $this->channelManager->getServerId(),
'socketId' => null, ]),
'serverId' => $this->channelManager->getServerId(), ]);
]),
]);
} }
} }
@ -329,19 +312,17 @@ class PublicChannelTest extends TestCase
$this->assertSame([], json_decode($response->getContent(), true)); $this->assertSame([], json_decode($response->getContent(), true));
if (method_exists($this->channelManager, 'getPublishClient')) { if (method_exists($this->channelManager, 'getPublishClient')) {
$this->channelManager $this->channelManager->getPublishClient()->assertCalledWithArgsCount(1, 'publish', [
->getPublishClient() $this->channelManager->getRedisKey('1234', 'public-channel'),
->assertCalledWithArgsCount(1, 'publish', [ json_encode([
$this->channelManager->getRedisKey('1234', 'public-channel'), 'event' => 'some-event',
json_encode([ 'channel' => 'public-channel',
'event' => 'some-event', 'data' => json_encode(['some-data' => 'yes']),
'channel' => 'public-channel', 'appId' => '1234',
'data' => json_encode(['some-data' => 'yes']), 'socketId' => null,
'appId' => '1234', 'serverId' => $this->channelManager->getServerId(),
'socketId' => null, ]),
'serverId' => $this->channelManager->getServerId(), ]);
]),
]);
} }
$wsConnection->assertSentEvent('some-event', [ $wsConnection->assertSentEvent('some-event', [

112
tests/RedisPongRemovalTest.php
View File

@ -19,31 +19,23 @@ class RedisPongRemovalTest extends TestCase
// Make the connection look like it was lost 1 day ago. // Make the connection look like it was lost 1 day ago.
$this->channelManager->addConnectionToSet($obsoleteConnection, Carbon::now()->subDays(1)); $this->channelManager->addConnectionToSet($obsoleteConnection, Carbon::now()->subDays(1));
$this->channelManager $this->channelManager->getGlobalConnectionsCount('1234', 'public-channel')->then(function ($count) {
->getGlobalConnectionsCount('1234', 'public-channel') $this->assertEquals(2, $count);
->then(function ($count) { });
$this->assertEquals(2, $count);
});
$this->channelManager $this->channelManager->getConnectionsFromSet(0, Carbon::now()->subMinutes(2)->format('U'))->then(function ($expiredConnections) {
->getConnectionsFromSet(0, Carbon::now()->subMinutes(2)->format('U')) $this->assertCount(1, $expiredConnections);
->then(function ($expiredConnections) { });
$this->assertCount(1, $expiredConnections);
});
$this->channelManager->removeObsoleteConnections(); $this->channelManager->removeObsoleteConnections();
$this->channelManager $this->channelManager->getGlobalConnectionsCount('1234', 'public-channel')->then(function ($count) {
->getGlobalConnectionsCount('1234', 'public-channel') $this->assertEquals(1, $count);
->then(function ($count) { });
$this->assertEquals(1, $count);
});
$this->channelManager $this->channelManager->getConnectionsFromSet(0, Carbon::now()->subMinutes(2)->format('U'))->then(function ($expiredConnections) {
->getConnectionsFromSet(0, Carbon::now()->subMinutes(2)->format('U')) $this->assertCount(0, $expiredConnections);
->then(function ($expiredConnections) { });
$this->assertCount(0, $expiredConnections);
});
} }
public function test_not_ponged_connections_do_get_removed_on_redis_for_private_channels() public function test_not_ponged_connections_do_get_removed_on_redis_for_private_channels()
@ -59,31 +51,23 @@ class RedisPongRemovalTest extends TestCase
// Make the connection look like it was lost 1 day ago. // Make the connection look like it was lost 1 day ago.
$this->channelManager->addConnectionToSet($obsoleteConnection, Carbon::now()->subDays(1)); $this->channelManager->addConnectionToSet($obsoleteConnection, Carbon::now()->subDays(1));
$this->channelManager $this->channelManager->getGlobalConnectionsCount('1234', 'private-channel')->then(function ($count) {
->getGlobalConnectionsCount('1234', 'private-channel') $this->assertEquals(2, $count);
->then(function ($count) { });
$this->assertEquals(2, $count);
});
$this->channelManager $this->channelManager->getConnectionsFromSet(0, Carbon::now()->subMinutes(2)->format('U'))->then(function ($expiredConnections) {
->getConnectionsFromSet(0, Carbon::now()->subMinutes(2)->format('U')) $this->assertCount(1, $expiredConnections);
->then(function ($expiredConnections) { });
$this->assertCount(1, $expiredConnections);
});
$this->channelManager->removeObsoleteConnections(); $this->channelManager->removeObsoleteConnections();
$this->channelManager $this->channelManager->getGlobalConnectionsCount('1234', 'private-channel')->then(function ($count) {
->getGlobalConnectionsCount('1234', 'private-channel') $this->assertEquals(1, $count);
->then(function ($count) { });
$this->assertEquals(1, $count);
});
$this->channelManager $this->channelManager->getConnectionsFromSet(0, Carbon::now()->subMinutes(2)->format('U'))->then(function ($expiredConnections) {
->getConnectionsFromSet(0, Carbon::now()->subMinutes(2)->format('U')) $this->assertCount(0, $expiredConnections);
->then(function ($expiredConnections) { });
$this->assertCount(0, $expiredConnections);
});
} }
public function test_not_ponged_connections_do_get_removed_on_redis_for_presence_channels() public function test_not_ponged_connections_do_get_removed_on_redis_for_presence_channels()
@ -99,42 +83,30 @@ class RedisPongRemovalTest extends TestCase
// Make the connection look like it was lost 1 day ago. // Make the connection look like it was lost 1 day ago.
$this->channelManager->addConnectionToSet($obsoleteConnection, Carbon::now()->subDays(1)); $this->channelManager->addConnectionToSet($obsoleteConnection, Carbon::now()->subDays(1));
$this->channelManager $this->channelManager->getGlobalConnectionsCount('1234', 'presence-channel')->then(function ($count) {
->getGlobalConnectionsCount('1234', 'presence-channel') $this->assertEquals(2, $count);
->then(function ($count) { });
$this->assertEquals(2, $count);
});
$this->channelManager $this->channelManager->getConnectionsFromSet(0, Carbon::now()->subMinutes(2)->format('U'))->then(function ($expiredConnections) {
->getConnectionsFromSet(0, Carbon::now()->subMinutes(2)->format('U')) $this->assertCount(1, $expiredConnections);
->then(function ($expiredConnections) { });
$this->assertCount(1, $expiredConnections);
});
$this->channelManager $this->channelManager->getChannelMembers('1234', 'presence-channel')->then(function ($members) {
->getChannelMembers('1234', 'presence-channel') $this->assertCount(2, $members);
->then(function ($members) { });
$this->assertCount(2, $members);
});
$this->channelManager->removeObsoleteConnections(); $this->channelManager->removeObsoleteConnections();
$this->channelManager $this->channelManager->getGlobalConnectionsCount('1234', 'presence-channel')->then(function ($count) {
->getGlobalConnectionsCount('1234', 'presence-channel') $this->assertEquals(1, $count);
->then(function ($count) { });
$this->assertEquals(1, $count);
});
$this->channelManager $this->channelManager->getConnectionsFromSet(0, Carbon::now()->subMinutes(2)->format('U'))->then(function ($expiredConnections) {
->getConnectionsFromSet(0, Carbon::now()->subMinutes(2)->format('U')) $this->assertCount(0, $expiredConnections);
->then(function ($expiredConnections) { });
$this->assertCount(0, $expiredConnections);
});
$this->channelManager $this->channelManager->getChannelMembers('1234', 'presence-channel')->then(function ($members) {
->getChannelMembers('1234', 'presence-channel') $this->assertCount(1, $members);
->then(function ($members) { });
$this->assertCount(1, $members);
});
} }
} }