RedisClient::subscribe() - Code Metrics - Inspection of "WIP: Redis as a replication backend for scalabilit..." - beyondcode/laravel-websockets - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#140)

unknown

created 2019-04-05 19:33 UTC

RedisClient::subscribe() A

↳ Parent: RedisClient

Complexity

Conditions	2
Paths	2

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
dl	0
loc	13
rs	9.8333
c	0
b	0
f	0
cc	2
nc	2
nop	2

<?php

namespace BeyondCode\LaravelWebSockets\PubSub\Redis;

use stdClass;
use Illuminate\Support\Str;
use Clue\React\Redis\Client;
use Clue\React\Redis\Factory;
use React\EventLoop\LoopInterface;
use React\Promise\PromiseInterface;
use BeyondCode\LaravelWebSockets\PubSub\ReplicationInterface;
use BeyondCode\LaravelWebSockets\WebSockets\Channels\ChannelManager;

class RedisClient implements ReplicationInterface
{
    /**
     * @var LoopInterface
     */
    protected $loop;

    /**
     * @var string
     */
    protected $serverId;

    /**
     * @var Client
     */
    protected $publishClient;

    /**
     * @var Client
     */
    protected $subscribeClient;

    /**
     * Mapping of subscribed channels, where the key is the channel name,
     * and the value is the amount of connections which are subscribed to
     * that channel. Used to keep track of whether we still need to stay
     * subscribed to those channels with Redis.
     *
     * @var int[]
     */
    protected $subscribedChannels = [];

    /**
     * RedisClient constructor.
     */
    public function __construct()
    {
        $this->serverId = Str::uuid()->toString();
    }

    /**
     * Boot the RedisClient, initializing the connections.
     *
     * @param LoopInterface $loop
     * @return ReplicationInterface
     */
    public function boot(LoopInterface $loop): ReplicationInterface
    {
        $this->loop = $loop;

        $connectionUri = $this->getConnectionUri();
        $factory = new Factory($this->loop);

        $this->publishClient = $factory->createLazyClient($connectionUri);
        $this->subscribeClient = $factory->createLazyClient($connectionUri);

        $this->subscribeClient->on('message', function ($channel, $payload) {
            $this->onMessage($channel, $payload);
        });

        return $this;
    }

    /**
     * Handle a message received from Redis on a specific channel.
     *
     * @param string $redisChannel
     * @param string $payload
     */
    protected function onMessage(string $redisChannel, string $payload)
    {
        $payload = json_decode($payload);

        // Ignore messages sent by ourselves
        if (isset($payload->serverId) && $this->serverId === $payload->serverId) {
            return;
        }

        // Pull out the app ID. See RedisPusherBroadcaster
        $appId = $payload->appId;

        // We need to put the channel name in the payload.
        // We strip the app ID from the channel name, websocket clients
        // expect the channel name to not include the app ID.
        $payload->channel = Str::after($redisChannel, "$appId:");

        /* @var $channelManager ChannelManager */
        $channelManager = app(ChannelManager::class);

        // Load the Channel instance, if any
        $channel = $channelManager->find($appId, $payload->channel);

        // If no channel is found, none of our connections want to
        // receive this message, so we ignore it.
        if (! $channel) {
            return;
        }

        $socket = $payload->socket ?? null;

        // Remove fields intended for internal use from the payload
        unset($payload->socket);
        unset($payload->serverId);
        unset($payload->appId);

        // Push the message out to connected websocket clients
        $channel->broadcastToEveryoneExcept($payload, $socket);
    }

    /**
     * Subscribe to a channel on behalf of websocket user.
     *
     * @param string $appId
     * @param string $channel
     * @return bool
     */
    public function subscribe(string $appId, string $channel): bool
    {
        if (! isset($this->subscribedChannels["$appId:$channel"])) {
            // We're not subscribed to the channel yet, subscribe and set the count to 1
            $this->subscribeClient->__call('subscribe', ["$appId:$channel"]);
            $this->subscribedChannels["$appId:$channel"] = 1;
        } else {
            // Increment the subscribe count if we've already subscribed
            $this->subscribedChannels["$appId:$channel"]++;
        }

        return true;
    }

    /**
     * Unsubscribe from a channel on behalf of a websocket user.
     *
     * @param string $appId
     * @param string $channel
     * @return bool
     */
    public function unsubscribe(string $appId, string $channel): bool
    {
        if (! isset($this->subscribedChannels["$appId:$channel"])) {
            return false;
        }

        // Decrement the subscription count for this channel
        $this->subscribedChannels["$appId:$channel"]--;

        // If we no longer have subscriptions to that channel, unsubscribe
        if ($this->subscribedChannels["$appId:$channel"] < 1) {
            $this->subscribeClient->__call('unsubscribe', ["$appId:$channel"]);
            unset($this->subscribedChannels["$appId:$channel"]);
        }

        return true;
    }

    /**
     * Publish a message to a channel on behalf of a websocket user.
     *
     * @param string $appId
     * @param string $channel
     * @param stdClass $payload
     * @return bool
     */
    public function publish(string $appId, string $channel, stdClass $payload): bool
    {
        $payload->appId = $appId;
        $payload->serverId = $this->serverId;

        $this->publishClient->__call('publish', ["$appId:$channel", json_encode($payload)]);

        return true;
    }

    /**
     * Add a member to a channel. To be called when they have
     * subscribed to the channel.
     *
     * @param string $appId
     * @param string $channel
     * @param string $socketId
     * @param string $data
     */
    public function joinChannel(string $appId, string $channel, string $socketId, string $data)
    {
        $this->publishClient->__call('hset', ["$appId:$channel", $socketId, $data]);
    }

    /**
     * Remove a member from the channel. To be called when they have
     * unsubscribed from the channel.
     *
     * @param string $appId
     * @param string $channel
     * @param string $socketId
     */
    public function leaveChannel(string $appId, string $channel, string $socketId)
    {
        $this->publishClient->__call('hdel', ["$appId:$channel", $socketId]);
    }

    /**
     * Retrieve the full information about the members in a presence channel.
     *
     * @param string $appId
     * @param string $channel
     * @return PromiseInterface
     */
    public function channelMembers(string $appId, string $channel): PromiseInterface
    {
        return $this->publishClient->__call('hgetall', ["$appId:$channel"])
            ->then(function ($members) {
                // The data is expected as objects, so we need to JSON decode
                return array_map(function ($user) {
                    return json_decode($user);
                }, $members);
            });
    }

    /**
     * Get the amount of users subscribed for each presence channel.
     *
     * @param string $appId
     * @param array $channelNames
     * @return PromiseInterface
     */
    public function channelMemberCounts(string $appId, array $channelNames): PromiseInterface
    {
        $this->publishClient->__call('multi', []);

        foreach ($channelNames as $channel) {
            $this->publishClient->__call('hlen', ["$appId:$channel"]);
        }

        return $this->publishClient->__call('exec', [])
            ->then(function ($data) use ($channelNames) {
                return array_combine($channelNames, $data);
            });
    }

    /**
     * Build the Redis connection URL from Laravel database config.
     *
     * @return string
     */
    protected function getConnectionUri()
    {
        $name = config('websockets.replication.connection') ?? 'default';
        $config = config("database.redis.$name");
        $host = $config['host'];
        $port = $config['port'] ? (':'.$config['port']) : ':6379';

        $query = [];
        if ($config['password']) {
            $query['password'] = $config['password'];
        }
        if ($config['database']) {
            $query['database'] = $config['database'];
        }
        $query = http_build_query($query);

        return "redis://$host$port".($query ? '?'.$query : '');
    }
}


1			<?php
2
3			namespace BeyondCode\LaravelWebSockets\PubSub\Redis;
4
5			use stdClass;
6			use Illuminate\Support\Str;
7			use Clue\React\Redis\Client;
8			use Clue\React\Redis\Factory;
9			use React\EventLoop\LoopInterface;
10			use React\Promise\PromiseInterface;
11			use BeyondCode\LaravelWebSockets\PubSub\ReplicationInterface;
12			use BeyondCode\LaravelWebSockets\WebSockets\Channels\ChannelManager;
13
14			class RedisClient implements ReplicationInterface
15			{
16			/**
17			* @var LoopInterface
18			*/
19			protected $loop;
20
21			/**
22			* @var string
23			*/
24			protected $serverId;
25
26			/**
27			* @var Client
28			*/
29			protected $publishClient;
30
31			/**
32			* @var Client
33			*/
34			protected $subscribeClient;
35
36			/**
37			* Mapping of subscribed channels, where the key is the channel name,
38			* and the value is the amount of connections which are subscribed to
39			* that channel. Used to keep track of whether we still need to stay
40			* subscribed to those channels with Redis.
41			*
42			* @var int[]
43			*/
44			protected $subscribedChannels = [];
45
46			/**
47			* RedisClient constructor.
48			*/
49			public function __construct()
50			{
51			$this->serverId = Str::uuid()->toString();
52			}
53
54			/**
55			* Boot the RedisClient, initializing the connections.
56			*
57			* @param LoopInterface $loop
58			* @return ReplicationInterface
59			*/
60			public function boot(LoopInterface $loop): ReplicationInterface
61			{
62			$this->loop = $loop;
63
64			$connectionUri = $this->getConnectionUri();
65			$factory = new Factory($this->loop);
66
67			$this->publishClient = $factory->createLazyClient($connectionUri);
68			$this->subscribeClient = $factory->createLazyClient($connectionUri);
69
70			$this->subscribeClient->on('message', function ($channel, $payload) {
71			$this->onMessage($channel, $payload);
72			});
73
74			return $this;
75			}
76
77			/**
78			* Handle a message received from Redis on a specific channel.
79			*
80			* @param string $redisChannel
81			* @param string $payload
82			*/
83			protected function onMessage(string $redisChannel, string $payload)
84			{
85			$payload = json_decode($payload);
86
87			// Ignore messages sent by ourselves
88			if (isset($payload->serverId) && $this->serverId === $payload->serverId) {
89			return;
90			}
91
92			// Pull out the app ID. See RedisPusherBroadcaster
93			$appId = $payload->appId;
94
95			// We need to put the channel name in the payload.
96			// We strip the app ID from the channel name, websocket clients
97			// expect the channel name to not include the app ID.
98			$payload->channel = Str::after($redisChannel, "$appId:");
99
100			/* @var $channelManager ChannelManager */
101			$channelManager = app(ChannelManager::class);
102
103			// Load the Channel instance, if any
104			$channel = $channelManager->find($appId, $payload->channel);
105
106			// If no channel is found, none of our connections want to
107			// receive this message, so we ignore it.
108			if (! $channel) {
109			return;
110			}
111
112			$socket = $payload->socket ?? null;
113
114			// Remove fields intended for internal use from the payload
115			unset($payload->socket);
116			unset($payload->serverId);
117			unset($payload->appId);
118
119			// Push the message out to connected websocket clients
120			$channel->broadcastToEveryoneExcept($payload, $socket);
121			}
122
123			/**
124			* Subscribe to a channel on behalf of websocket user.
125			*
126			* @param string $appId
127			* @param string $channel
128			* @return bool
129			*/
130			public function subscribe(string $appId, string $channel): bool
131			{
132			if (! isset($this->subscribedChannels["$appId:$channel"])) {
133			// We're not subscribed to the channel yet, subscribe and set the count to 1
134			$this->subscribeClient->__call('subscribe', ["$appId:$channel"]);
135			$this->subscribedChannels["$appId:$channel"] = 1;
136			} else {
137			// Increment the subscribe count if we've already subscribed
138			$this->subscribedChannels["$appId:$channel"]++;
139			}
140
141			return true;
142			}
143
144			/**
145			* Unsubscribe from a channel on behalf of a websocket user.
146			*
147			* @param string $appId
148			* @param string $channel
149			* @return bool
150			*/
151			public function unsubscribe(string $appId, string $channel): bool
152			{
153			if (! isset($this->subscribedChannels["$appId:$channel"])) {
154			return false;
155			}
156
157			// Decrement the subscription count for this channel
158			$this->subscribedChannels["$appId:$channel"]--;
159
160			// If we no longer have subscriptions to that channel, unsubscribe
161			if ($this->subscribedChannels["$appId:$channel"] < 1) {
162			$this->subscribeClient->__call('unsubscribe', ["$appId:$channel"]);
163			unset($this->subscribedChannels["$appId:$channel"]);
164			}
165
166			return true;
167			}
168
169			/**
170			* Publish a message to a channel on behalf of a websocket user.
171			*
172			* @param string $appId
173			* @param string $channel
174			* @param stdClass $payload
175			* @return bool
176			*/
177			public function publish(string $appId, string $channel, stdClass $payload): bool
178			{
179			$payload->appId = $appId;
180			$payload->serverId = $this->serverId;
181
182			$this->publishClient->__call('publish', ["$appId:$channel", json_encode($payload)]);
183
184			return true;
185			}
186
187			/**
188			* Add a member to a channel. To be called when they have
189			* subscribed to the channel.
190			*
191			* @param string $appId
192			* @param string $channel
193			* @param string $socketId
194			* @param string $data
195			*/
196			public function joinChannel(string $appId, string $channel, string $socketId, string $data)
197			{
198			$this->publishClient->__call('hset', ["$appId:$channel", $socketId, $data]);
199			}
200
201			/**
202			* Remove a member from the channel. To be called when they have
203			* unsubscribed from the channel.
204			*
205			* @param string $appId
206			* @param string $channel
207			* @param string $socketId
208			*/
209			public function leaveChannel(string $appId, string $channel, string $socketId)
210			{
211			$this->publishClient->__call('hdel', ["$appId:$channel", $socketId]);
212			}
213
214			/**
215			* Retrieve the full information about the members in a presence channel.
216			*
217			* @param string $appId
218			* @param string $channel
219			* @return PromiseInterface
220			*/
221			public function channelMembers(string $appId, string $channel): PromiseInterface
222			{
223			return $this->publishClient->__call('hgetall', ["$appId:$channel"])
224			->then(function ($members) {
225			// The data is expected as objects, so we need to JSON decode
226			return array_map(function ($user) {
227			return json_decode($user);
228			}, $members);
229			});
230			}
231
232			/**
233			* Get the amount of users subscribed for each presence channel.
234			*
235			* @param string $appId
236			* @param array $channelNames
237			* @return PromiseInterface
238			*/
239			public function channelMemberCounts(string $appId, array $channelNames): PromiseInterface
240			{
241			$this->publishClient->__call('multi', []);
242
243			foreach ($channelNames as $channel) {
244			$this->publishClient->__call('hlen', ["$appId:$channel"]);
245			}
246
247			return $this->publishClient->__call('exec', [])
248			->then(function ($data) use ($channelNames) {
249			return array_combine($channelNames, $data);
250			});
251			}
252
253			/**
254			* Build the Redis connection URL from Laravel database config.
255			*
256			* @return string
257			*/
258			protected function getConnectionUri()
259			{
260			$name = config('websockets.replication.connection') ?? 'default';
261			$config = config("database.redis.$name");
262			$host = $config['host'];
263			$port = $config['port'] ? (':'.$config['port']) : ':6379';
264
265			$query = [];
266			if ($config['password']) {
267			$query['password'] = $config['password'];
268			}
269			if ($config['database']) {
270			$query['database'] = $config['database'];
271			}
272			$query = http_build_query($query);
273
274			return "redis://$host$port".($query ? '?'.$query : '');
275			}
276			}
277

beyondcode / laravel-websockets

Pull Request — master (#140)

RedisClient::subscribe() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like