refactor: unify useStream and useSingleStream (#31326)

apps/meteor/client/hooks/useAppActionButtons.ts

@@ -1,6 +1,6 @@
 import type { IUIActionButton, UIActionButtonContext } from '@rocket.chat/apps-engine/definition/ui';
 import { useDebouncedCallback } from '@rocket.chat/fuselage-hooks';
-import { useEndpoint, useSingleStream, useToastMessageDispatch, useUserId } from '@rocket.chat/ui-contexts';
+import { useEndpoint, useStream, useToastMessageDispatch, useUserId } from '@rocket.chat/ui-contexts';
 import type { UseQueryResult } from '@tanstack/react-query';
 import { useQuery, useQueryClient } from '@tanstack/react-query';
 import { useEffect, useMemo } from 'react';
@@ -19,7 +19,7 @@ const getIdForActionButton = ({ appId, actionId }: IUIActionButton): string => `
 export const useAppActionButtons = <TContext extends `${UIActionButtonContext}`>(context?: TContext) => {
 	const queryClient = useQueryClient();
 
-	const apps = useSingleStream('apps');
+	const apps = useStream('apps');
 	const uid = useUserId();
 
 	const getActionButtons = useEndpoint('GET', '/apps/actionButtons');

apps/meteor/client/hooks/useAppSlashCommands.ts

@@ -1,5 +1,5 @@
 import { useDebouncedCallback } from '@rocket.chat/fuselage-hooks';
-import { useEndpoint, useSingleStream, useUserId } from '@rocket.chat/ui-contexts';
+import { useEndpoint, useStream, useUserId } from '@rocket.chat/ui-contexts';
 import { useQuery, useQueryClient } from '@tanstack/react-query';
 import { useEffect } from 'react';
 
@@ -8,7 +8,7 @@ import { slashCommands } from '../../app/utils/lib/slashCommand';
 export const useAppSlashCommands = () => {
 	const queryClient = useQueryClient();
 
-	const apps = useSingleStream('apps');
+	const apps = useStream('apps');
 	const uid = useUserId();
 
 	const invalidate = useDebouncedCallback(

apps/meteor/client/hooks/useLicense.ts

@@ -1,6 +1,6 @@
 import type { Serialized } from '@rocket.chat/core-typings';
 import type { OperationResult } from '@rocket.chat/rest-typings';
-import { useEndpoint, useSingleStream, useUserId } from '@rocket.chat/ui-contexts';
+import { useEndpoint, useStream, useUserId } from '@rocket.chat/ui-contexts';
 import type { QueryClient, UseQueryResult } from '@tanstack/react-query';
 import { useQuery, useQueryClient } from '@tanstack/react-query';
 import { useEffect } from 'react';
@@ -36,7 +36,7 @@ export const useLicenseBase = <TData = LicenseDataType>({
 
 	const invalidateQueries = useInvalidateLicense();
 
-	const notify = useSingleStream('notify-all');
+	const notify = useStream('notify-all');
 
 	useEffect(() => notify('license', () => invalidateQueries()), [notify, invalidateQueries]);
 

apps/meteor/client/hooks/useTranslationsForApps.ts

@@ -1,5 +1,5 @@
 import { normalizeLanguage } from '@rocket.chat/tools';
-import { useEndpoint, useSingleStream, useUserId } from '@rocket.chat/ui-contexts';
+import { useEndpoint, useStream, useUserId } from '@rocket.chat/ui-contexts';
 import { useQuery, useQueryClient } from '@tanstack/react-query';
 import { useEffect } from 'react';
 import { useTranslation } from 'react-i18next';
@@ -32,7 +32,7 @@ export const useTranslationsForApps = () => {
 	}, [i18n, data, isSuccess]);
 
 	const queryClient = useQueryClient();
-	const subscribeToApps = useSingleStream('apps');
+	const subscribeToApps = useStream('apps');
 	const uid = useUserId();
 
 	useEffect(() => {

apps/meteor/client/providers/AppsProvider.tsx

@@ -1,5 +1,5 @@
 import { useDebouncedCallback } from '@rocket.chat/fuselage-hooks';
-import { usePermission, useSingleStream } from '@rocket.chat/ui-contexts';
+import { usePermission, useStream } from '@rocket.chat/ui-contexts';
 import { useQuery, useQueryClient } from '@tanstack/react-query';
 import type { FC } from 'react';
 import React, { useEffect } from 'react';
@@ -39,7 +39,7 @@ const AppsProvider: FC = ({ children }) => {
 	const invalidateAppsCountQuery = useInvalidateAppsCountQueryCallback();
 	const invalidateLicenseQuery = useInvalidateLicense();
 
-	const stream = useSingleStream('apps');
+	const stream = useStream('apps');
 
 	const invalidate = useDebouncedCallback(
 		() => {

apps/meteor/client/providers/ServerProvider.tsx

@@ -59,54 +59,53 @@ const callEndpoint = <TMethod extends Method, TPathPattern extends PathPattern>(
 
 const uploadToEndpoint = (endpoint: PathFor<'POST'>, formData: any): Promise<UploadResult> => sdk.rest.post(endpoint as any, formData);
 
-const getStream = <N extends StreamNames, K extends StreamKeys<N>>(
-	streamName: N,
-	_options?: {
-		retransmit?: boolean | undefined;
-		retransmitToSelf?: boolean | undefined;
-	},
-): ((eventName: K, callback: (...args: StreamerCallbackArgs<N, K>) => void) => () => void) => {
-	return (eventName, callback): (() => void) => {
-		return sdk.stream(streamName, [eventName], callback as (...args: any[]) => void).stop;
-	};
+type EventMap<N extends StreamNames = StreamNames, K extends StreamKeys<N> = StreamKeys<N>> = {
+	[key in `${N}/${K}`]: StreamerCallbackArgs<N, K>;
 };
 
-const ee = new Emitter<Record<string, void>>();
+const ee = new Emitter<EventMap>();
 
 const events = new Map<string, () => void>();
 
-const getSingleStream = <N extends StreamNames, K extends StreamKeys<N>>(
+const getStream = <N extends StreamNames>(
 	streamName: N,
 	_options?: {
 		retransmit?: boolean | undefined;
 		retransmitToSelf?: boolean | undefined;
 	},
-): ((eventName: K, callback: (...args: StreamerCallbackArgs<N, K>) => void) => () => void) => {
-	const stream = getStream(streamName);
-	return (eventName, callback): (() => void) => {
-		ee.on(`${streamName}/${eventName}`, callback);
+) => {
+	return <K extends StreamKeys<N>>(eventName: K, callback: (...args: StreamerCallbackArgs<N, K>) => void): (() => void) => {
+		const eventLiteral = `${streamName}/${eventName}` as const;
+		const emitterCallback = (args?: unknown): void => {
+			if (!args || !Array.isArray(args)) {
+				throw new Error('Invalid streamer callback');
+			}
+			callback(...(args as StreamerCallbackArgs<N, K>));
+		};
+
+		ee.on(eventLiteral, emitterCallback);
 
-		const handler = (...args: any[]): void => {
-			ee.emit(`${streamName}/${eventName}`, ...args);
+		const streamHandler = (...args: StreamerCallbackArgs<N, K>): void => {
+			ee.emit(eventLiteral, args);
 		};
 
 		const stop = (): void => {
 			// If someone is still listening, don't unsubscribe
-			ee.off(`${streamName}/${eventName}`, callback);
+			ee.off(eventLiteral, emitterCallback);
 
-			if (ee.has(`${streamName}/${eventName}`)) {
+			if (ee.has(eventLiteral)) {
 				return;
 			}
 
-			const unsubscribe = events.get(`${streamName}/${eventName}`);
+			const unsubscribe = events.get(eventLiteral);
 			if (unsubscribe) {
 				unsubscribe();
-				events.delete(`${streamName}/${eventName}`);
+				events.delete(eventLiteral);
 			}
 		};
 
-		if (!events.has(`${streamName}/${eventName}`)) {
-			events.set(`${streamName}/${eventName}`, stream(eventName, handler));
+		if (!events.has(eventLiteral)) {
+			events.set(eventLiteral, sdk.stream(streamName, [eventName], streamHandler).stop);
 		}
 		return stop;
 	};
@@ -119,7 +118,6 @@ const contextValue = {
 	callEndpoint,
 	uploadToEndpoint,
 	getStream,
-	getSingleStream,
 };
 
 const ServerProvider: FC = ({ children }) => <ServerContext.Provider children={children} value={contextValue} />;

apps/meteor/tests/mocks/client/ServerProviderMock.tsx

@@ -56,7 +56,6 @@ const uploadToEndpoint = async () => {
 	throw new Error('not implemented');
 }; // to be implemented
 const getStream = () => () => () => undefined; // to be implemented
-const getSingleStream = () => () => () => undefined; // to be implemented
 const callEndpoint = () => {
 	throw new Error('not implemented');
 }; // to be implemented
@@ -67,7 +66,6 @@ const contextValue = {
 	// callMethod,
 	callEndpoint,
 	uploadToEndpoint,
-	getSingleStream,
 	getStream,
 };
 

packages/fuselage-ui-kit/src/blocks/VideoConferenceBlock/hooks/useVideoConfDataStream.ts

@@ -1,5 +1,5 @@
 import type { IRoom } from '@rocket.chat/core-typings';
-import { useSingleStream } from '@rocket.chat/ui-contexts';
+import { useStream } from '@rocket.chat/ui-contexts';
 import { useQueryClient } from '@tanstack/react-query';
 import { useEffect } from 'react';
 
@@ -14,7 +14,7 @@ export const useVideoConfDataStream = ({
 }) => {
   const queryClient = useQueryClient();
 
-  const subscribeNotifyRoom = useSingleStream('notify-room');
+  const subscribeNotifyRoom = useStream('notify-room');
 
   useEffect(() => {
     return subscribeNotifyRoom(

packages/mock-providers/src/MockedAppRootBuilder.tsx

@@ -48,7 +48,6 @@ export class MockedAppRootBuilder {
 		}): Promise<Serialized<OperationResult<TMethod, TPathPattern>>> => {
 			throw new Error('not implemented');
 		},
-		getSingleStream: () => () => () => undefined,
 		getStream: () => () => () => undefined,
 		uploadToEndpoint: () => Promise.reject(new Error('not implemented')),
 		callMethod: () => Promise.reject(new Error('not implemented')),

packages/ui-contexts/src/ServerContext/ServerContext.ts

@@ -38,13 +38,6 @@ export type ServerContextValue = {
 			retransmitToSelf?: boolean | undefined;
 		},
 	) => (eventName: K, callback: (...args: StreamerCallbackArgs<N, K>) => void) => () => void;
-	getSingleStream: <N extends StreamNames, K extends StreamKeys<N>>(
-		streamName: N,
-		_options?: {
-			retransmit?: boolean | undefined;
-			retransmitToSelf?: boolean | undefined;
-		},
-	) => (eventName: K, callback: (...args: StreamerCallbackArgs<N, K>) => void) => () => void;
 };
 
 export const ServerContext = createContext<ServerContextValue>({
@@ -57,5 +50,4 @@ export const ServerContext = createContext<ServerContextValue>({
 		throw new Error('not implemented');
 	},
 	getStream: () => () => (): void => undefined,
-	getSingleStream: () => () => (): void => undefined,
 });

packages/ui-contexts/src/hooks/useStream.ts

@@ -28,21 +28,3 @@ export function useStream<N extends StreamNames>(
 	const { getStream } = useContext(ServerContext);
 	return useMemo(() => getStream(streamName, options), [getStream, streamName, options]);
 }
-
-/*
- * @param streamName The name of the stream to subscribe to
- * @returns A function that can be used to subscribe to the stream
- * the main difference between this and useStream is that this function
- * will only subscribe to the `stream + key` only once, but you can still add multiple callbacks
- * to the same path
- */
-export function useSingleStream<N extends StreamNames>(
-	streamName: N,
-	options?: {
-		retransmit?: boolean;
-		retransmitToSelf?: boolean;
-	},
-): StreamerCallback<N> {
-	const { getSingleStream } = useContext(ServerContext);
-	return useMemo(() => getSingleStream(streamName, options), [getSingleStream, streamName, options]);
-}

packages/ui-contexts/src/index.ts

@@ -66,7 +66,7 @@ export { useSettings } from './hooks/useSettings';
 export { useSettingsDispatch } from './hooks/useSettingsDispatch';
 export { useSettingSetValue } from './hooks/useSettingSetValue';
 export { useSettingStructure } from './hooks/useSettingStructure';
-export { useStream, useSingleStream } from './hooks/useStream';
+export { useStream } from './hooks/useStream';
 export { useToastMessageDispatch } from './hooks/useToastMessageDispatch';
 export { useTooltipClose } from './hooks/useTooltipClose';
 export { useTooltipOpen } from './hooks/useTooltipOpen';