From 0b613798752d4597c9d6a9bc4b3273c165915b58 Mon Sep 17 00:00:00 2001 From: Henry Jameson Date: Thu, 13 Aug 2026 18:17:19 +0300 Subject: [PATCH] using our own events for socket messages --- src/api/websocket.js | 35 ++++++---- src/components/notification/notification.js | 2 +- src/stores/notifications.js | 32 +++------ src/stores/statuses.js | 9 ++- src/stores/streaming.js | 76 +++++++++++++++++---- 5 files changed, 95 insertions(+), 59 deletions(-) diff --git a/src/api/websocket.js b/src/api/websocket.js index 863e609f5..6f6cfa605 100644 --- a/src/api/websocket.js +++ b/src/api/websocket.js @@ -26,8 +26,26 @@ const PLEROMA_STREAMING_EVENTS = new Set([ 'pleroma:respond', ]) +export const WSConnectionStatus = Object.freeze({ + JOINED: 1, + CLOSED: 2, + ERROR: 3, + DISABLED: 4, + STARTING: 5, + STARTING_INITIAL: 6, +}) + +export class WSEvent extends Event { + data + + constructor(name, data, original) { + super(name) + this.data = data + } +} + // A thin wrapper around WebSocket API that allows adding a pre-processor to it -// Uses EventTarget and a CustomEvent to proxy events +// Uses EventTarget and a WSEvent to proxy events export const ProcessedWS = ({ url, preprocessor = handleMastoWS, @@ -39,9 +57,7 @@ export const ProcessedWS = ({ if (!socket) throw new Error(`Failed to create socket ${id}`) const proxy = (original, eventName, processor = (a) => a) => { original.addEventListener(eventName, (eventData) => { - eventTarget.dispatchEvent( - new CustomEvent(eventName, { detail: processor(eventData) }), - ) + eventTarget.dispatchEvent(new WSEvent(eventName, processor(eventData))) }) } socket.addEventListener('open', (wsEvent) => { @@ -75,7 +91,7 @@ export const ProcessedWS = ({ /**/ const onAuthenticated = () => { - eventTarget.dispatchEvent(new CustomEvent('pleroma:authenticated')) + eventTarget.dispatchEvent(new WSEvent('pleroma:authenticated')) } proxy(socket, 'open') @@ -163,12 +179,3 @@ export const handleMastoWS = ( return null } } - -export const WSConnectionStatus = Object.freeze({ - JOINED: 1, - CLOSED: 2, - ERROR: 3, - DISABLED: 4, - STARTING: 5, - STARTING_INITIAL: 6, -}) diff --git a/src/components/notification/notification.js b/src/components/notification/notification.js index aa3f3561b..44c07383d 100644 --- a/src/components/notification/notification.js +++ b/src/components/notification/notification.js @@ -16,9 +16,9 @@ import { import { useInstanceStore } from 'src/stores/instance.js' import { useMergedConfigStore } from 'src/stores/merged_config.js' import { useOAuthStore } from 'src/stores/oauth.js' +import { useStatusesStore } from 'src/stores/statuses.js' import { useUserHighlightStore } from 'src/stores/user_highlight.js' import { useUsersStore } from 'src/stores/users.js' -import { useStatusesStore } from 'src/stores/statuses.js' import { approveUser, denyUser } from 'src/api/user.js' import generateProfileLink from 'src/services/user_profile_link_generator/user_profile_link_generator' diff --git a/src/stores/notifications.js b/src/stores/notifications.js index c03479c68..078b66b89 100644 --- a/src/stores/notifications.js +++ b/src/stores/notifications.js @@ -40,26 +40,11 @@ export const useNotificationsStore = defineStore('notifications', { // Init attachSocket() { const et = new EventTarget() - const handleNotificationMessage = ({ data, timestamp }) => { - this.addNewNotifications({ data: [data.notification], timestamp }) - } - const notificationHandler = ({ detail: message }) => { - handleNotificationMessage(message) - } - const openHandler = () => this.onStreamConnect() - const closeHandler = () => this.onStreamDisconnect() - const socket = { - et, - handlers: { - openHandler, - closeHandler, - notificationHandler, - }, - } + const socket = { et } - et.addEventListener('notification', notificationHandler) - et.addEventListener('open', openHandler) - et.addEventListener('close', closeHandler) + et.addEventListener('notification', this.addNewNotifications) + et.addEventListener('open', this.onStreamConnect) + et.addEventListener('close', this.onStreamDisconnect) useStreamingStore().addSubscriber(socket) this.socket = socket @@ -82,12 +67,11 @@ export const useNotificationsStore = defineStore('notifications', { } useStreamingStore().removeSubscriber(this.socket) - const { openHandler, closeHandler, notificationHandler } = - this.socket.handlers - this.socket.et.removeEventListener('notification', openHandler) - this.socket.et.removeEventListener('notification', closeHandler) - this.socket.et.removeEventListener('notification', notificationHandler) + const { et } = this.socket + et.removeEventListener('notification', this.addNewNotifications) + et.removeEventListener('open', this.onStreamConnect) + et.removeEventListener('close', this.onStreamDisconnect) const blankState = defaultState() Object.keys(blankState).forEach((k) => { diff --git a/src/stores/statuses.js b/src/stores/statuses.js index e5e82f6cd..40b9da0da 100644 --- a/src/stores/statuses.js +++ b/src/stores/statuses.js @@ -82,11 +82,10 @@ export const useStatusesStore = defineStore('statuses', { // Init attachSocket() { const et = new EventTarget() - const handleStatusMessage = ({ data, timestamp }) => { - this.addNewStatuses({ statuses: [data.status], timestamp }) - } - const handleUpdate = ({ detail: message }) => handleStatusMessage(message) - const handleDelete = ({ detail: message }) => this.setDeleted(message.data.id) + const handleUpdate = ({ data, timestamp }) => + this.addNewStatuses({ statuses: data, timestamp }) + const handleDelete = ({ data }) => + data.forEach((id) => this.setDeleted(id)) const socket = { et, diff --git a/src/stores/streaming.js b/src/stores/streaming.js index d2433be76..57ba9de38 100644 --- a/src/stores/streaming.js +++ b/src/stores/streaming.js @@ -23,6 +23,37 @@ export const TIMELINE_STREAM_MAP = { const retryTimeout = (multiplier) => 1000 * multiplier +export class StreamStateEvent extends Event { + original + + constructor(name, original) { + super(name) + this.original = original + } +} + +export class StreamErrorEvent extends Event { + error + + constructor(error) { + super('error', error) + this.error = error + } +} + +export class StreamMessageEvent extends Event { + data + stream + timestamp + + constructor(name, stream, data) { + super(name) + this.data = data + this.stream = stream + this.timestamp = Date.now() + } +} + export const useStreamingStore = defineStore('streaming', { state: () => ({ socket: null, @@ -56,7 +87,7 @@ export const useStreamingStore = defineStore('streaming', { this.subscribers.add(subscriber) if (this.state === WSConnectionStatus.JOINED) { this.socket.subscribe(...this.getSubArgs(stream)) - et.dispatchEvent(new CustomEvent('open')) + et.dispatchEvent(new StreamStateEvent('open')) } }, removeSubscriber(subscriber) { @@ -105,7 +136,7 @@ export const useStreamingStore = defineStore('streaming', { }, onAuth() { this.subscribers.forEach(({ stream, et }) => { - et.dispatchEvent(new CustomEvent('authenticated')) + et.dispatchEvent(new StreamStateEvent('authenticated')) if (stream) { this.socket.subscribe(...this.getSubArgs(stream)) @@ -115,12 +146,11 @@ export const useStreamingStore = defineStore('streaming', { }, onOpen() { this.subscribers.forEach(({ stream, et }) => { - et.dispatchEvent(new CustomEvent('open')) + et.dispatchEvent(new StreamStateEvent('open')) }) }, - onMessage({ detail: message }) { + onMessage({ data: message }) { if (!message) return // pings - const timestamp = Date.now() const { event: eventName, stream: eventStream, ...data } = message const [streamName, streamArgument] = eventStream ?? [] @@ -131,23 +161,39 @@ export const useStreamingStore = defineStore('streaming', { subscriber, ].filter(Boolean) + const eventData = (() => { + switch (eventName) { + case 'status.update': + case 'update': + return [data.status] + case 'notification': + return [data.notification] + case 'delete': + return [data.id] + default: + return data + } + })() + + const event = new StreamMessageEvent( + eventName, + { name: streamName, argument: streamArgument }, + eventData, + ) + totalSubs.forEach(({ stream, et }) => { - et.dispatchEvent( - new CustomEvent(eventName, { - detail: { streamName, streamArgument, data, timestamp }, - }), - ) + et.dispatchEvent(event) }) console.log('WS', message) }, - onError({ detail: error }) { + onError({ data: error }) { this.subscribers.forEach(({ stream, et }) => { - et.dispatchEvent(new CustomEvent('error', error)) + et.dispatchEvent(new StreamErrorEvent(error)) }) console.error('Error in MastoAPI websocket:', error) }, - onClose({ detail: closeEvent }) { + onClose({ data: closeEvent }) { const ignoreCodes = new Set([ 1000, // Normal (intended) closure 1001, // Going away @@ -162,7 +208,7 @@ export const useStreamingStore = defineStore('streaming', { this.state = WSConnectionStatus.CLOSED this.subscribers.forEach(({ et }) => { - et.dispatchEvent(new CustomEvent('close', closeEvent)) + et.dispatchEvent(new StreamStateEvent('close', closeEvent)) }) } else { console.warn( @@ -177,7 +223,7 @@ export const useStreamingStore = defineStore('streaming', { if (this.state !== WSConnectionStatus.ERROR) { this.subscribers.forEach(({ et }) => { - et.dispatchEvent(new CustomEvent('close', closeEvent)) + et.dispatchEvent(new StreamStateEvent('close', closeEvent)) }) }