using our own events for socket messages

This commit is contained in:
Henry Jameson 2026-08-13 18:17:19 +03:00
commit 0b61379875
5 changed files with 95 additions and 59 deletions

View file

@ -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,
})

View file

@ -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'

View file

@ -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) => {

View file

@ -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,

View file

@ -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))
})
}