LocalChannelManager::findOrCreate() - Code Metrics - Inspection of "[2.x] Remove obsolete connections older than 120 s..." - beyondcode/laravel-websockets - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#547)

by Alexandru

created 2020-09-19 15:42 UTC

LocalChannelManager::findOrCreate() A

↳ Parent: LocalChannelManager

Complexity

Conditions	2
Paths	2

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
dl	0
loc	10
rs	9.9332
c	0
b	0
f	0
cc	2
nc	2
nop	2

<?php

namespace BeyondCode\LaravelWebSockets\ChannelManagers;

use BeyondCode\LaravelWebSockets\Channels\Channel;
use BeyondCode\LaravelWebSockets\Channels\PresenceChannel;
use BeyondCode\LaravelWebSockets\Channels\PrivateChannel;
use BeyondCode\LaravelWebSockets\Contracts\ChannelManager;
use BeyondCode\LaravelWebSockets\Helpers;
use Carbon\Carbon;
use Illuminate\Cache\ArrayLock;
use Illuminate\Cache\ArrayStore;
use Illuminate\Support\Str;
use Ratchet\ConnectionInterface;
use React\EventLoop\LoopInterface;
use React\Promise\PromiseInterface;
use stdClass;

class LocalChannelManager implements ChannelManager
{
    /**
     * The list of stored channels.
     *
     * @var array
     */
    protected $channels = [];

    /**
     * The list of users that joined the presence channel.
     *
     * @var array
     */
    protected $users = [];

    /**
     * The list of users by socket and their attached id.
     *
     * @var array
     */
    protected $userSockets = [];

    /**
     * Wether the current instance accepts new connections.
     *
     * @var bool
     */
    protected $acceptsNewConnections = true;

    /**
     * The lock name to use on Array to avoid multiple
     * actions that might lead to multiple processings.
     *
     * @var string
     */
    protected static $lockName = 'laravel-websockets:channel-manager:lock';

    /**
     * Create a new channel manager instance.
     *
     * @param  LoopInterface  $loop
     * @param  string|null  $factoryClass
     * @return void

     */
    public function __construct(LoopInterface $loop, $factoryClass = null)
    {
        //
    }

    /**
     * Find the channel by app & name.
     *
     * @param  string|int  $appId
     * @param  string  $channel
     * @return null|BeyondCode\LaravelWebSockets\Channels\Channel
     */
    public function find($appId, string $channel)
    {
        return $this->channels[$appId][$channel] ?? null;
    }

    /**
     * Find a channel by app & name or create one.
     *
     * @param  string|int  $appId
     * @param  string  $channel
     * @return BeyondCode\LaravelWebSockets\Channels\Channel
     */
    public function findOrCreate($appId, string $channel)
    {
        if (! $channelInstance = $this->find($appId, $channel)) {
            $class = $this->getChannelClassName($channel);

            $this->channels[$appId][$channel] = new $class($channel);
        }

        return $this->channels[$appId][$channel];
    }

    /**
     * Get the local connections, regardless of the channel
     * they are connected to.
     *
     * @return \React\Promise\PromiseInterface
     */
    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();

        return Helpers::createFulfilledPromise($connections);
    }

    /**
     * Get all channels for a specific app
     * for the current instance.
     *
     * @param  string|int  $appId
     * @return \React\Promise\PromiseInterface[array]

     */
    public function getLocalChannels($appId): PromiseInterface
    {
        return Helpers::createFulfilledPromise(
            $this->channels[$appId] ?? []
        );
    }

    /**
     * Get all channels for a specific app
     * across multiple servers.
     *
     * @param  string|int  $appId
     * @return \React\Promise\PromiseInterface[array]

     */
    public function getGlobalChannels($appId): PromiseInterface
    {
        return $this->getLocalChannels($appId);
    }

    /**
     * Remove connection from all channels.
     *
     * @param  \Ratchet\ConnectionInterface  $connection
     * @return PromiseInterface[bool]

     */
    public function unsubscribeFromAllChannels(ConnectionInterface $connection): PromiseInterface
    {
        if (! isset($connection->app)) {
interface SomeInterface { }
class SomeClass implements SomeInterface {
    public $a;
}

function someFunction(SomeInterface $object) {
    if ($object instanceof SomeClass) {
        $a = $object->a;
    }
}
            return new FuilfilledPromise(false);
        }

        $this->getLocalChannels($connection->app->id)
interface SomeInterface { }
class SomeClass implements SomeInterface {
    public $a;
}

function someFunction(SomeInterface $object) {
    if ($object instanceof SomeClass) {
        $a = $object->a;
    }
}
            ->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]);
                    });
            });

        $this->getLocalChannels($connection->app->id)
interface SomeInterface { }
class SomeClass implements SomeInterface {
    public $a;
}

function someFunction(SomeInterface $object) {
    if ($object instanceof SomeClass) {
        $a = $object->a;
    }
}
            ->then(function ($channels) use ($connection) {
                if (count($channels) === 0) {
                    unset($this->channels[$connection->app->id]);
                }
            });

        return Helpers::createFulfilledPromise(true);
    }

    /**
     * Subscribe the connection to a specific channel.
     *
     * @param  \Ratchet\ConnectionInterface  $connection
     * @param  string  $channelName
     * @param  stdClass  $payload
     * @return PromiseInterface[bool]

     */
    public function subscribeToChannel(ConnectionInterface $connection, string $channelName, stdClass $payload): PromiseInterface
    {
        $channel = $this->findOrCreate($connection->app->id, $channelName);
interface SomeInterface { }
class SomeClass implements SomeInterface {
    public $a;
}

function someFunction(SomeInterface $object) {
    if ($object instanceof SomeClass) {
        $a = $object->a;
    }
}

        return Helpers::createFulfilledPromise(
            $channel->subscribe($connection, $payload)
        );
    }

    /**
     * Unsubscribe the connection from the channel.
     *
     * @param  \Ratchet\ConnectionInterface  $connection
     * @param  string  $channelName
     * @param  stdClass  $payload
     * @return PromiseInterface[bool]

     */
    public function unsubscribeFromChannel(ConnectionInterface $connection, string $channelName, stdClass $payload): PromiseInterface
    {
        $channel = $this->findOrCreate($connection->app->id, $channelName);
interface SomeInterface { }
class SomeClass implements SomeInterface {
    public $a;
}

function someFunction(SomeInterface $object) {
    if ($object instanceof SomeClass) {
        $a = $object->a;
    }
}

        return Helpers::createFulfilledPromise(
            $channel->unsubscribe($connection, $payload)
        );
    }

    /**
     * Subscribe the connection to a specific channel, returning
     * a promise containing the amount of connections.
     *
     * @param  string|int  $appId
     * @return PromiseInterface[int]

     */
    public function subscribeToApp($appId): PromiseInterface
    {
        return Helpers::createFulfilledPromise(0);
    }

    /**
     * Unsubscribe the connection from the channel, returning
     * a promise containing the amount of connections after decrement.
     *
     * @param  string|int  $appId
     * @return PromiseInterface[int]

     */
    public function unsubscribeFromApp($appId): PromiseInterface
    {
        return Helpers::createFulfilledPromise(0);
    }

    /**
     * Get the connections count on the app
     * for the current server instance.
     *
     * @param  string|int  $appId
     * @param  string|null  $channelName
     * @return PromiseInterface[int]

     */
    public function getLocalConnectionsCount($appId, string $channelName = null): PromiseInterface
    {
        return $this->getLocalChannels($appId)
            ->then(function ($channels) use ($channelName) {
                return collect($channels)->when(! is_null($channelName), function ($collection) use ($channelName) {
                    return $collection->filter(function (Channel $channel) use ($channelName) {
                        return $channel->getName() === $channelName;
                    });
                })
                ->flatMap(function (Channel $channel) {
                    return collect($channel->getConnections())->pluck('socketId');
                })
                ->unique()->count();
            });
    }

    /**
     * Get the connections count
     * across multiple servers.
     *
     * @param  string|int  $appId
     * @param  string|null  $channelName
     * @return PromiseInterface[int]

     */
    public function getGlobalConnectionsCount($appId, string $channelName = null): PromiseInterface
    {
        return $this->getLocalConnectionsCount($appId, $channelName);
    }

    /**
     * Broadcast the message across multiple servers.
     *
     * @param  string|int  $appId
     * @param  string|null  $socketId
     * @param  string  $channel
     * @param  stdClass  $payload
     * @param  string|null  $serverId
     * @return PromiseInterface[bool]

     */
    public function broadcastAcrossServers($appId, ?string $socketId, string $channel, stdClass $payload, string $serverId = null): PromiseInterface
    {
        return Helpers::createFulfilledPromise(true);
    }

    /**
     * Handle the user when it joined a presence channel.
     *
     * @param  \Ratchet\ConnectionInterface  $connection
     * @param  stdClass  $user
     * @param  string  $channel
     * @param  stdClass  $payload
     * @return PromiseInterface[bool]

     */
    public function userJoinedPresenceChannel(ConnectionInterface $connection, stdClass $user, string $channel, stdClass $payload): PromiseInterface
    {
        $this->users["{$connection->app->id}:{$channel}"][$connection->socketId] = json_encode($user);
interface SomeInterface { }
class SomeClass implements SomeInterface {
    public $a;
}

function someFunction(SomeInterface $object) {
    if ($object instanceof SomeClass) {
        $a = $object->a;
    }
}
        $this->userSockets["{$connection->app->id}:{$channel}:{$user->user_id}"][] = $connection->socketId;
interface SomeInterface { }
class SomeClass implements SomeInterface {
    public $a;
}

function someFunction(SomeInterface $object) {
    if ($object instanceof SomeClass) {
        $a = $object->a;
    }
}

        return Helpers::createFulfilledPromise(true);
    }

    /**
     * Handle the user when it left a presence channel.
     *
     * @param  \Ratchet\ConnectionInterface  $connection
     * @param  stdClass  $user
     * @param  string  $channel
     * @param  stdClass  $payload
/**
 * @param array $germany
 * @param array $island
 * @param array $italy
 */
function finale($germany, $island) {
    return "2:1";
}
     * @return PromiseInterface[bool]

     */
    public function userLeftPresenceChannel(ConnectionInterface $connection, stdClass $user, string $channel): PromiseInterface
    {
        unset($this->users["{$connection->app->id}:{$channel}"][$connection->socketId]);

        $deletableSocketKey = array_search(
            $connection->socketId,
interface SomeInterface { }
class SomeClass implements SomeInterface {
    public $a;
}

function someFunction(SomeInterface $object) {
    if ($object instanceof SomeClass) {
        $a = $object->a;
    }
}
            $this->userSockets["{$connection->app->id}:{$channel}:{$user->user_id}"]
interface SomeInterface { }
class SomeClass implements SomeInterface {
    public $a;
}

function someFunction(SomeInterface $object) {
    if ($object instanceof SomeClass) {
        $a = $object->a;
    }
}
        );

        if ($deletableSocketKey !== false) {
            unset($this->userSockets["{$connection->app->id}:{$channel}:{$user->user_id}"][$deletableSocketKey]);

            if (count($this->userSockets["{$connection->app->id}:{$channel}:{$user->user_id}"]) === 0) {
interface SomeInterface { }
class SomeClass implements SomeInterface {
    public $a;
}

function someFunction(SomeInterface $object) {
    if ($object instanceof SomeClass) {
        $a = $object->a;
    }
}
                unset($this->userSockets["{$connection->app->id}:{$channel}:{$user->user_id}"]);
            }
        }

        return Helpers::createFulfilledPromise(true);
    }

    /**
     * Get the presence channel members.
     *
     * @param  string|int  $appId
     * @param  string  $channel
     * @return \React\Promise\PromiseInterface
     */
    public function getChannelMembers($appId, string $channel): PromiseInterface
    {
        $members = $this->users["{$appId}:{$channel}"] ?? [];

        $members = collect($members)->map(function ($user) {
            return json_decode($user);
        })->unique('user_id')->toArray();

        return Helpers::createFulfilledPromise($members);
    }

    /**
     * Get a member from a presence channel based on connection.
     *
     * @param  \Ratchet\ConnectionInterface  $connection
     * @param  string  $channel
     * @return \React\Promise\PromiseInterface
     */
    public function getChannelMember(ConnectionInterface $connection, string $channel): PromiseInterface
    {
        $member = $this->users["{$connection->app->id}:{$channel}"][$connection->socketId] ?? null;
interface SomeInterface { }
class SomeClass implements SomeInterface {
    public $a;
}

function someFunction(SomeInterface $object) {
    if ($object instanceof SomeClass) {
        $a = $object->a;
    }
}

        return Helpers::createFulfilledPromise($member);
    }

    /**
     * Get the presence channels total members count.
     *
     * @param  string|int  $appId
     * @param  array  $channelNames
     * @return \React\Promise\PromiseInterface
     */
    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;

                return $results;
            }, []);

        return Helpers::createFulfilledPromise($results);
    }

    /**
     * Get the socket IDs for a presence channel member.
     *
     * @param  string|int  $userId
     * @param  string|int  $appId
     * @param  string  $channelName
     * @return \React\Promise\PromiseInterface
     */
    public function getMemberSockets($userId, $appId, $channelName): PromiseInterface
    {
        return Helpers::createFulfilledPromise(
            $this->userSockets["{$appId}:{$channelName}:{$userId}"] ?? []
        );
    }

    /**
     * Keep tracking the connections availability when they pong.
     *
     * @param  \Ratchet\ConnectionInterface  $connection
     * @return PromiseInterface[bool]

     */
    public function connectionPonged(ConnectionInterface $connection): PromiseInterface
    {
        $connection->lastPongedAt = Carbon::now();
interface SomeInterface { }
class SomeClass implements SomeInterface {
    public $a;
}

function someFunction(SomeInterface $object) {
    if ($object instanceof SomeClass) {
        $a = $object->a;
    }
}

        return $this->updateConnectionInChannels($connection);
    }

    /**
     * Remove the obsolete connections that didn't ponged in a while.
     *
     * @return PromiseInterface[bool]

     */
    public function removeObsoleteConnections(): PromiseInterface
    {
        if (! $this->lock()->acquire()) {
            return Helpers::createFulfilledPromise(false);
        }

        $this->getLocalConnections()->then(function ($connections) {
            foreach ($connections as $connection) {
                $differenceInSeconds = $connection->lastPongedAt->diffInSeconds(Carbon::now());

                if ($differenceInSeconds > 120) {
                    $this->unsubscribeFromAllChannels($connection);
                }
            }
        });

        return Helpers::createFulfilledPromise(
            $this->lock()->release()
        );
    }

    /**
     * Update the connection in all channels.
     *
     * @param  ConnectionInterface  $connection
     * @return PromiseInterface[bool]

     */
    public function updateConnectionInChannels($connection): PromiseInterface
    {
        return $this->getLocalChannels($connection->app->id)
interface SomeInterface { }
class SomeClass implements SomeInterface {
    public $a;
}

function someFunction(SomeInterface $object) {
    if ($object instanceof SomeClass) {
        $a = $object->a;
    }
}
            ->then(function ($channels) use ($connection) {
                foreach ($channels as $channel) {
                    if ($channel->hasConnection($connection)) {
                        $channel->saveConnection($connection);
                    }
                }

                return true;
            });
    }

    /**
     * Mark the current instance as unable to accept new connections.
     *
     * @return $this
     */
    public function declineNewConnections()
    {
        $this->acceptsNewConnections = false;

        return $this;
    }

    /**
     * Check if the current server instance
     * accepts new connections.
     *
     * @return bool
     */
    public function acceptsNewConnections(): bool
    {
        return $this->acceptsNewConnections;
    }

    /**
     * Get the channel class by the channel name.
     *
     * @param  string  $channelName
     * @return string
     */
    protected function getChannelClassName(string $channelName): string
    {
        if (Str::startsWith($channelName, 'private-')) {
            return PrivateChannel::class;
        }

        if (Str::startsWith($channelName, 'presence-')) {
            return PresenceChannel::class;
        }

        return Channel::class;
    }

    /**
     * Get a new ArrayLock instance to avoid race conditions.
     *
     * @return \Illuminate\Cache\CacheLock
     */
    protected function lock()
    {
        return new ArrayLock(new ArrayStore, static::$lockName, 0);
    }
}


Pull Request — master (#547)

LocalChannelManager::findOrCreate() A

Complexity

Size

Duplication

Importance

Available Fixes

Available Fixes

Available Fixes

Available Fixes

Available Fixes

Available Fixes

Available Fixes

Available Fixes

Available Fixes

Available Fixes

Available Fixes

Available Fixes

Available Fixes

Available Fixes

Available Fixes

Available Fixes

1			<?php
2
3			namespace BeyondCode\LaravelWebSockets\ChannelManagers;
4
5			use BeyondCode\LaravelWebSockets\Channels\Channel;
6			use BeyondCode\LaravelWebSockets\Channels\PresenceChannel;
7			use BeyondCode\LaravelWebSockets\Channels\PrivateChannel;
8			use BeyondCode\LaravelWebSockets\Contracts\ChannelManager;
9			use BeyondCode\LaravelWebSockets\Helpers;
10			use Carbon\Carbon;
11			use Illuminate\Cache\ArrayLock;
12			use Illuminate\Cache\ArrayStore;
13			use Illuminate\Support\Str;
14			use Ratchet\ConnectionInterface;
15			use React\EventLoop\LoopInterface;
16			use React\Promise\PromiseInterface;
17			use stdClass;
18
19			class LocalChannelManager implements ChannelManager
20			{
21			/**
22			* The list of stored channels.
23			*
24			* @var array
25			*/
26			protected $channels = [];
27
28			/**
29			* The list of users that joined the presence channel.
30			*
31			* @var array
32			*/
33			protected $users = [];
34
35			/**
36			* The list of users by socket and their attached id.
37			*
38			* @var array
39			*/
40			protected $userSockets = [];

beyondcode / laravel-websockets

Pull Request — master (#547)

LocalChannelManager::findOrCreate() A

Complexity

Size

Duplication

Importance

Available Fixes

Available Fixes

Available Fixes

Available Fixes

Available Fixes

Available Fixes

Available Fixes

Available Fixes

Available Fixes

Available Fixes

Available Fixes

Available Fixes

Available Fixes

Available Fixes

Available Fixes

Available Fixes

Duplication Side-by-Side

Filter issues like