Repository: directus/directus
File: sdk/src/realtime/composable.ts (TypeScript)

Code before:
				if (state.code !== 'open') {
					// TODO use directus error
					throw new Error(
						'Cannot send messages without an open connection. Make sure you are calling "await client.connect()".',
					);
				}

				if (typeof message === 'string') {
					return state.connection.send(message);
				}

				if ('uid' in message === false) {
					message['uid'] = uid.next().value;
				}

				state.connection.send(JSON.stringify(message));
			},
			async subscribe<Collection extends keyof Schema, const Options extends SubscribeOptions<Schema, Collection>>(
				collection: Collection,
				options = {} as Options,
			) {
				if ('uid' in options === false) options.uid = uid.next().value;

				if ('query' in options && options.query) {
					options.query = parseQueryParams(options.query);
				}

				subscriptions.add({ ...options, collection, type: 'subscribe' });

				if (state.code !== 'open') {
					debug('info', 'No connection available for subscribing!');
					await this.connect();
				}

				this.sendMessage({ ...options, collection, type: 'subscribe' });
				let subscribed = true;

				async function* subscriptionGenerator(): AsyncGenerator<
					SubscriptionOutput<Schema, Collection, Options['query'], SubscriptionEvents>,
					void,
					unknown
				> {
					while (subscribed && state.code === 'open') {
						const message = await messageCallback(state.connection).catch(() => {
							/* let the loop continue */
						});

						if (!message) continue;

						if (
							'type' in message &&

Review comment (refactor):
Let's simplify this to 
```suggestion
				if (options.query) {
					options.query = queryToParams({ ...options.query });
				}
```

Code after:
				if (state.code !== 'open') {
					// TODO use directus error
					throw new Error(
						'Cannot send messages without an open connection. Make sure you are calling "await client.connect()".',
					);
				}

				if (typeof message === 'string') {
					return state.connection.send(message);
				}

				if ('uid' in message === false) {
					message['uid'] = uid.next().value;
				}

				state.connection.send(JSON.stringify(message));
			},
			async subscribe<Collection extends keyof Schema, const Options extends SubscribeOptions<Schema, Collection>>(
				collection: Collection,
				options = {} as Options,
			) {
				if ('uid' in options === false) options.uid = uid.next().value;

				if (options.query) {
					options.query = queryToParams(options.query as ExtendedQuery<Schema, Schema[Collection]>);
				}

				subscriptions.add({ ...options, collection, type: 'subscribe' });

				if (state.code !== 'open') {
					debug('info', 'No connection available for subscribing!');
					await this.connect();
				}

				this.sendMessage({ ...options, collection, type: 'subscribe' });
				let subscribed = true;

				async function* subscriptionGenerator(): AsyncGenerator<
					SubscriptionOutput<Schema, Collection, Options['query'], SubscriptionEvents>,
					void,
					unknown
				> {
					while (subscribed && state.code === 'open') {
						const message = await messageCallback(state.connection).catch(() => {
							/* let the loop continue */
						});

						if (!message) continue;

						if (
							'type' in message &&