diff --git a/@xen-orchestra/rest-api/src/abstract-classes/listener.mts b/@xen-orchestra/rest-api/src/abstract-classes/listener.mts new file mode 100644 index 0000000000..ba86b6b61e --- /dev/null +++ b/@xen-orchestra/rest-api/src/abstract-classes/listener.mts @@ -0,0 +1,81 @@ +import type { EventEmitter } from 'node:events' +import type { Subscriber } from '../events/event.class.mjs' +import type { EventType } from '../events/event.type.mjs' + +export abstract class Listener { + #subscribers: Map = new Map() + #eventEmitter: EventEmitter + #eventCallbacks: Map void> = new Map() + #watchedEvent: EventType[] + + constructor(eventEmitter: EventEmitter, watchedEvent: EventType[]) { + this.#eventEmitter = eventEmitter + this.#watchedEvent = watchedEvent + } + + get subscribers() { + return this.#subscribers + } + + get eventEmitter() { + return this.#eventEmitter + } + + addSubscriber(subscriber: Subscriber, fields: '*' | string[] = '*') { + this.#subscribers.set(subscriber.id, { fields, subscriber }) + + if (this.#subscribers.size === 1) { + this.#watchedEvent.forEach(event => this.#addEventListener(event)) + } + } + + removeSubscriber(subscriberId: Subscriber['id']) { + this.#subscribers.delete(subscriberId) + + if (this.#subscribers.size === 0) { + this.removeAllEventListeners() + } + } + + removeAllEventListeners() { + this.#eventCallbacks.forEach((cb, event) => { + this.#eventEmitter.off(event, cb) + }) + } + + clear() { + this.removeAllEventListeners() + this.#eventCallbacks.clear() + this.#subscribers.clear() + this.#watchedEvent = [] + } + + #addEventListener(event: EventType) { + if (this.#eventCallbacks.has(event)) { + return + } + + const broadcastAllSubscriber = (...args: unknown[]) => { + this.#subscribers.forEach(conf => { + if (!conf.subscriber.isAlive) { + this.removeSubscriber(conf.subscriber.id) + return + } + + const data = this.handleData({ ...conf, event }, ...args) + if (data === undefined) { + return + } + + conf.subscriber.broadcast(event, data) + }) + } + this.#eventCallbacks.set(event, broadcastAllSubscriber) + this.#eventEmitter.on(event, broadcastAllSubscriber) + } + + abstract handleData( + conf: { fields: '*' | string[]; subscriber: Subscriber; event: EventType }, + ...args: unknown[] + ): object | undefined +} diff --git a/@xen-orchestra/rest-api/src/events/event.class.mts b/@xen-orchestra/rest-api/src/events/event.class.mts new file mode 100644 index 0000000000..641c04bc82 --- /dev/null +++ b/@xen-orchestra/rest-api/src/events/event.class.mts @@ -0,0 +1,165 @@ +import pick from 'lodash/pick.js' +import isEqual from 'lodash/isEqual.js' +import { createLogger } from '@xen-orchestra/log' +import type { Response } from 'express' +import { EventEmitter } from 'node:events' +import { noSuchObject } from 'xo-common/api-errors.js' + +import { Listener } from '../abstract-classes/listener.mjs' +import type { CollectionEventType, EventType, SubscriberId, XapiXoListenerType } from './event.type.mjs' +import type { XapiXoRecord, XoAlarm } from '@vates/types' +import type { AlarmService } from '../alarms/alarm.service.mjs' + +const log = createLogger('xo:rest-api:event-service') + +export class Subscriber { + #id: SubscriberId + #manager: SubscriberManager + #connection: Response + #isAlive: boolean + + get id() { + return this.#id + } + + get isAlive() { + return this.#isAlive + } + + get connection() { + return this.#connection + } + + constructor(res: Response, manager: SubscriberManager) { + this.#id = crypto.randomUUID() as SubscriberId + + const headers = new Headers({ + 'Content-Type': 'text/event-stream', + Connection: 'keep-alive', + 'Cache-Control': 'no-cache, no-transform', + }) + res.setHeaders(headers) + res.on('close', () => this.clear()) + + manager.addSubscriber(this) + + this.#connection = res + this.#manager = manager + this.#isAlive = true + } + + broadcast(event: EventType | 'init', data: object) { + if (!this.#isAlive) { + log.warn('broadcast called on a subscriber that is not alive, but still in memory! Force clear and do nothing') + this.clear() + return + } + this.#connection.write(`event:${event}\n`) + this.#connection.write(`data:${JSON.stringify(data)}\n\n`) + } + + clear() { + this.#isAlive = false + this.#manager.removeSubscriber(this.id) + } +} + +export class XapiXoListener extends Listener { + #type: XapiXoListenerType + #alarmService?: AlarmService + + constructor(type: XapiXoListenerType, eventEmitter: EventEmitter, alarmService?: AlarmService) { + super(eventEmitter, ['add', 'update', 'remove']) + this.#type = type + this.#alarmService = alarmService + } + + handleData>( + { fields, event }: { fields: '*' | string[]; subscriber: Subscriber; event: CollectionEventType }, + object: T | undefined, + previousObj?: T + ): Partial | undefined { + let _object: Partial | undefined = object + let _prevObject: Partial | undefined = previousObj + + if (this.#type === 'alarm') { + if (object?.type === 'message' && this.#alarmService?.isAlarm(object)) { + _object = this.#alarmService.parseAlarm(object) + } + if (previousObj?.type === 'message' && this.#alarmService?.isAlarm(previousObj)) { + _prevObject = this.#alarmService.parseAlarm(previousObj) + } + } + + if (fields !== '*') { + if (_object !== undefined) { + _object = pick(_object, fields) + } + if (_prevObject !== undefined) { + _prevObject = pick(_prevObject, fields) + } + } + + if (event === 'update' && fields !== '*' && isEqual(_object, _prevObject)) { + // if no changes from user perspective, don't send update + return + } + + // if _object === undefined, this means we are on a remove event, so _prevObject will not be undefined + return _object ?? _prevObject! + } +} + +export class PingListener extends Listener { + #intervalId: NodeJS.Timeout + constructor() { + super(new EventEmitter(), ['ping']) + + this.#intervalId = setInterval(() => { + this.eventEmitter.emit('ping') + }, 1000 * 30) + } + + handleData(): { ping: number } { + return { ping: Date.now() } + } + + clear(): void { + clearInterval(this.#intervalId) + super.clear() + } +} + +export class SubscriberManager { + #subscribers: Map = new Map() + + get subscribers() { + return this.#subscribers + } + + getSubscriber(id: SubscriberId) { + const subscriber = this.#subscribers.get(id) + if (subscriber === undefined) { + throw noSuchObject(id, 'event') + } + + return subscriber + } + + addSubscriber(subscriber: Subscriber) { + this.#subscribers.set(subscriber.id, subscriber) + } + + removeSubscriber(id: string) { + this.#subscribers.delete(id) + } + + clear() { + this.#subscribers.forEach(subscriber => { + if (!subscriber.connection.closed) { + subscriber.connection.destroy() + } + this.#subscribers.delete(subscriber.id) + }) + } +} diff --git a/@xen-orchestra/rest-api/src/events/event.controller.mts b/@xen-orchestra/rest-api/src/events/event.controller.mts new file mode 100644 index 0000000000..8289a00de4 --- /dev/null +++ b/@xen-orchestra/rest-api/src/events/event.controller.mts @@ -0,0 +1,109 @@ +import { + Body, + Controller, + Example, + Delete, + Get, + Middlewares, + Path, + Post, + Request, + Response, + Route, + Security, + SuccessResponse, + Tags, +} from 'tsoa' + +import { + badRequestResp, + createdResp, + noContentResp, + notFoundResp, + unauthorizedResp, +} from '../open-api/common/response.common.mjs' +import { provide } from 'inversify-binding-decorators' +import { inject } from 'inversify' +import { EventService } from './event.service.mjs' +import type { AuthenticatedRequest } from '../helpers/helper.type.mjs' +import { json } from 'express' +import type { SubscriberId, XapiXoListenerType } from './event.type.mjs' +import { addSubscription } from '../open-api/oa-examples/event.oa-example.mjs' + +@Route('events') +@Security('*') +@Response(badRequestResp.status, badRequestResp.description) +@Response(unauthorizedResp.status, unauthorizedResp.description) +@Tags('events') +@provide(EventController) +export class EventController extends Controller { + #eventService: EventService + + constructor(@inject(EventService) eventService: EventService) { + super() + this.#eventService = eventService + } + + /** + * Opens an SSE (Server-Sent Events) connection. + * + * By default, there are no active subscriptions in the stream. + * To add subscriptions, use the following endpoint: + * + * POST /rest/v0/events/:id/subscriptions + * + * + * Events you will receive: + * - **init**: The first event you will receive. + * Data: the connection ID. + * + * - **ping**: A simple event used to keep the connection alive between the server and the client. + * Data: the event timestamp. + * + * - **add**: Triggered when an object has been added. + * Data: the added object. + * + * - **update**: Triggered when an object has been updated. + * Data: the updated object. + * + * - **remove**: Triggered when an object has been removed. + * Data: the removed object. + */ + @Get('') + @SuccessResponse(200, 'OK') + openSseConnection(@Request() req: AuthenticatedRequest) { + this.#eventService.createSseSubscriber(req.res) + } + + /** + * Add a subscription + * + * @example id "0d8b28c6-e9bf-4c9d-a382-3c9e0d7cfbff" + * @example body {"collection": "VM", "fields": ["id", "name_label"]} + */ + @Example(addSubscription) + @Post('{id}/subscriptions') + @Middlewares(json()) + @SuccessResponse(createdResp.status, createdResp.description) + @Response(notFoundResp.status, notFoundResp.description) + addSubscription( + @Path() id: string, + @Body() body: { collection: XapiXoListenerType; fields?: '*' | string[] } + ): { id: string } { + this.#eventService.addListenerFor(id as SubscriberId, { ...body, type: body.collection }) + return { id: body.collection } + } + + /** + * Remove a subscription + * + * @example id "0d8b28c6-e9bf-4c9d-a382-3c9e0d7cfbff" + * @example subscriptionId: "VM" + */ + @Delete('{id}/subscriptions/{subscriptionId}') + @SuccessResponse(noContentResp.status, noContentResp.description) + @Response(notFoundResp.status, notFoundResp.description) + removeSubscription(@Path() id: string, @Path() subscriptionId: XapiXoListenerType): void { + this.#eventService.removeListenerFor(id as SubscriberId, subscriptionId) + } +} diff --git a/@xen-orchestra/rest-api/src/events/event.service.mts b/@xen-orchestra/rest-api/src/events/event.service.mts new file mode 100644 index 0000000000..35d0772815 --- /dev/null +++ b/@xen-orchestra/rest-api/src/events/event.service.mts @@ -0,0 +1,71 @@ +import { createLogger } from '@xen-orchestra/log' +import type { Response } from 'express' + +import type { Listener } from '../abstract-classes/listener.mjs' +import { PingListener, Subscriber, SubscriberManager, XapiXoListener } from './event.class.mjs' +import type { ListenerType, SubscriberId } from './event.type.mjs' +import { AlarmService } from '../alarms/alarm.service.mjs' +import type { RestApi } from '../rest-api/rest-api.mjs' + +const log = createLogger('xo:rest-api:event-service') + +export class EventService { + #alarmService: AlarmService + #restApi: RestApi + #listeners: Map = new Map() + #subscriberManager = new SubscriberManager() + + constructor(restApi: RestApi) { + process.on('SIGTERM', () => { + log.debug(`SIGTERM received, close all SSE subscribers (nb: ${this.#subscriberManager.subscribers.size})`) + this.#subscriberManager.clear() + this.#listeners.forEach(l => l.clear()) + this.#listeners.clear() + }) + + this.#restApi = restApi + this.#alarmService = restApi.ioc.get(AlarmService) + } + + #getListener(type: ListenerType): Listener { + if (!this.#listeners.has(type)) { + if (type === 'ping') { + this.#listeners.set(type, new PingListener()) + } else { + const isAlarm = type === 'alarm' + // alarm is purely XO-related; it doesn't exist at the XAPI level. + // alarm is a message with parsed values. So, in the case of an alarm listener, it listens for message collection. + const ee = this.#restApi.xoApp.objects.allIndexes.type.getEventEmitterByType(isAlarm ? 'message' : type) + this.#listeners.set(type, new XapiXoListener(type, ee, isAlarm ? this.#alarmService : undefined)) + } + } + + return this.#listeners.get(type)! + } + + createSseSubscriber(res: Response): SubscriberId { + const subscriber = new Subscriber(res, this.#subscriberManager) + subscriber.broadcast('init', { id: subscriber.id }) + + this.addListenerFor(subscriber.id, { type: 'ping' }) + + log.debug(`new SSE subscriber added: ${subscriber.id}`) + log.debug(`nb subscriber: ${this.#subscriberManager.subscribers.size}`) + + return subscriber.id + } + + addListenerFor(id: SubscriberId, { fields, type }: { fields?: '*' | string[]; type: ListenerType }) { + const subscriber = this.#subscriberManager.getSubscriber(id) + const listener = this.#getListener(type) + + listener.addSubscriber(subscriber, fields) + } + + removeListenerFor(id: SubscriberId, type: ListenerType): void { + const subscriber = this.#subscriberManager.getSubscriber(id) + const listener = this.#getListener(type) + + listener.removeSubscriber(subscriber.id) + } +} diff --git a/@xen-orchestra/rest-api/src/events/event.type.mts b/@xen-orchestra/rest-api/src/events/event.type.mts new file mode 100644 index 0000000000..0a9a2d98a8 --- /dev/null +++ b/@xen-orchestra/rest-api/src/events/event.type.mts @@ -0,0 +1,11 @@ +import type { Branded, XapiXoRecord } from '@vates/types' + +export type SubscriberId = Branded<'Subscriber'> + +export type XapiXoListenerType = XapiXoRecord['type'] | 'alarm' + +export type ListenerType = XapiXoListenerType | 'ping' + +export type CollectionEventType = 'add' | 'update' | 'remove' + +export type EventType = CollectionEventType | 'ping' diff --git a/@xen-orchestra/rest-api/src/ioc/ioc.mts b/@xen-orchestra/rest-api/src/ioc/ioc.mts index ff44566558..088067a2ce 100644 --- a/@xen-orchestra/rest-api/src/ioc/ioc.mts +++ b/@xen-orchestra/rest-api/src/ioc/ioc.mts @@ -13,6 +13,7 @@ import { VdiService } from '../vdis/vdi.service.mjs' import { UserService } from '../users/user.service.mjs' import { BackupJobService } from '../backup-jobs/backup-job.service.mjs' import { BackupLogService } from '../backup-logs/backup-log.service.mjs' +import { EventService } from '../events/event.service.mjs' const iocContainer = new Container() @@ -99,6 +100,14 @@ export function setupContainer(xoApp: XoApp) { return new BackupLogService() }) .inSingletonScope() + + iocContainer + .bind(EventService) + .toDynamicValue(ctx => { + const restApi = ctx.container.get(RestApi) + return new EventService(restApi) + }) + .inSingletonScope() } export { iocContainer } diff --git a/@xen-orchestra/rest-api/src/open-api/oa-examples/event.oa-example.mts b/@xen-orchestra/rest-api/src/open-api/oa-examples/event.oa-example.mts new file mode 100644 index 0000000000..dc0c6c166f --- /dev/null +++ b/@xen-orchestra/rest-api/src/open-api/oa-examples/event.oa-example.mts @@ -0,0 +1,3 @@ +export const addSubscription = { + id: 'VM' +} \ No newline at end of file diff --git a/@xen-orchestra/rest-api/src/rest-api/rest-api.type.mts b/@xen-orchestra/rest-api/src/rest-api/rest-api.type.mts index 361407a029..e98a02c3f9 100644 --- a/@xen-orchestra/rest-api/src/rest-api/rest-api.type.mts +++ b/@xen-orchestra/rest-api/src/rest-api/rest-api.type.mts @@ -77,6 +77,15 @@ export type XoApp = { getOptional(path: string): Record | undefined getOptionalDuration(path: string): number | undefined } + + objects: EventEmitter & { + allIndexes: { + type: { + getEventEmitterByType(type: XapiXoRecord['type']): EventEmitter + } + } + } + tasks: EventEmitter & { abort(id: XoTask['id'], reason?: string): Promise clearLogs(): Promise diff --git a/CHANGELOG.unreleased.md b/CHANGELOG.unreleased.md index 40fc689ec5..44144a226f 100644 --- a/CHANGELOG.unreleased.md +++ b/CHANGELOG.unreleased.md @@ -11,6 +11,10 @@ > Users must be able to say: “Nice enhancement, I'm eager to test it” +- [REST API] Expose `GET /rest/v0/events` to open an SSE connection (PR [#9130](https://github.com/vatesfr/xen-orchestra/pull/9130)) +- [REST API] Expose `POST /rest/v0/events/:id/subscriptions` to add a subscription in the SSE connection (PR [#9130](https://github.com/vatesfr/xen-orchestra/pull/9130)) +- [REST API] Expose `DELETE /rest/v0/events/:id/subscriptions` to remove a subscription in the SSE connection (PR [#9130](https://github.com/vatesfr/xen-orchestra/pull/9130)) + - **XO 6:** ### Bug fixes @@ -33,4 +37,7 @@ +- @xen-orchestra/rest-api minor +- xo-collection minor + diff --git a/packages/xo-collection/src/collection.js b/packages/xo-collection/src/collection.js index 2ec2f45f0b..cc97af0508 100644 --- a/packages/xo-collection/src/collection.js +++ b/packages/xo-collection/src/collection.js @@ -110,6 +110,10 @@ export class Collection extends EventEmitter { return this._size } + get allIndexes() { + return this._indexes + } + // ----------------------------------------------------------------- // Manipulation // ----------------------------------------------------------------- diff --git a/packages/xo-collection/src/index.js b/packages/xo-collection/src/index.js index 3769a327f5..a7e82df4ea 100644 --- a/packages/xo-collection/src/index.js +++ b/packages/xo-collection/src/index.js @@ -1,47 +1,53 @@ import iteratee from 'lodash/iteratee' +import { EventEmitter } from 'node:events' import clearObject from './clear-object' -import isEmpty from './is-empty' -import NotImplemented from './not-implemented' import { ACTION_ADD, ACTION_UPDATE, ACTION_REMOVE } from './collection' // =================================================================== export class Index { - constructor(computeHash) { - if (computeHash) { - this.computeHash = iteratee(computeHash) + #indexType + + constructor(indexType) { + if (indexType === undefined) { + throw new Error('indexType must be passed') } - this._itemsByHash = Object.create(null) - this._keysToHash = Object.create(null) + this.#indexType = typeof indexType === 'function' ? indexType : iteratee(indexType) + + this._itemsByType = Object.create(null) + this._keysToType = Object.create(null) + this._eeByType = Object.create(null) // Bound versions of listeners. this._onAdd = this._onAdd.bind(this) this._onUpdate = this._onUpdate.bind(this) this._onRemove = this._onRemove.bind(this) + + this.getIndexType = this.getIndexType.bind(this) } - // This method is used to compute the hash under which an item must - // be saved. - computeHash(value, key) { - throw new NotImplemented('this method must be overridden') - } - - // Remove empty items lists. - sweep() { - const { _itemsByHash: itemsByHash } = this - for (const hash in itemsByHash) { - if (isEmpty(itemsByHash[hash])) { - delete itemsByHash[hash] - } - } + getIndexType(object) { + return this.#indexType(object) } // ----------------------------------------------------------------- get items() { - return this._itemsByHash + return this._itemsByType + } + + getEventEmitterByType(type) { + if (type === undefined) { + throw new Error('type is required') + } + + if (this._eeByType[type] === undefined) { + this._eeByType[type] = new EventEmitter() + } + + return this._eeByType[type] } // ----------------------------------------------------------------- @@ -63,65 +69,69 @@ export class Index { collection.removeListener(ACTION_UPDATE, this._onUpdate) collection.removeListener(ACTION_REMOVE, this._onRemove) - clearObject(this._itemsByHash) - clearObject(this._keysToHash) + clearObject(this._itemsByType) + clearObject(this._keysToType) } // ----------------------------------------------------------------- _onAdd(items) { - const { computeHash, _itemsByHash: itemsByHash, _keysToHash: keysToHash } = this + const { getIndexType, _itemsByType: itemsByType, _keysToType: keysToType } = this for (const key in items) { const value = items[key] - const hash = computeHash(value, key) + const type = getIndexType(value) - if (hash != null) { - ;(itemsByHash[hash] || + if (type != null) { + ;(itemsByType[type] || // FIXME: We do not use objects without prototype for now // because it breaks Angular in xo-web, change it back when // this is fixed. - (itemsByHash[hash] = {}))[key] = value + (itemsByType[type] = {}))[key] = value - keysToHash[key] = hash + keysToType[key] = type + this.getEventEmitterByType(type)?.emit('add', value) } } } _onUpdate(items) { - const { computeHash, _itemsByHash: itemsByHash, _keysToHash: keysToHash } = this + const { getIndexType, _itemsByType: itemsByType, _keysToType: keysToType } = this for (const key in items) { const value = items[key] - const prev = keysToHash[key] - const hash = computeHash(value, key) + const prevType = keysToType[key] + const previousObj = itemsByType[prevType]?.[key] + const type = getIndexType(value) - // Removes item from the previous hash's list if any. - if (prev != null) delete itemsByHash[prev][key] + // Removes item from the previous type's list if any. + if (prevType != null) delete itemsByType[prevType][key] - // Inserts item into the new hash's list if any. - if (hash != null) { - ;(itemsByHash[hash] || + // Inserts item into the new type's list if any. + if (type != null) { + ;(itemsByType[type] || // FIXME: idem: change back to Object.create(null) - (itemsByHash[hash] = {}))[key] = value + (itemsByType[type] = {}))[key] = value - keysToHash[key] = hash + keysToType[key] = type + this.getEventEmitterByType(type)?.emit('update', value, previousObj) } else { - delete keysToHash[key] + delete keysToType[key] } } } _onRemove(items) { - const { _itemsByHash: itemsByHash, _keysToHash: keysToHash } = this + const { _itemsByType: itemsByType, _keysToType: keysToType } = this for (const key in items) { - const prev = keysToHash[key] + const prev = keysToType[key] if (prev != null) { - delete itemsByHash[prev][key] - delete keysToHash[key] + this.getEventEmitterByType(prev)?.emit('remove', undefined, itemsByType[prev][key]) + delete itemsByType[prev][key] + delete keysToType[key] } } } diff --git a/packages/xo-collection/src/index.test.js b/packages/xo-collection/src/index.test.js index d2c8225094..5f47127aea 100644 --- a/packages/xo-collection/src/index.test.js +++ b/packages/xo-collection/src/index.test.js @@ -158,23 +158,4 @@ describe('Index', function () { }) }) }) - - describe('#sweep()', function () { - it('removes empty items lists', function () { - col.remove(item2) - - return waitTicks().then(() => { - byGroup.sweep() - - assert.deepEqual(JSON.parse(JSON.stringify(col.indexes)), { - byGroup: { - foo: { - [item1.id]: item1, - [item3.id]: item3, - }, - }, - }) - }) - }) - }) }) diff --git a/packages/xo-collection/src/unique-index.js b/packages/xo-collection/src/unique-index.js index da7a2b501e..23254e07d7 100644 --- a/packages/xo-collection/src/unique-index.js +++ b/packages/xo-collection/src/unique-index.js @@ -1,36 +1,39 @@ import iteratee from 'lodash/iteratee' import clearObject from './clear-object' -import NotImplemented from './not-implemented' import { ACTION_ADD, ACTION_UPDATE, ACTION_REMOVE } from './collection' // =================================================================== export class UniqueIndex { - constructor(computeHash) { - if (computeHash) { - this.computeHash = iteratee(computeHash) + #indexRef + + constructor(indexRef) { + if (indexRef === undefined) { + throw new Error('indexRef must be passed') } - this._itemByHash = Object.create(null) - this._keysToHash = Object.create(null) + this.#indexRef = typeof indexRef === 'function' ? indexRef : iteratee(indexRef) + + this._itemByRef = Object.create(null) + this._keysToRef = Object.create(null) // Bound versions of listeners. this._onAdd = this._onAdd.bind(this) this._onUpdate = this._onUpdate.bind(this) this._onRemove = this._onRemove.bind(this) + + this.getIndexRef = this.getIndexRef.bind(this) } - // This method is used to compute the hash under which an item must - // be saved. - computeHash(value, key) { - throw new NotImplemented('this method must be overridden') + getIndexRef(object) { + return this.#indexRef(object) } // ----------------------------------------------------------------- get items() { - return this._itemByHash + return this._itemByRef } // ----------------------------------------------------------------- @@ -52,57 +55,57 @@ export class UniqueIndex { collection.removeListener(ACTION_UPDATE, this._onUpdate) collection.removeListener(ACTION_REMOVE, this._onRemove) - clearObject(this._itemByHash) - clearObject(this._keysToHash) + clearObject(this._itemByRef) + clearObject(this._keysToRef) } // ----------------------------------------------------------------- _onAdd(items) { - const { computeHash, _itemByHash: itemByHash, _keysToHash: keysToHash } = this + const { getIndexRef, _itemByRef: itemByRef, _keysToRef: keysToRef } = this for (const key in items) { const value = items[key] - const hash = computeHash(value, key) + const ref = getIndexRef(value) - if (hash != null) { - itemByHash[hash] = value - keysToHash[key] = hash + if (ref != null) { + itemByRef[ref] = value + keysToRef[key] = ref } } } _onUpdate(items) { - const { computeHash, _itemByHash: itemByHash, _keysToHash: keysToHash } = this + const { getIndexRef, _itemByRef: itemByRef, _keysToRef: keysToRef } = this for (const key in items) { const value = items[key] - const prev = keysToHash[key] - const hash = computeHash(value, key) + const prevRef = keysToRef[key] + const ref = getIndexRef(value) - // Removes item from the previous hash's list if any. - if (prev != null) delete itemByHash[prev] + // Removes item from the previous ref's list if any. + if (prevRef != null) delete itemByRef[prevRef] - // Inserts item into the new hash's list if any. - if (hash != null) { - itemByHash[hash] = value - keysToHash[key] = hash + // Inserts item into the new ref's list if any. + if (ref != null) { + itemByRef[ref] = value + keysToRef[key] = ref } else { - delete keysToHash[key] + delete keysToRef[key] } } } _onRemove(items) { - const { _itemByHash: itemByHash, _keysToHash: keysToHash } = this + const { _itemByRef: itemByRef, _keysToRef: keysToRef } = this for (const key in items) { - const prev = keysToHash[key] - if (prev != null) { - delete itemByHash[prev] - delete keysToHash[key] + const prevRef = keysToRef[key] + if (prevRef != null) { + delete itemByRef[prevRef] + delete keysToRef[key] } } }