From 7b7b79ee4c82dd714a7e70e43b0f6c24a2153efa Mon Sep 17 00:00:00 2001 From: alexMolex Date: Sun, 15 Sep 2024 19:07:03 +0300 Subject: [PATCH 1/3] feat: adding sync remote and local state --- emitter-sync-pattern.ts | 141 ++++++++++++++++++++++++++++++++++++---- event-repository.ts | 2 +- 2 files changed, 128 insertions(+), 15 deletions(-) diff --git a/emitter-sync-pattern.ts b/emitter-sync-pattern.ts index ff0dafa..b32490c 100644 --- a/emitter-sync-pattern.ts +++ b/emitter-sync-pattern.ts @@ -1,7 +1,7 @@ /* Check the comments first */ import { EventEmitter } from "./emitter"; -import { EventDelayedRepository } from "./event-repository"; +import { EventDelayedRepository, EventRepositoryError } from "./event-repository"; import { EventStatistics } from "./event-statistics"; import { ResultsTester } from "./results-tester"; import { triggerRandomly } from "./utils"; @@ -59,29 +59,142 @@ function init() { */ class EventHandler extends EventStatistics { - // Feel free to edit this class - repository: EventRepository; constructor(emitter: EventEmitter, repository: EventRepository) { super(); this.repository = repository; - emitter.subscribe(EventName.EventA, () => - this.repository.saveEventData(EventName.EventA, 1) - ); + EVENT_NAMES.forEach((eventName) => { + emitter.subscribe(eventName, () => { + const currentLocalValue = this.getStats(eventName) + 1; + + this.setStats(eventName, currentLocalValue) + + this.repository.saveEventData(eventName, currentLocalValue) + }); + }) } } +class Throttler { + private func: (...args: unknown[]) => void; + private limit: number; + private lastFunc: ReturnType | null = null; + private lastRan: number | null = null; + + constructor(func: (...args: unknown[]) => void, limit: number) { + this.func = func; + this.limit = limit; + } + + public throttle(...args: unknown[]): void { + const context = this; + + if (this.lastRan === null) { + this.func.apply(context, args); + this.lastRan = Date.now(); + } else { + if (this.lastFunc !== null) { + clearTimeout(this.lastFunc); + } + + this.lastFunc = setTimeout(() => { + if ((Date.now() - (context.lastRan as number)) >= context.limit) { + context.func.apply(context, args); + context.lastRan = Date.now(); + } + }, this.limit - (Date.now() - (this.lastRan as number))); + } + } +} + +type EventThread = { + previousSyncedValue: number; + isUpdating: boolean; +}; + +const thredNotFoundError = new Error('Thread not found'); +const REQUEST_REMOTE_UPDATE_THROTTLE = 150; + class EventRepository extends EventDelayedRepository { - // Feel free to edit this class - - async saveEventData(eventName: EventName, _: number) { - try { - await this.updateEventStatsBy(eventName, 1); - } catch (e) { - // const _error = e as EventRepositoryError; - // console.warn(error); + eventUpdatingThrteads = new Map(); + requestRemoteUpdateTrhrottled: Throttler + + constructor() { + super(); + + this.requestRemoteUpdateTrhrottled = new Throttler(this.requestRemoteUpdate, REQUEST_REMOTE_UPDATE_THROTTLE); + } + + public saveEventData = async (eventName: EventName, currentLocalValue: number) => { + this.requestRemoteUpdateTrhrottled.throttle(eventName, currentLocalValue); + } + + private lockThread = (eventName: EventName) => { + const thread = this.eventUpdatingThrteads.get(eventName) + + if(!thread) { + throw thredNotFoundError + } + + this.eventUpdatingThrteads.set(eventName, { + ...thread, + isUpdating: true + }) + } + + private unlockThread = (eventName: EventName) => { + const thread = this.eventUpdatingThrteads.get(eventName) + + if(!thread) { + throw thredNotFoundError + } + + this.eventUpdatingThrteads.set(eventName, { + ...thread, + isUpdating: false + }) + } + + private unlockAndUpdateThread = (eventName: EventName, currentSyncedValue: number) => { + this.eventUpdatingThrteads.set(eventName, { + previousSyncedValue: currentSyncedValue, + isUpdating: false + }) + } + + private requestRemoteUpdate = async (eventName: EventName, currentLocalValue: number) => { + if(!this.eventUpdatingThrteads.has(eventName)) { + this.eventUpdatingThrteads.set(eventName, { + previousSyncedValue: 0, + isUpdating: false + }) + } + + const { isUpdating, previousSyncedValue } = this.eventUpdatingThrteads.get(eventName) as EventThread + + if (isUpdating) { + return; + } + + try { + this.lockThread(eventName); + + await this.updateEventStatsBy(eventName, currentLocalValue - previousSyncedValue) + + + this.unlockAndUpdateThread(eventName, currentLocalValue) + } catch (error) { + if(error.message === thredNotFoundError.message){ + throw error + } + + if (error === EventRepositoryError.RESPONSE_FAIL) { + this.unlockAndUpdateThread(eventName, currentLocalValue) + } else { + this.unlockThread(eventName) + } } } } diff --git a/event-repository.ts b/event-repository.ts index 64b7a6f..7d61dfa 100644 --- a/event-repository.ts +++ b/event-repository.ts @@ -14,7 +14,7 @@ import { awaitTimeout, randomTo } from "./utils"; const EVENT_SAVE_DELAY_MS = 3 * 100; -enum EventRepositoryError { +export enum EventRepositoryError { TOO_MANY = "Too many requests", RESPONSE_FAIL = "Response delivery fail", REQUEST_FAIL = "Request fail", From 5651034abaaaec271eb16c19f7ac5278ec6a62a5 Mon Sep 17 00:00:00 2001 From: alexMolex Date: Mon, 16 Sep 2024 00:25:53 +0300 Subject: [PATCH 2/3] refactor: chore --- emitter-sync-pattern.ts | 28 ++++++++++++++-------------- 1 file changed, 14 insertions(+), 14 deletions(-) diff --git a/emitter-sync-pattern.ts b/emitter-sync-pattern.ts index b32490c..950a443 100644 --- a/emitter-sync-pattern.ts +++ b/emitter-sync-pattern.ts @@ -109,16 +109,16 @@ class Throttler { } } -type EventThread = { +type TUpdatingThread = { previousSyncedValue: number; isUpdating: boolean; }; -const thredNotFoundError = new Error('Thread not found'); +const threadNotFoundError = new Error('Thread not found'); const REQUEST_REMOTE_UPDATE_THROTTLE = 150; class EventRepository extends EventDelayedRepository { - eventUpdatingThrteads = new Map(); + eventUpdatingThreads = new Map(); requestRemoteUpdateTrhrottled: Throttler constructor() { @@ -132,47 +132,47 @@ class EventRepository extends EventDelayedRepository { } private lockThread = (eventName: EventName) => { - const thread = this.eventUpdatingThrteads.get(eventName) + const thread = this.eventUpdatingThreads.get(eventName) if(!thread) { - throw thredNotFoundError + throw threadNotFoundError } - this.eventUpdatingThrteads.set(eventName, { + this.eventUpdatingThreads.set(eventName, { ...thread, isUpdating: true }) } private unlockThread = (eventName: EventName) => { - const thread = this.eventUpdatingThrteads.get(eventName) + const thread = this.eventUpdatingThreads.get(eventName) if(!thread) { - throw thredNotFoundError + throw threadNotFoundError } - this.eventUpdatingThrteads.set(eventName, { + this.eventUpdatingThreads.set(eventName, { ...thread, isUpdating: false }) } private unlockAndUpdateThread = (eventName: EventName, currentSyncedValue: number) => { - this.eventUpdatingThrteads.set(eventName, { + this.eventUpdatingThreads.set(eventName, { previousSyncedValue: currentSyncedValue, isUpdating: false }) } private requestRemoteUpdate = async (eventName: EventName, currentLocalValue: number) => { - if(!this.eventUpdatingThrteads.has(eventName)) { - this.eventUpdatingThrteads.set(eventName, { + if(!this.eventUpdatingThreads.has(eventName)) { + this.eventUpdatingThreads.set(eventName, { previousSyncedValue: 0, isUpdating: false }) } - const { isUpdating, previousSyncedValue } = this.eventUpdatingThrteads.get(eventName) as EventThread + const { isUpdating, previousSyncedValue } = this.eventUpdatingThreads.get(eventName) as TUpdatingThread if (isUpdating) { return; @@ -186,7 +186,7 @@ class EventRepository extends EventDelayedRepository { this.unlockAndUpdateThread(eventName, currentLocalValue) } catch (error) { - if(error.message === thredNotFoundError.message){ + if(error.message === threadNotFoundError.message){ throw error } From b3ca8f625cdce6dc31ed79f548058a9cfdfaae37 Mon Sep 17 00:00:00 2001 From: alexMolex Date: Mon, 16 Sep 2024 00:41:23 +0300 Subject: [PATCH 3/3] refactor: adding comments --- emitter-sync-pattern.ts | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/emitter-sync-pattern.ts b/emitter-sync-pattern.ts index 950a443..228578a 100644 --- a/emitter-sync-pattern.ts +++ b/emitter-sync-pattern.ts @@ -118,6 +118,10 @@ const threadNotFoundError = new Error('Thread not found'); const REQUEST_REMOTE_UPDATE_THROTTLE = 150; class EventRepository extends EventDelayedRepository { + /* + I split the synchronization of events into threads so that updating one does not block the other. + Blocking the threads is needed to avoid race conditions. + */ eventUpdatingThreads = new Map(); requestRemoteUpdateTrhrottled: Throttler @@ -183,7 +187,6 @@ class EventRepository extends EventDelayedRepository { await this.updateEventStatsBy(eventName, currentLocalValue - previousSyncedValue) - this.unlockAndUpdateThread(eventName, currentLocalValue) } catch (error) { if(error.message === threadNotFoundError.message){