I am creating a small backend library that uses a persistent queue, where the queue is just lines in a text file. The purpose is simply to allow multiple processes to read from the text file, so that work can be coordinated across processes. The constructor for the queue is asynchronous, because we use async i/o to ensure the text file exists on the filesystem as well as to initialize a locking mechanism via websockets, to control access to our "text file database". I have looked at some async constructor design patterns, and one I thought was reasonable involved re-calling the initialization function in each method - if initialization is finished then it just becomes basically quick and synchronous, so not too much of a code overhead or performance penalty.
In general, I believe it makes sense to use observables for this library, because a user of the lib might want to listen to the 'add/enqueue' event - whenever an item is added to the queue, and this represents an ongoing stream of events, perfect for observables.
I tried to distill the code down, here is what I have. I have two problems/concerns:
- I want to create an observable that fires events when items get added to the queue. The queue in memory is an array which belongs to the queue instance. What I do is tail the text file, and then when the tail callback fires, I call onNext on the Rx.Subject.
// constructor
function Queue(obj) {
const fp = this.filepath = obj.filepath;
this.lock = uuidV4();
this.dateCreated = new Date();
this.obsEnqueue = new Rx.Subject();
this.obsDequeue = new Rx.Subject();
this.isReady = false;
var callable = true;
const ee = this.ee = new EE();
// init both creates the queue file if it does not exist, and finds/initializes the live-mutex
// async initialization
// every instance method will call this before acting
this.init = function (isPublish) {
if (this.isReady) {
return makeGenericObservable(null, {isPublish: isPublish});
}
if (!callable) {
return makeEEObservable(ee, {isPublish: isPublish});
}
callable = false;
var lockAcquired = false;
const promise = lmUtils.conditionallyLaunchSocketServer({port: 7029});
return Rx.Observable.fromPromise(promise)
.flatMap(() => {
this.client = new Client({port: 7029});
return acquireLock(this);
})
.flatMap(() => {
lockAcquired = true;
return ifFileExistAndIsAllWhiteSpaceThenTruncate(this)
})
.flatMap(() => {
return genericAppendFile(this, '\n')
})
.flatMap(() => {
return releaseLock(this)
}).map(() => {
ee.emit('ready');
this.isReady = true;
//start tailing, only after we know that the file exists, etc.
tail(fp).on('data', data => {
data = String(data).split('\n')
.filter(ln => String(ln).trim().length > 0)
.map(ln => String(ln).trim());
data.map(function (d) {
try {
return JSON.parse(d);
}
catch (err) {
return '';
}
}).filter(function (d) {
return String(d).trim().length > 0;
}).forEach(d => {
this.obsEnqueue.onNext(d);
});
});
})
.catch(e => {
console.error(e.stack || e);
if (lockAcquired) {
return releaseLock(this);
}
else {
return makeGenericObservable();
}
});
};
}
if anyone could help me out and let me know how to problems 1 and 2 could be mitigated that would be super nice. Also, I probably overuse flatMap as I don't know many of the other instances methods, I don't event really understand what flatMap does anyway. Please don't hesitate to offer any advice with the code, will take zero offense! thanks
