I am using events to listen to a data change stream https://docs.mongodb.com/manual/changeStreams
Each event that comes in is handled via an aync handleChange function. The first part of the function i do the normal processing of the data and the second part of the function i want to store a resumeToken.
This resumeToken is important because it allows me to restart my listener incase there is a crash.
However i have a problem - when i am processing each event with an async handler, the resume token in the end that gets stored is the last event that was processed and not emitted. This means the resumeToken may lag behind and when i restart the process, it may resume at point which it has already previously processed.
The code below is the smallest isolated example of what is happening.
const EventEmitter = require('events');
const myEmitter = new EventEmitter();
let resumeToken; // Stored in external db;
// simulate an update to resumeToken where time taken could be different
const updateResumeToken = async (token) => {
const randomTime = Math.floor(Math.random() * 10) * 1000;
await new Promise(resolve => setTimeout(() => {
resumeToken = token;
resolve();
}, randomTime));
}
const handleChange = async (data) => {
console.log("current", data, resumeToken)
await updateResumeToken(data.value);
console.log("new", data, resumeToken);
}
myEmitter.on('change', handleChange);
myEmitter.emit('change', {"token": "first", value: 1} );
myEmitter.emit('change', {"token": "second", value: 2});
myEmitter.emit('change', {"token": "third", value: 3});