[NEW] Stream to get individual presence updates

app/notifications/client/lib/Notifications.js

@@ -1,6 +1,8 @@
 import { Meteor } from 'meteor/meteor';
 import { Tracker } from 'meteor/tracker';
 
+import './Presence';
+
 class Notifications {
 	constructor(...args) {
 		this.logged = Meteor.userId() !== null;
@@ -17,6 +19,7 @@ class Notifications {
 		this.streamRoom = new Meteor.Streamer('notify-room');
 		this.streamRoomUsers = new Meteor.Streamer('notify-room-users');
 		this.streamUser = new Meteor.Streamer('notify-user');
+
 		if (this.debug === true) {
 			this.onAll(function() {
 				return console.log('RocketChat.Notifications: onAll', args);

app/notifications/client/lib/Presence.ts

@@ -0,0 +1,13 @@
+import { Meteor } from 'meteor/meteor';
+
+import { Presence, STATUS_MAP } from '../../../../client/lib/presence';
+
+// TODO implement API on Streamer to be able to listen to all streamed data
+// this is a hacky way to listen to all streamed data from user-presense Streamer
+(Meteor as any).StreamerCentral.on('stream-user-presence', (uid: string, args: unknown) => {
+	if (!Array.isArray(args)) {
+		throw new Error('Presence event must be an array');
+	}
+	const [username, status, statusText] = args as [string, number, string | undefined];
+	Presence.notify({ _id: uid, username, status: STATUS_MAP[status], statusText });
+});

app/notifications/server/lib/Notifications.ts

@@ -11,6 +11,7 @@ import {
 	Users as UsersRaw,
 	Settings as SettingsRaw,
 } from '../../../models/server/raw';
+import './Presence';
 
 // TODO: Replace this in favor of the api.broadcast
 // StreamerCentral.on('broadcast', (name, eventName, args) => {

app/notifications/server/lib/Presence.ts

@@ -0,0 +1,103 @@
+import { Emitter } from '@rocket.chat/emitter';
+
+import { IUser } from '../../../../definition/IUser';
+import { IPublication, IStreamerConstructor, Connection, IStreamer } from '../../../../server/modules/streamer/streamer.module';
+
+export type UserPresenseStreamProps = {
+	added: IUser['_id'][];
+	removed: IUser['_id'][];
+}
+
+export type UserPresenseStreamArgs = {
+	'uid': string;
+	args: unknown;
+}
+
+const e = new Emitter<{
+	[key: string]: UserPresenseStreamArgs;
+}>();
+
+
+const clients = new WeakMap<Connection, UserPresence>();
+
+
+export class UserPresence {
+	private readonly streamer: IStreamer;
+
+	private readonly publication: IPublication;
+
+	private readonly listeners: Set<string>;
+
+	constructor(publication: IPublication, streamer: IStreamer) {
+		this.listeners = new Set();
+		this.publication = publication;
+		this.streamer = streamer;
+	}
+
+	listen(uid: string): void {
+		if (this.listeners.has(uid)) {
+			return;
+		}
+		e.on(uid, this.run);
+		this.listeners.add(uid);
+	}
+
+	off = (uid: string): void => {
+		e.off(uid, this.run);
+		this.listeners.delete(uid);
+	}
+
+	run = (args: UserPresenseStreamArgs): void => {
+		const payload = this.streamer.changedPayload(this.streamer.subscriptionName, args.uid, { ...args, eventName: args.uid }); // there is no good explanation to keep eventName, I just want to save one 'DDPCommon.parseDDP' on the client side, so I'm trying to fit the Meteor Streamer's payload
+		(this.publication as any)._session.socket.send(payload);
+	}
+
+	stop(): void {
+		this.listeners.forEach(this.off);
+		clients.delete(this.publication.connection);
+	}
+
+	static getClient(publication: IPublication, streamer: IStreamer): [UserPresence, boolean] {
+		const { connection } = publication;
+		const stored = clients.get(connection);
+
+		const client = stored || new UserPresence(publication, streamer);
+
+		const main = Boolean(!stored);
+
+		clients.set(connection, client);
+
+		return [client, main];
+	}
+}
+
+export class StreamPresence {
+	static getInstance(Streamer: IStreamerConstructor, name = 'user-presence'): IStreamer {
+		return new class StreamPresence extends Streamer {
+			async _publish(publication: IPublication, _eventName: string, options: boolean | {useCollection?: boolean; args?: any} = false): Promise<void> {
+				const { added, removed } = (typeof options !== 'boolean' ? options : {}) as unknown as UserPresenseStreamProps;
+
+
+				const [client, main] = UserPresence.getClient(publication, this);
+
+				added?.forEach((uid) => client.listen(uid));
+				removed?.forEach((uid) => client.off(uid));
+
+
+				if (!main) {
+					publication.stop();
+					return;
+				}
+
+				publication.ready();
+
+				publication.onStop(() => client.stop());
+			}
+		}(name);
+	}
+}
+
+
+export const emit = (uid: string, args: UserPresenseStreamArgs): void => {
+	e.emit(uid, { uid, args });
+};

app/ui-message/client/message.html

@@ -40,7 +40,6 @@
 					<button type="button" class="user user-card-message color-primary-font-color" data-username="{{msg.u.username}}" tabindex="1">
 						{{getName}}{{#if showUsername}} <span class="message-alias border-component-color color-info-font-color">@{{msg.u.username}}</span>{{/if}}
 					</button>
-					{{> StatusMessage uid=msg.u._id}}
 					<span class="info border-component-color color-info-font-color"></span>
 					{{# if showRoles }}
 						{{#each role in roleTags}}

app/ui-sidenav/client/userPresence.js

@@ -2,58 +2,12 @@ import { Meteor } from 'meteor/meteor';
 import { Accounts } from 'meteor/accounts-base';
 import { Template } from 'meteor/templating';
 import { Tracker } from 'meteor/tracker';
-import _ from 'underscore';
-import mem from 'mem';
 
-import { APIClient } from '../../utils/client';
-import { saveUser, interestedUserIds } from '../../../client/startup/listenActiveUsers';
 import { Presence } from '../../../client/lib/presence';
 
 import './userPresence.html';
 
 const data = new Map();
-const promises = new Map();
-const pending = new Map();
-
-const getAll = _.debounce(async function getAll() {
-	const ids = Array.from(pending.keys());
-
-	if (ids.length === 0) {
-		return;
-	}
-
-	const params = {
-		ids,
-	};
-
-	try {
-		const {
-			users,
-		} = await APIClient.v1.get('users.presence', params);
-
-		users.forEach((user) => saveUser(user, true));
-
-		ids.forEach((id) => {
-			const { resolve } = promises.get(id);
-			resolve();
-		});
-	} catch (e) {
-		ids.forEach((id) => {
-			const { reject } = promises.get(id);
-			reject();
-		});
-	}
-}, 100);
-
-export const get = mem(function get(id) {
-	interestedUserIds.add(id);
-	const promise = pending.get(id) || new Promise((resolve, reject) => {
-		promises.set(id, { resolve, reject });
-	});
-	pending.set(id, promise);
-	return promise;
-});
-
 const options = {
 	threshold: 0.1,
 };
@@ -63,10 +17,8 @@ const handleEntries = function(entries) {
 	lastEntries = entries.filter(({ isIntersecting }) => isIntersecting);
 	lastEntries.forEach(async (entry) => {
 		const { uid } = data.get(entry.target);
-		await get(uid);
-		pending.delete(uid);
+		Presence.get(uid);
 	});
-	getAll();
 };
 
 const featureExists = !!window.IntersectionObserver;
@@ -80,7 +32,6 @@ Tracker.autorun(() => {
 		Presence.reset();
 		return Meteor.users.update({ status: { $exists: true } }, { $unset: { status: true } }, { multi: true });
 	}
-	mem.clear(get);
 
 	Presence.restart();
 
@@ -93,11 +44,8 @@ Tracker.autorun(() => {
 	}
 
 
-	getAll();
-
 	Accounts.onLogout(() => {
 		Presence.reset();
-		interestedUserIds.clear();
 	});
 });
 

client/components/Message/StatusMessage.tsx

@@ -1,10 +1,14 @@
 import { Box, Icon } from '@rocket.chat/fuselage';
-import React, { ReactElement, memo } from 'react';
+import React, { ReactElement, memo, useEffect } from 'react';
 
-import { useUserData } from '../../hooks/useUserData';
+import { usePresence } from '../../hooks/usePresence';
 
+// TODO: deprecate this component
 const StatusMessage = ({ uid }: { uid: string }): ReactElement | null => {
-	const data = useUserData(uid);
+	const data = usePresence(uid);
+	useEffect(() => {
+		process.env.NODE_ENV === 'development' && console.warn('StatusMessage component is deprecated');
+	}, [data]);
 
 	if (!data || !data.statusText) {
 		return null;

client/components/UserStatus/ReactiveUserStatus.js

@@ -4,7 +4,7 @@ import { usePresence } from '../../hooks/usePresence';
 import UserStatus from './UserStatus';
 
 const ReactiveUserStatus = ({ uid, ...props }) => {
-	const status = usePresence(uid);
+	const status = usePresence(uid)?.status;
 	return <UserStatus status={status} {...props} />;
 };
 

client/hooks/usePresence.ts

@@ -1,14 +1,30 @@
-import { useUserData } from './useUserData';
+import { useMemo } from 'react';
+import { useSubscription } from 'use-subscription';
+
+import { Presence, UserPresence } from '../lib/presence';
 
 type Presence = 'online' | 'offline' | 'busy' | 'away' | 'loading';
 
 /**
  * Hook to fetch and subscribe users presence
  *
  * @param uid - User Id
- * @returns User Presence - 'online' | 'offline' | 'busy' | 'away' | 'loading'
+ * @returns UserPresence
  * @public
  */
+export const usePresence = (uid: string): UserPresence | undefined => {
+	const subscription = useMemo(
+		() => ({
+			getCurrentValue: (): UserPresence | undefined => (uid ? Presence.store.get(uid) : undefined),
+			subscribe: (callback: any): any => {
+				uid && Presence.listen(uid, callback);
+				return (): void => {
+					uid && Presence.stop(uid, callback);
+				};
+			},
+		}),
+		[uid],
+	);
 
-export const usePresence = (uid: string, presence: Presence): Presence =>
-	useUserData(uid)?.status || presence;
+	return useSubscription(subscription);
+};