javascript - 如何检测一连串相关事件已完成，然后发出一个事件

Question

我的第一个问题!我真的很想问对它，所以如果我可以问得更好，请帮助我改进它。

这是我发现的唯一一个似乎有点相关的问题，但我无法弄清楚如何将它与我正在尝试做的事情联系起来(他们的问题是特定于 JQuery 的；我的是特定于 Node.JS [尽管我找到了 EventEmitter 的浏览器版本并且也能够在浏览器中进行测试]): Run function once per event burst with jQuery

---

问题

我有一个进程，我知道它会在一段时间内发出一连串的事件。

为了模拟这个过程，我编写了这段代码:

/*******************************************************/
/*         This part taken directly from               */
/*       https://nodejs.org/api/events.html            */
/*          (with addition of "burstID")               */
/* */                                               /* */
/* */ const EventEmitter = require('events');       /* */
/* */                                               /* */
/* */ class MyEmitter extends EventEmitter {}       /* */
/* */                                               /* */
/* */ const myEmitter = new MyEmitter();            /* */
/* */ myEmitter.on('event', (burstID) => {          /* */
/* */   console.log('an event occurred!', burstID); /* */
/* */ });                                           /* */
/* */                                               /* */
/*******************************************************/

const millisecondsToSustainBurst = 3000 ;
const millisecondsBetweenPossibleEventEmissions = 200 ;
const millisecondsUntilStartNextBurst = 5000 ;
const millisecondsUntilNoMoreBursts = 23000 ;

const now = new Date() ;
console.log('Time now: ' + now + '; should run until about ' + new Date(now.getTime() + millisecondsUntilNoMoreBursts)) ;

const doRandomEmitBurst = (startTimestamp, millisecondsToSustainBurst, burstID) => {
    if (Math.random() > 0.5) myEmitter.emit('event', burstID) ;
    if (
        !((new Date()) - startTimestamp > millisecondsToSustainBurst)
    ) setTimeout(() => doRandomEmitBurst(startTimestamp, millisecondsToSustainBurst, burstID), millisecondsBetweenPossibleEventEmissions) ;
}

const doRegularRandomBursts = (startTimestamp, millisecondsUntilStartNextBurst, millisecondsUntilNoMoreBursts, callback) => {
    if (
        !((new Date()) - startTimestamp > millisecondsUntilNoMoreBursts)
    ) {
        const now = new Date() ;
        console.log('Time now: ' + now + '; starting random-event burst which will run for ' + (millisecondsToSustainBurst/1000) + ' seconds.  ') ;
        setTimeout(() => doRegularRandomBursts(startTimestamp, millisecondsUntilStartNextBurst, millisecondsUntilNoMoreBursts, callback), millisecondsUntilStartNextBurst) ;
        doRandomEmitBurst(new Date(), millisecondsToSustainBurst, 'burstThatStartedAt' + now.getHours() + 'h' + now.getMinutes() + 'm' + now.getSeconds() + 's') ;
    } else callback() ;
}

doRegularRandomBursts(new Date(), millisecondsUntilStartNextBurst, millisecondsUntilNoMoreBursts, () => console.log('Done at ' + (new Date()))) ;

const myBurstDetectedEmitter = new MyEmitter() ;
// NOW, what do I do HERE to say:
// I've seen a number of events occur in a 5-second period
// Now they've stopped
// Therefore I'm going to emit a different kind of event

---

现在，假设我想听听这些爆发的发生。

在我采取进一步行动之前，我想确保爆发已经结束。

我该怎么做？

---

到目前为止我尝试了什么

首先，我可以像这样创建一个全局“var”(糟糕——我想避免使用可变变量):

var timeLastUpdated = {} ;

……然后……

function keepCheckingTimeLastUpdated(keyForUpdateCheck, callback) {
    const timestampToCheckInOneSecond = (typeof timeLastUpdated[keyForUpdateCheck] !== 'undefined' ? timeLastUpdated[keyForUpdateCheck] : (new Date())) ;
    setTimeout(() => {
        console.log(
            'checking if modifications to "' +
            keyForUpdateCheck +
            '" have occurred since ' +
            timestampToCheckInOneSecond
        ) ;
        if (timeLastUpdated[keyForUpdateCheck] === timestampToCheckInOneSecond) {
            delete timeLastUpdated[keyForUpdateCheck] ;
            callback() ;
        }
        else keepCheckingTimeLastUpdated(keyForUpdateCheck, callback) ;
    }, 5000) ;
}

const makeNotificationHandler = () => (keyForUpdateCheck) => {
    const timeNow = new Date() ;
    if (typeof timeLastUpdated[keyForUpdateCheck] === 'undefined') keepCheckingTimeLastUpdated(keyForUpdateCheck, () => console.log(keyForUpdateCheck + ' changed')) ;
    timeLastUpdated[keyForUpdateCheck] = timeNow ;
} ;

myEmitter.on('event', makeNotificationHandler()) ;

只是这似乎是一种反模式(我希望我使用的是正确的术语)。我的直觉告诉我，拥有一个全局对象在这里是错误的方法，并且有一个更像函数式编程的解决方案。

---

仅限感兴趣的人:

(出于回答问题的目的，请随意忽略)

增加了复杂性:在我的示例代码中，“burstID”永远不会相同，但在真实世界的示例中，它可能是相同的。我想等到自上次“burstID”出现以来经过了一定的时间，以便弄清楚变化的爆发是否真的完成了。

对于上下文，在实际应用程序中，我正在使用 node-postgres 在 PostGres 数据库上设置“LISTEN” . “burstID”在一个表中是主键，在其他多个表中也被用作外键。我正在监听所有使用共享 key 的表，并且我收到的消息包含此 key 。

Answer 1

在上述评论的帮助下回答我自己的问题。

一声大大的“谢谢!”至 Scott Sauyet感谢他的帮助和鼓励。

---

我决定创建一个我称之为“累加器”的东西:一个自动累加器。消息被路由到这些 Accumulomators，我认为他们是仓库里拿着箱子和秒表的人。

一旦 Accumulomator 被实例化，它就会开始查看秒表。每次秒表走到尽头时，累加器都会查看是否有与上次相同的消息堆。如果和上次一样，Accumulomator 会打包它的消息箱，将它们交给仓库，然后在阳光明媚的地方退休。

完整的更新代码如下。我欢迎任何修改以改进代码。

---

// To test in a browser, use:
// https://github.com/Olical/EventEmitter
// …in place of const EventEmitter = require('events');

/*******************************************************/
/*         This part taken directly from               */
/*       https://nodejs.org/api/events.html            */
/*          (with addition of "burstID")               */
/* */                                               /* */
/* */ const EventEmitter = require('events');       /* */
/* */                                               /* */
/* */ class MyEmitter extends EventEmitter {}       /* */
/* */                                               /* */
/* */ const myEmitter = new MyEmitter();            /* */
/* */ myEmitter.on('event', (burstID) => {          /* */
/*      console.log('an event occurred!', burstID);    */
/* */ });                                           /* */
/* */                                               /* */
/*******************************************************/

const millisecondsToSustainBurst = 3000 ;
const millisecondsBetweenPossibleEventEmissions = 200 ;
const millisecondsUntilStartNextBurst = 5000 ;
const millisecondsUntilNoMoreBursts = 23000 ;

const now = new Date() ;
console.log('Bursts starting.  Time now: ' + now + '; should run until about ' + new Date(now.getTime() + millisecondsUntilNoMoreBursts)) ;

const doRandomEmitBurst = (startTimestamp, millisecondsToSustainBurst, burstID) => {
    if (Math.random() > 0.5) myEmitter.emit('event', burstID) ;
    if (
        !((new Date()) - startTimestamp > millisecondsToSustainBurst)
    ) setTimeout(() => doRandomEmitBurst(startTimestamp, millisecondsToSustainBurst, burstID), millisecondsBetweenPossibleEventEmissions) ;
}

const doRegularRandomBursts = (startTimestamp, millisecondsUntilStartNextBurst, millisecondsUntilNoMoreBursts, callback) => {
    if (
        !((new Date()) - startTimestamp > millisecondsUntilNoMoreBursts)
    ) {
        const now = new Date() ;
        console.log('Time now: ' + now + '; starting random-event burst which will run for ' + (millisecondsToSustainBurst/1000) + ' seconds.  ') ;
        setTimeout(() => doRegularRandomBursts(startTimestamp, millisecondsUntilStartNextBurst, millisecondsUntilNoMoreBursts, callback), millisecondsUntilStartNextBurst) ;
        doRandomEmitBurst(new Date(), millisecondsToSustainBurst, 'burstThatStartedAt' + now.getHours() + 'h' + now.getMinutes() + 'm' + now.getSeconds() + 's') ;
    } else callback() ;
}

doRegularRandomBursts(new Date(), millisecondsUntilStartNextBurst, millisecondsUntilNoMoreBursts, () => console.log('Done at ' + (new Date()))) ;


const makeAccumulomator = (config) => {
    if (typeof config !== 'object') throw new Error('Must specify configuration object.') ;
    if (typeof config.callback !== 'function') throw new Error('Must specify callback function for when the end of new messages is reached.') ;
    if (typeof config.millisecondsBetweenChecks !== 'number') throw new Error('Must specify milliseconds between checks.') ;
    if (Number.isInteger(config.millisecondsBetweenChecks) === false) throw new Error('Must specify milliseconds between checks as an integer.') ;
    if (typeof config.onStop !== 'function' && typeof config.onStop !== 'undefined') throw new Error('If defined at all, onStop must be a function.') ;

    const accumulomator = {} ;


    var accumulatedMessages = [] ;
    var stop = false ;

    const keepCheckingTimeLastUpdated = (callback) => {
        const timestampToCheckInOneSecond = (accumulatedMessages.length > 0 ? accumulatedMessages[accumulatedMessages.length - 1].timestamp : (new Date())) ;
        setTimeout(() => {
            if (stop) { if (typeof config.onStop === 'function') config.onStop() ; }
            else if (accumulatedMessages.length < 1) keepCheckingTimeLastUpdated(callback) ;
            else if (accumulatedMessages[accumulatedMessages.length - 1].timestamp === timestampToCheckInOneSecond) { stop = true ; callback() ; }
            else keepCheckingTimeLastUpdated(callback) ;
        }, config.millisecondsBetweenChecks) ;
    } ;
    keepCheckingTimeLastUpdated(config.callback) ;

    accumulomator.receiveMessage = (message) => accumulatedMessages.push({ message: message, timestamp: (new Date())}) ;

    accumulomator.stopOnNextCheck = () => {
        if (stop === true) throw new Error('Accumulomator is already stopped.') ;
        else stop = true ;
    }

    accumulomator.isActive = () => stop === false ;

    accumulomator.getAccumulatedMessages = () => accumulatedMessages ;


    return accumulomator ;

}

const makeAccumulomatorWarehouse = (config) => {
    if (typeof config !== 'object') throw new Error('Must specify configuration object.') ;
    if (typeof config.callback !== 'function') throw new Error('Must specify callback function.') ;
    if (typeof config.millisecondsBetweenChecks !== 'number') throw new Error('Must specify milliseconds between checks.') ;
    if (Number.isInteger(config.millisecondsBetweenChecks) === false) throw new Error('Must specify milliseconds between checks as an integer.') ;
    if (typeof config.messageRouter !== 'function') throw new Error('Must specify message router function.') ;
    if (typeof config.sendCallbackAccumulatedMessages !== 'undefined') if (typeof config.sendCallbackAccumulatedMessages !== 'boolean') throw new Error('Must specify whether or not to send callback accumulated messages as a boolean (if unspecified, accumulated messages will not be included).') ;
    if (typeof config.onAccumulomatorStop !== 'function' && typeof config.onAccumulomatorStop !== 'undefined') throw new Error('If defined at all, onAccumulomatorStop must be a function.') ;

    var sendCallbackAccumulatedMessages = false ;
    if (typeof config.sendCallbackAccumulatedMessages !== 'undefined') sendCallbackAccumulatedMessages = config.sendCallbackAccumulatedMessages ;

    const accumulomatorWarehouse = {} ;


    const accumulomators = {} ;

    var warehouseIsShuttingDown = false ;

    accumulomatorWarehouse.receiveMessage = (message) => {
        accumulomatorName = config.messageRouter(message) ;
        if (typeof accumulomatorName !== 'string') throw new Error('The value returned from messageRouter must be a string with a unique identifier.') ;
        if (typeof accumulomators[accumulomatorName] === 'object') if (accumulomators[accumulomatorName].isActive() === false) delete accumulomators[accumulomatorName] ;
        if (typeof accumulomators[accumulomatorName] === 'undefined') {
            if (warehouseIsShuttingDown === false) accumulomators[accumulomatorName] = makeAccumulomator({
                callback: () => config.callback(
                    (() => {
                        objectToReturn = {} ;
                        objectToReturn.key = accumulomatorName ;
                        if (sendCallbackAccumulatedMessages === true) objectToReturn.messages = accumulomators[accumulomatorName].getAccumulatedMessages() ;
                        return objectToReturn ;
                    })()
                ) ,
                millisecondsBetweenChecks: config.millisecondsBetweenChecks ,
                onStop: () => config.onAccumulomatorStop(accumulomatorName)
            }) ;
        }
        if (typeof accumulomators[accumulomatorName] === 'object') accumulomators[accumulomatorName].receiveMessage(message) ;
    }

    periodicallyRetireAccumulomators = () => {
        Object.keys(accumulomators).forEach((accumulomator) => {
            if (accumulomators[accumulomator].isActive() === false) delete accumulomators[accumulomator] ;
        }) ;
        if (!(warehouseIsShuttingDown === true && Object.keys(accumulomators).length === 0)) setTimeout(periodicallyRetireAccumulomators, 10000) ;
    } ;
    periodicallyRetireAccumulomators() ;

    accumulomatorWarehouse.shutDownWarehouse = () => {
        Object.keys(accumulomators).forEach((accumulomator) => {
            if (accumulomators[accumulomator].isActive() === true) accumulomators[accumulomator].stopOnNextCheck() ;
        }) ;
        warehouseIsShuttingDown = true ;
    }


    return accumulomatorWarehouse ;

}

myAccumulomatorWarehouse = makeAccumulomatorWarehouse({
    callback: (accumulomatorWarehousePackage) => console.log('Done with accumulomator.', accumulomatorWarehousePackage.key, accumulomatorWarehousePackage.messages) ,
    millisecondsBetweenChecks: 2000 ,
    messageRouter: (message) => message ,
    sendCallbackAccumulatedMessages: true ,
    onAccumulomatorStop: (accumulomatorName) => console.log('Accumulomator for ' + accumulomatorName + ' manually stopped')
}) ;

myEmitter.on('event', myAccumulomatorWarehouse.receiveMessage) ;

setTimeout(myAccumulomatorWarehouse.shutDownWarehouse, 10000) ;