Improving observable persistent queue mini-library

Question

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:

1. 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

Answer 1

I'm not very good at js & I can't really suggest ways to clean things up, but you've got this odd mix of OOP & FP, callbacks & promises, and arrow notation & function(). This code style is overall inconsistent. A good first step would be to try to achieve some consistency in style.

Now, it may be that the inconsistency stems from using different libraries that each utilize a different style & philosophy, but it's important that your code is written consistently.

So, first, you need to decide which style you prefer. OOP or FP? Callbacks or Promises? Ok. Good. Now create abstractions that adapt libs in the other styles to your chosen style. Remember, it's your code and you must be the one to drive the design, not the libraries you're using.