552719872

Build # 552719872

Build Type

github

Committed by GitHub

Commit Message

Merge f28c444be into d9c28c068

Pull Request Pull Request #145: Separate documents by a unique sequence

Run Details

533 of 569 branches covered (93.67%)

Branch coverage included in aggregate %.

26 of 27 new or added lines in 4 files covered. (96.3%)

8 existing lines in 2 files now uncovered.

1262 of 1277 relevant lines covered (98.83%)

1007.81 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

96.23

/src/Partition/WritablePartition.js

const fs = require('fs');
const mkdirpSync = require('mkdirp').sync;
const ReadablePartition = require('./ReadablePartition');
const { assert, buildMetadataHeader } = require('../util');
const Clock = require('../Clock');

const DEFAULT_WRITE_BUFFER_SIZE = 16 * 1024;
const DOCUMENT_HEADER_SIZE = 16;
const DOCUMENT_ALIGNMENT = 4;
const DOCUMENT_SEPARATOR = "\x00\x00\x1E\n";
const DOCUMENT_PAD = ' '.repeat(16 - DOCUMENT_SEPARATOR.length) + DOCUMENT_SEPARATOR;

const NES_EPOCH = new Date('2020-01-01T00:00:00');

/**
 * @param {number} dataSize
 * @returns {string} The data padded to 16 bytes alignment and ended with \0x1E (record separator) and a line break.
 */
function padData(dataSize) {
    const padSize = (DOCUMENT_ALIGNMENT - ((dataSize + DOCUMENT_SEPARATOR.length) % DOCUMENT_ALIGNMENT)) % DOCUMENT_ALIGNMENT;
    return DOCUMENT_PAD.substr(-padSize - DOCUMENT_SEPARATOR.length);
}

/**
 * A partition is a single file where the storage will write documents to depending on some partitioning rules.
 * In the case of an event store, this is most likely the (write) streams.
 */
class WritablePartition extends ReadablePartition {

    /**
     * @param {string} name The name of the partition.
     * @param {object} [config] An object with storage parameters.
     * @param {string} [config.dataDirectory] The path where the storage data should reside. Default '.'.
     * @param {number} [config.readBufferSize] Size of the read buffer in bytes. Default 4096.
     * @param {number} [config.writeBufferSize] Size of the write buffer in bytes. Default 16384.
     * @param {number} [config.maxWriteBufferDocuments] How many documents to have in the write buffer at max. 0 means as much as possible. Default 0.
     * @param {boolean} [config.syncOnFlush] If fsync should be called on write buffer flush. Set this if you need strict durability. Defaults to false.
     * @param {boolean} [config.dirtyReads] If dirty reads should be allowed. This means that writes that are in write buffer but not yet flushed can be read. Defaults to true.
     * @param {object} [config.metadata] A metadata object that will be written to the file header.
     * @param {typeof Clock} [config.clock] The constructor to a clock interface
     */
    constructor(name, config = {}) {
        let defaults = {
            writeBufferSize: DEFAULT_WRITE_BUFFER_SIZE,
            maxWriteBufferDocuments: 0,
            syncOnFlush: false,
            dirtyReads: true,
            metadata: {
                epoch: Date.now()
            },
            clock: Clock
        };
        config = Object.assign(defaults, config);
        super(name, config);
        if (!fs.existsSync(this.dataDirectory)) {
            mkdirpSync(this.dataDirectory);
        }
        this.fileMode = 'a+';
        this.writeBufferSize = config.writeBufferSize >>> 0; // jshint ignore:line
        this.maxWriteBufferDocuments = config.maxWriteBufferDocuments >>> 0; // jshint ignore:line
        this.syncOnFlush = !!config.syncOnFlush;
        this.dirtyReads = !!config.dirtyReads;
        this.metadata = config.metadata;
        assert(typeof(config.clock.prototype) === 'object' && typeof(config.clock.prototype.time) === 'function', 'Clock needs to implement the method time()');
        this.ClockConstructor = config.clock;
    }

    /**
     * Open the partition storage and create read and write buffers.
     *
     * @api
     * @returns {boolean}
     */
    open() {
        if (this.fd) {
            return true;
        }

        this.writeBuffer = Buffer.allocUnsafeSlow(this.writeBufferSize);
        // Where inside the write buffer the next write is added
        this.writeBufferCursor = 0;
        // How many documents are currently in the write buffer
        this.writeBufferDocuments = 0;
        this.flushCallbacks = [];

        if (super.open() === false) {
            const stat = fs.statSync(this.fileName);
            if (stat.size !== 0) {
                return false;
            }
            this.metadata.epoch = Date.now();
            this.writeMetadata();
            this.size = 0;
        }
        this.clock = new this.ClockConstructor(this.metadata.epoch || NES_EPOCH.getTime());

        return true;
    }

    /**
     * Close the partition and frees up all resources.
     *
     * @api
     * @returns void
     */
    close() {
        if (this.fd) {
            this.flush();
            fs.fsyncSync(this.fd);
        }
        if (this.writeBuffer) {
            this.writeBuffer = null;
            this.writeBufferCursor = 0;
            this.writeBufferDocuments = 0;
        }
        super.close();
    }

    /**
     * Write the header and metadata to the file.
     *
     * @private
     * @returns void
     */
    writeMetadata() {
        const metadataBuffer = buildMetadataHeader(ReadablePartition.HEADER_MAGIC, this.metadata);
        fs.writeSync(this.fd, metadataBuffer, 0, metadataBuffer.byteLength, 0);
        this.headerSize = metadataBuffer.byteLength;
    }

    /**
     * Flush the write buffer to disk.
     * This is a sync method and will invoke all previously registered flush callbacks.
     *
     * @returns {boolean}
     */
    flush() {
        if (!this.fd) {
            return false;
        }
        if (this.writeBufferCursor === 0) {
            return false;
        }

        fs.writeSync(this.fd, this.writeBuffer, 0, this.writeBufferCursor);
        if (this.syncOnFlush) {
            fs.fsyncSync(this.fd);
        }

        this.writeBufferCursor = 0;
        this.writeBufferDocuments = 0;
        this.flushCallbacks.forEach(callback => callback());
        this.flushCallbacks = [];

        return true;
    }

    /**
     * @private
     * @param {number} dataSize The size of the data that needs to go into the write buffer next.
     */
    flushIfWriteBufferTooSmall(dataSize) {
        if (this.writeBufferCursor > 0 && dataSize + this.writeBufferCursor > this.writeBuffer.byteLength) {
            this.flush();
        }
        if (this.writeBufferCursor === 0) {
            process.nextTick(() => this.flush());
        }
    }

    /**
     * @private
     */
    flushIfWriteBufferDocumentsFull() {
        if (this.maxWriteBufferDocuments > 0 && this.writeBufferDocuments >= this.maxWriteBufferDocuments) {
            this.flush();
        }
    }

    /**
     * @private
     * @param {Buffer} buffer The buffer to write the document header to
     * @param {number} offset The offset inside the buffer to start writing at
     * @param {number} dataSize The size of the document
     * @param {number|null} [sequenceNumber] The external sequence number for this document
     * @param {number|null} [time64] The number of microseconds relative to the partition base since the creation of the document
     * @returns {number} The size of the document header
     */
    writeDocumentHeader(buffer, offset, dataSize, sequenceNumber = null, time64 = null) {
        if (sequenceNumber === null) {
            sequenceNumber = 0;
        }
        if (time64 === null) {
            time64 = this.clock.time();
        }
        if (time64 < 0) {
            throw new Error('Time may not be negative!');
        }
        buffer.writeUInt32BE(dataSize, offset);
        buffer.writeUInt32BE(sequenceNumber, offset + 4);
        buffer.writeDoubleBE(time64, offset + 8);
        return DOCUMENT_HEADER_SIZE;
    }

    /**
     * Write the given data to the partition without buffering.
     * @private
     * @param {string} data The (padded) data to write to storage.
     * @param {number} dataSize The size of the raw document without padding.
     * @param {number} [sequenceNumber] The external sequence number to store with the document.
     * @param {function} [callback] A function that will be called when the document is written to disk.
     * @returns {number} Number of bytes written.
     */
    writeUnbuffered(data, dataSize, sequenceNumber, callback) {
        this.flush();
        const dataHeader = Buffer.alloc(DOCUMENT_HEADER_SIZE);
        this.writeDocumentHeader(dataHeader, 0, dataSize, sequenceNumber);

        let bytesWritten = 0;
        bytesWritten += fs.writeSync(this.fd, dataHeader);
        bytesWritten += fs.writeSync(this.fd, data);
        if (typeof callback === 'function') {
            process.nextTick(callback);
        }
        return bytesWritten;
    }

    /**
     * Write the given data to the partition with buffering. Will flush the write buffer if it is necessary.
     * @private
     * @param {string} data The (padded) data to write to storage.
     * @param {number} dataSize The size of the raw document without padding.
     * @param {number} [sequenceNumber] The external sequence number to store with the document.
     * @param {function} [callback] A function that will be called when the document is written to disk.
     * @returns {number} Number of bytes written.
     */
    writeBuffered(data, dataSize, sequenceNumber, callback) {
        const bytesToWrite = Buffer.byteLength(data, 'utf8') + DOCUMENT_HEADER_SIZE;
        this.flushIfWriteBufferTooSmall(bytesToWrite);

        let bytesWritten = 0;
        bytesWritten += this.writeDocumentHeader(this.writeBuffer, this.writeBufferCursor, dataSize, sequenceNumber);
        bytesWritten += this.writeBuffer.write(data, this.writeBufferCursor + bytesWritten, 'utf8');
        this.writeBufferCursor += bytesWritten;
        this.writeBufferDocuments++;
        if (typeof callback === 'function') {
            this.flushCallbacks.push(callback);
        }
        this.flushIfWriteBufferDocumentsFull();
        return bytesWritten;
    }

    /**
     * @api
     * @param {string} data The data to write to storage.
     * @param {number} [sequenceNumber] The external sequence number to store with the document.
     * @param {function} [callback] A function that will be called when the document is written to disk.
     * @returns {number|boolean} The file position at which the data was written or false on error.
     */
    write(data, sequenceNumber, callback) {
        if (!this.fd) {
            return false;
        }
        if (typeof sequenceNumber === 'function') {
            callback = sequenceNumber;
            sequenceNumber = null;
        }
        const dataSize = Buffer.byteLength(data, 'utf8');
        assert(dataSize <= 64 * 1024 * 1024, 'Document is too large! Maximum is 64 MB');

        data += padData(dataSize);
        const dataPosition = this.size;
        if (dataSize + DOCUMENT_HEADER_SIZE >= this.writeBuffer.byteLength * 4 / 5) {
            this.size += this.writeUnbuffered(data, dataSize, sequenceNumber, callback);
        } else {
            this.size += this.writeBuffered(data, dataSize, sequenceNumber, callback);
        }
        assert(this.size >= dataPosition + dataSize + DOCUMENT_HEADER_SIZE, `Error while writing document at position ${dataPosition}.`);
        return dataPosition;
    }

    /**
     * Prepare the read buffer for reading from the specified position.
     *
     * @protected
     * @param {number} position The position in the file to prepare the read buffer for reading from.
     * @returns {object} A reader object with properties `buffer`, `cursor` and `length`.
     */
    prepareReadBuffer(position) {
        if (position + DOCUMENT_HEADER_SIZE >= this.size) {
            return { buffer: null, cursor: 0, length: 0 };
        }
        const bufferPos = this.size - this.writeBufferCursor;
        // Handle the case when data that is still in write buffer is supposed to be read
        if (this.dirtyReads && this.writeBufferCursor > 0 && position >= bufferPos) {
            return { buffer: this.writeBuffer, cursor: position - bufferPos, length: this.writeBufferCursor };
        }
        return super.prepareReadBuffer(position);
    }

    /**
     * Truncate the internal read buffer after the given position.
     * @param {number} after The byte position to truncate the read buffer after.
     */
    truncateReadBuffer(after) {
        if (this.readBufferPos >= after) {
            this.readBufferPos = -1;
            this.readBufferLength = 0;
        } else if (this.readBufferPos + this.readBufferLength > after) {
            this.readBufferLength -= (this.readBufferPos + this.readBufferLength) - after;
        }
    }

    /**
     * Truncate the partition storage at the given position.
     *
     * @param {number} after The file position after which to truncate the partition.
     */
    truncate(after) {
        if (after > this.size) {
            return;
        }
        if (after < 0) {
            after = 0;
        }
        this.flush();

        let position = after, data;
        try {
            data = this.readFrom(position);
        } catch (e) {
            throw new Error('Can only truncate on valid document boundaries.');
        }
        // copy all truncated documents to some delete log
        const deletedBranch = new WritablePartition(this.name + '-' + after + '.branch', { dataDirectory: this.dataDirectory });
        deletedBranch.open();
        while (data) {
            deletedBranch.write(data);
            position += this.documentWriteSize(Buffer.byteLength(data, 'utf8'));
            data = this.readFrom(position);
        }
        deletedBranch.close();

        fs.truncateSync(this.fileName, this.headerSize + after);
        this.truncateReadBuffer(after);
        this.size = after;
    }
}

module.exports = WritablePartition;
module.exports.CorruptFileError = ReadablePartition.CorruptFileError;

1	const fs = require('fs');	4✔
2	const mkdirpSync = require('mkdirp').sync;	4✔
3	const ReadablePartition = require('./ReadablePartition');	4✔
4	const { assert, buildMetadataHeader } = require('../util');	4✔
5	const Clock = require('../Clock');	4✔
6
7	const DEFAULT_WRITE_BUFFER_SIZE = 16 * 1024;	4✔
8	const DOCUMENT_HEADER_SIZE = 16;	4✔
9	const DOCUMENT_ALIGNMENT = 4;	4✔
10	const DOCUMENT_SEPARATOR = "\x00\x00\x1E\n";	4✔
11	const DOCUMENT_PAD = ' '.repeat(16 - DOCUMENT_SEPARATOR.length) + DOCUMENT_SEPARATOR;	4✔
12
13	const NES_EPOCH = new Date('2020-01-01T00:00:00');	4✔
14
15	/**
16	* @param {number} dataSize
17	* @returns {string} The data padded to 16 bytes alignment and ended with \0x1E (record separator) and a line break.
18	*/
19	function padData(dataSize) {
20	const padSize = (DOCUMENT_ALIGNMENT - ((dataSize + DOCUMENT_SEPARATOR.length) % DOCUMENT_ALIGNMENT)) % DOCUMENT_ALIGNMENT;	7,152✔
21	return DOCUMENT_PAD.substr(-padSize - DOCUMENT_SEPARATOR.length);	7,152✔
22	}
23
24	/**
25	* A partition is a single file where the storage will write documents to depending on some partitioning rules.
26	* In the case of an event store, this is most likely the (write) streams.
27	*/
28	class WritablePartition extends ReadablePartition {
29
30	/**
31	* @param {string} name The name of the partition.
32	* @param {object} [config] An object with storage parameters.
33	* @param {string} [config.dataDirectory] The path where the storage data should reside. Default '.'.
34	* @param {number} [config.readBufferSize] Size of the read buffer in bytes. Default 4096.
35	* @param {number} [config.writeBufferSize] Size of the write buffer in bytes. Default 16384.
36	* @param {number} [config.maxWriteBufferDocuments] How many documents to have in the write buffer at max. 0 means as much as possible. Default 0.
37	* @param {boolean} [config.syncOnFlush] If fsync should be called on write buffer flush. Set this if you need strict durability. Defaults to false.
38	* @param {boolean} [config.dirtyReads] If dirty reads should be allowed. This means that writes that are in write buffer but not yet flushed can be read. Defaults to true.
39	* @param {object} [config.metadata] A metadata object that will be written to the file header.
40	* @param {typeof Clock} [config.clock] The constructor to a clock interface
41	*/
42	constructor(name, config = {}) {	8✔
43	let defaults = {	812✔
44	writeBufferSize: DEFAULT_WRITE_BUFFER_SIZE,
45	maxWriteBufferDocuments: 0,
46	syncOnFlush: false,
47	dirtyReads: true,
48	metadata: {
49	epoch: Date.now()
50	},
51	clock: Clock
52	};
53	config = Object.assign(defaults, config);	812✔
54	super(name, config);	812✔
55	if (!fs.existsSync(this.dataDirectory)) {	804✔
56	mkdirpSync(this.dataDirectory);	4✔
57	}
58	this.fileMode = 'a+';	804✔
59	this.writeBufferSize = config.writeBufferSize >>> 0; // jshint ignore:line	804✔
60	this.maxWriteBufferDocuments = config.maxWriteBufferDocuments >>> 0; // jshint ignore:line	804✔
61	this.syncOnFlush = !!config.syncOnFlush;	804✔
62	this.dirtyReads = !!config.dirtyReads;	804✔
63	this.metadata = config.metadata;	804✔
64	assert(typeof(config.clock.prototype) === 'object' && typeof(config.clock.prototype.time) === 'function', 'Clock needs to implement the method time()');	804✔
65	this.ClockConstructor = config.clock;	804✔
66	}
67
68	/**
69	* Open the partition storage and create read and write buffers.
70	*
71	* @api
72	* @returns {boolean}
73	*/
74	open() {
75	if (this.fd) {	5,064✔
76	return true;	4,152✔
77	}
78
79	this.writeBuffer = Buffer.allocUnsafeSlow(this.writeBufferSize);	912✔
80	// Where inside the write buffer the next write is added
81	this.writeBufferCursor = 0;	912✔
82	// How many documents are currently in the write buffer
83	this.writeBufferDocuments = 0;	912✔
84	this.flushCallbacks = [];	912✔
85
86	if (super.open() === false) {	912✔
87	const stat = fs.statSync(this.fileName);	728✔
88	if (stat.size !== 0) {	728!
UNCOV 89	return false;	×
90	}
91	this.metadata.epoch = Date.now();	728✔
92	this.writeMetadata();	728✔
93	this.size = 0;	728✔
94	}
95	this.clock = new this.ClockConstructor(this.metadata.epoch \|\| NES_EPOCH.getTime());	904!
96
97	return true;	904✔
98	}
99
100	/**
101	* Close the partition and frees up all resources.
102	*
103	* @api
104	* @returns void
105	*/
106	close() {
107	if (this.fd) {	1,200✔
108	this.flush();	912✔
109	fs.fsyncSync(this.fd);	912✔
110	}
111	if (this.writeBuffer) {	1,200✔
112	this.writeBuffer = null;	912✔
113	this.writeBufferCursor = 0;	912✔
114	this.writeBufferDocuments = 0;	912✔
115	}
116	super.close();	1,200✔
117	}
118
119	/**
120	* Write the header and metadata to the file.
121	*
122	* @private
123	* @returns void
124	*/
125	writeMetadata() {
126	const metadataBuffer = buildMetadataHeader(ReadablePartition.HEADER_MAGIC, this.metadata);	728✔
127	fs.writeSync(this.fd, metadataBuffer, 0, metadataBuffer.byteLength, 0);	728✔
128	this.headerSize = metadataBuffer.byteLength;	728✔
129	}
130
131	/**
132	* Flush the write buffer to disk.
133	* This is a sync method and will invoke all previously registered flush callbacks.
134	*
135	* @returns {boolean}
136	*/
137	flush() {
138	if (!this.fd) {	2,036✔
139	return false;	100✔
140	}
141	if (this.writeBufferCursor === 0) {	1,936✔
142	return false;	1,080✔
143	}
144
145	fs.writeSync(this.fd, this.writeBuffer, 0, this.writeBufferCursor);	856✔
146	if (this.syncOnFlush) {	856✔
147	fs.fsyncSync(this.fd);	40✔
148	}
149
150	this.writeBufferCursor = 0;	856✔
151	this.writeBufferDocuments = 0;	856✔
152	this.flushCallbacks.forEach(callback => callback());	856✔
153	this.flushCallbacks = [];	856✔
154
155	return true;	856✔
156	}
157
158	/**
159	* @private
160	* @param {number} dataSize The size of the data that needs to go into the write buffer next.
161	*/
162	flushIfWriteBufferTooSmall(dataSize) {
163	if (this.writeBufferCursor > 0 && dataSize + this.writeBufferCursor > this.writeBuffer.byteLength) {	7,144✔
164	this.flush();	100✔
165	}
166	if (this.writeBufferCursor === 0) {	7,144✔
167	process.nextTick(() => this.flush());	856✔
168	}
169	}
170
171	/**
172	* @private
173	*/
174	flushIfWriteBufferDocumentsFull() {
175	if (this.maxWriteBufferDocuments > 0 && this.writeBufferDocuments >= this.maxWriteBufferDocuments) {	7,144✔
176	this.flush();	4✔
177	}
178	}
179
180	/**
181	* @private
182	* @param {Buffer} buffer The buffer to write the document header to
183	* @param {number} offset The offset inside the buffer to start writing at
184	* @param {number} dataSize The size of the document
185	* @param {number\|null} [sequenceNumber] The external sequence number for this document
186	* @param {number\|null} [time64] The number of microseconds relative to the partition base since the creation of the document
187	* @returns {number} The size of the document header
188	*/
189	writeDocumentHeader(buffer, offset, dataSize, sequenceNumber = null, time64 = null) {	7,424✔
190	if (sequenceNumber === null) {	7,152✔
191	sequenceNumber = 0;	280✔
192	}
193	if (time64 === null) {	7,152!
194	time64 = this.clock.time();	7,152✔
195	}
196	if (time64 < 0) {	7,152!
UNCOV 197	throw new Error('Time may not be negative!');	×
198	}
199	buffer.writeUInt32BE(dataSize, offset);	7,152✔
200	buffer.writeUInt32BE(sequenceNumber, offset + 4);	7,152✔
201	buffer.writeDoubleBE(time64, offset + 8);	7,152✔
202	return DOCUMENT_HEADER_SIZE;	7,152✔
203	}
204
205	/**
206	* Write the given data to the partition without buffering.
207	* @private
208	* @param {string} data The (padded) data to write to storage.
209	* @param {number} dataSize The size of the raw document without padding.
210	* @param {number} [sequenceNumber] The external sequence number to store with the document.
211	* @param {function} [callback] A function that will be called when the document is written to disk.
212	* @returns {number} Number of bytes written.
213	*/
214	writeUnbuffered(data, dataSize, sequenceNumber, callback) {
215	this.flush();	8✔
216	const dataHeader = Buffer.alloc(DOCUMENT_HEADER_SIZE);	8✔
217	this.writeDocumentHeader(dataHeader, 0, dataSize, sequenceNumber);	8✔
218
219	let bytesWritten = 0;	8✔
220	bytesWritten += fs.writeSync(this.fd, dataHeader);	8✔
221	bytesWritten += fs.writeSync(this.fd, data);	8✔
222	if (typeof callback === 'function') {	8✔
223	process.nextTick(callback);	4✔
224	}
225	return bytesWritten;	8✔
226	}
227
228	/**
229	* Write the given data to the partition with buffering. Will flush the write buffer if it is necessary.
230	* @private
231	* @param {string} data The (padded) data to write to storage.
232	* @param {number} dataSize The size of the raw document without padding.
233	* @param {number} [sequenceNumber] The external sequence number to store with the document.
234	* @param {function} [callback] A function that will be called when the document is written to disk.
235	* @returns {number} Number of bytes written.
236	*/
237	writeBuffered(data, dataSize, sequenceNumber, callback) {
238	const bytesToWrite = Buffer.byteLength(data, 'utf8') + DOCUMENT_HEADER_SIZE;	7,144✔
239	this.flushIfWriteBufferTooSmall(bytesToWrite);	7,144✔
240
241	let bytesWritten = 0;	7,144✔
242	bytesWritten += this.writeDocumentHeader(this.writeBuffer, this.writeBufferCursor, dataSize, sequenceNumber);	7,144✔
243	bytesWritten += this.writeBuffer.write(data, this.writeBufferCursor + bytesWritten, 'utf8');	7,144✔
244	this.writeBufferCursor += bytesWritten;	7,144✔
245	this.writeBufferDocuments++;	7,144✔
246	if (typeof callback === 'function') {	7,144✔
247	this.flushCallbacks.push(callback);	564✔
248	}
249	this.flushIfWriteBufferDocumentsFull();	7,144✔
250	return bytesWritten;	7,144✔
251	}
252
253	/**
254	* @api
255	* @param {string} data The data to write to storage.
256	* @param {number} [sequenceNumber] The external sequence number to store with the document.
257	* @param {function} [callback] A function that will be called when the document is written to disk.
258	* @returns {number\|boolean} The file position at which the data was written or false on error.
259	*/
260	write(data, sequenceNumber, callback) {
261	if (!this.fd) {	7,156✔
262	return false;	4✔
263	}
264	if (typeof sequenceNumber === 'function') {	7,152✔
265	callback = sequenceNumber;	8✔
266	sequenceNumber = null;	8✔
267	}
268	const dataSize = Buffer.byteLength(data, 'utf8');	7,152✔
269	assert(dataSize <= 64 * 1024 * 1024, 'Document is too large! Maximum is 64 MB');	7,152✔
270
271	data += padData(dataSize);	7,152✔
272	const dataPosition = this.size;	7,152✔
273	if (dataSize + DOCUMENT_HEADER_SIZE >= this.writeBuffer.byteLength * 4 / 5) {	7,152✔
274	this.size += this.writeUnbuffered(data, dataSize, sequenceNumber, callback);	8✔
275	} else {
276	this.size += this.writeBuffered(data, dataSize, sequenceNumber, callback);	7,144✔
277	}
278	assert(this.size >= dataPosition + dataSize + DOCUMENT_HEADER_SIZE, `Error while writing document at position ${dataPosition}.`);	7,152✔
279	return dataPosition;	7,152✔
280	}
281
282	/**
283	* Prepare the read buffer for reading from the specified position.
284	*
285	* @protected
286	* @param {number} position The position in the file to prepare the read buffer for reading from.
287	* @returns {object} A reader object with properties `buffer`, `cursor` and `length`.
288	*/
289	prepareReadBuffer(position) {
290	if (position + DOCUMENT_HEADER_SIZE >= this.size) {	7,540✔
291	return { buffer: null, cursor: 0, length: 0 };	56✔
292	}
293	const bufferPos = this.size - this.writeBufferCursor;	7,484✔
294	// Handle the case when data that is still in write buffer is supposed to be read
295	if (this.dirtyReads && this.writeBufferCursor > 0 && position >= bufferPos) {	7,484✔
296	return { buffer: this.writeBuffer, cursor: position - bufferPos, length: this.writeBufferCursor };	2,072✔
297	}
298	return super.prepareReadBuffer(position);	5,412✔
299	}
300
301	/**
302	* Truncate the internal read buffer after the given position.
303	* @param {number} after The byte position to truncate the read buffer after.
304	*/
305	truncateReadBuffer(after) {
306	if (this.readBufferPos >= after) {	48✔
307	this.readBufferPos = -1;	44✔
308	this.readBufferLength = 0;	44✔
309	} else if (this.readBufferPos + this.readBufferLength > after) {	4!
UNCOV 310	this.readBufferLength -= (this.readBufferPos + this.readBufferLength) - after;	×
311	}
312	}
313
314	/**
315	* Truncate the partition storage at the given position.
316	*
317	* @param {number} after The file position after which to truncate the partition.
318	*/
319	truncate(after) {
320	if (after > this.size) {	56✔
321	return;	4✔
322	}
323	if (after < 0) {	52✔
324	after = 0;	4✔
325	}
326	this.flush();	52✔
327
328	let position = after, data;	52✔
329	try {	52✔
330	data = this.readFrom(position);	52✔
331	} catch (e) {
332	throw new Error('Can only truncate on valid document boundaries.');	4✔
333	}
334	// copy all truncated documents to some delete log
335	const deletedBranch = new WritablePartition(this.name + '-' + after + '.branch', { dataDirectory: this.dataDirectory });	48✔
336	deletedBranch.open();	48✔
337	while (data) {	48✔
338	deletedBranch.write(data);	200✔
339	position += this.documentWriteSize(Buffer.byteLength(data, 'utf8'));	200✔
340	data = this.readFrom(position);	200✔
341	}
342	deletedBranch.close();	48✔
343
344	fs.truncateSync(this.fileName, this.headerSize + after);	48✔
345	this.truncateReadBuffer(after);	48✔
346	this.size = after;	48✔
347	}
348	}
349
350	module.exports = WritablePartition;	4✔
351	module.exports.CorruptFileError = ReadablePartition.CorruptFileError;	4✔

albe / node-event-storage / 552719872

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous