the enqueue / enq method of an OPQ queue

enqueue.js

// enqueues 1 or more items, accepting a string or an array of strings 
enq(lines: string | Array<string>, opts: IEnqueueOpts): Observable<any> {

        opts = opts || {};

        if (opts.controlled) {
            return this._enqControlled(lines, opts);
        }

         //default priority is the lowest priority, which is 1
        const priority = opts.priority || 1;
        const isShare = opts.isShare === true;

        lines = _.flattenDeep([lines]).map(function (l) {
            // remove unicode characters because they will mess up
            // our replace-line algorithm
            return String(l).replace(/[^\x00-\x7F]/g, '');
        });

        let $add = this.init()
            .flatMap(() => {
                return acquireLock(this, '<enqueue>')
                    .flatMap(obj => {
                        return acquireLockRetry(this, obj)
                    });
            })
            .flatMap(obj => {
                return appendFile(this, lines, priority)
                    .map(() => obj);
            })
            .flatMap(obj => releaseLock(this, obj.id))
            .catch(err => {
                // catch errors here and release the lock if it hasn't been released already
            })
            .take(1);

        if (isShare) {
            // share() should be equivalent to publish().refCount()
            // using isShare will allow this method to run/execute even if the returned
            // observable has no subscribers
            $add = $add.share();
            $add.subscribe();
        }

        return $add;

    };