albe / node-event-storage / 27159296142

import ReadableStorage from './ReadableStorage.js';

import ReadablePartition from '../Partition/ReadablePartition.js';

import Watcher from '../Watcher.js';

/**

 * An append-only storage with highly performant positional range scans.

 * It's highly optimized for an event-store and hence does not support compaction or data-rewrite, nor any querying

 */

class ReadOnlyStorage extends ReadableStorage {

    /**

     * @inheritdoc

     */

    constructor(storageName = 'storage', config = {}) {

        super(storageName, config);

        this.storageFilesFilter = this.storageFilesFilter.bind(this);

        this.onStorageFileChanged = this.onStorageFileChanged.bind(this);

    }

    /**

     * Returns true if the given filename belongs to this storage.

     * @param {string} filename

     * @returns {boolean}

     */

    storageFilesFilter(filename) {

        return !filename.endsWith('.branch') && filename.substring(0, this.storageFile.length) === this.storageFile;

    }

    /**

     * Open the storage and indexes and create read and write buffers eagerly.

     * Will emit an 'opened' event if finished.

     *

     * @api

     * @param {function(): void} [callback] Called after indexes open, before `'opened'` is emitted.

     *   Can be used as a synchronous alternative to listening to the `'opened'` event.

     * @returns {boolean}

     */

    open(callback) {

        if (!this.watcher) {

            this.watcher = new Watcher([this.dataDirectory, this.indexDirectory], this.storageFilesFilter);

            this.watcher.on('rename', this.onStorageFileChanged);

        }

        return super.open(callback);

    }

    /**

     * @private

     * @param {string} filename

     */

    onStorageFileChanged(filename) {

        if (filename.endsWith('.index')) {

            const indexName = filename.substring(this.storageFile.length + 1, filename.length - 6);

            // New indexes are not automatically opened in the reader

            this.emit('index-created', indexName);

            return;

        }

        const partitionId = ReadablePartition.idFor(filename);

        if (!this.partitions.has(partitionId)) {

            const partition = this.createPartition(filename, this.partitionConfig);

            this.partitions.add(partition.id, partition);

            this.emit('partition-created', partition.id);

        }

    }

    /**

     * Close the storage and frees up all resources.

     * Will emit a 'closed' event when finished.

     *

     * @api

     * @returns void

     */

    close() {

        if (this.watcher) {

            this.watcher.close();

            this.watcher = null;

        }

        super.close();

    }

    /**

     * @protected

     * @param {string} name

     * @param {object} [options]

     * @returns {{ index: ReadableIndex, matcher?: object|function }}

     */

    createIndex(name, options = {}) {

        const { index } = super.createIndex(name, options);

        const indexShortName = name.replace(this.storageFile + '.', '').replace('.index', '');

        index.on('append', (prevLength, newLength) => {

            const entries = index.range(prevLength + 1, newLength);

            /* c8 ignore next 3 */

            if (entries === false) {

                return;

            }

            for (let entry of entries) {

                const document = this.readFrom(entry.partition, entry.position, entry.size);

                if (index === this.index) {

                    this.emit('wrote', document, entry, entry.position);

                } else {

                    this.emit('index-add', indexShortName, entry.number, document);

                }

            }

        });

        index.on('truncate', (prevLength, newLength) => {

            if (index === this.index) {

                this.emit('truncate', prevLength, newLength);

            }

        });

        return { index };

    }

}

export default ReadOnlyStorage;