Open way modify
This commit is contained in:
@@ -3,7 +3,7 @@ import * as fs from 'node:fs';
|
||||
import { DATA_HEADER_SIZE } from './constants.js';
|
||||
import { DataProtocol, DataHeader } from './protocol.js';
|
||||
import { IndexReader } from '../idx/index.js';
|
||||
import type { Serializer, DataEntry } from './types.js';
|
||||
import type { Serializer, DataEntry, DataFileReaderOptions } from './types.js';
|
||||
|
||||
export class DataReader<T> {
|
||||
private fd: number | null = null;
|
||||
@@ -12,17 +12,18 @@ export class DataReader<T> {
|
||||
private indexReader: IndexReader;
|
||||
private serializer: Serializer<T>;
|
||||
|
||||
readonly dataPath: string;
|
||||
readonly indexPath: string;
|
||||
private dataPath: string | null = null;
|
||||
private indexPath: string | null = null;
|
||||
|
||||
constructor(basePath: string, serializer: Serializer<T>) {
|
||||
this.dataPath = `${basePath}.dat`;
|
||||
this.indexPath = `${basePath}.idx`;
|
||||
this.serializer = serializer;
|
||||
this.indexReader = new IndexReader(this.indexPath);
|
||||
constructor(options: DataFileReaderOptions<T>) {
|
||||
this.serializer = options.serializer;
|
||||
this.indexReader = new IndexReader();
|
||||
}
|
||||
|
||||
open(): void {
|
||||
open(basePath: string): void {
|
||||
this.dataPath = `${basePath}.dat`;
|
||||
this.indexPath = `${basePath}.idx`;
|
||||
|
||||
this.fd = fs.openSync(this.dataPath, 'r');
|
||||
|
||||
// Read header only
|
||||
@@ -30,7 +31,7 @@ export class DataReader<T> {
|
||||
fs.readSync(this.fd, headerBuf, 0, DATA_HEADER_SIZE, 0);
|
||||
this.header = DataProtocol.readHeader(headerBuf);
|
||||
|
||||
this.indexReader.open();
|
||||
this.indexReader.open(this.indexPath);
|
||||
}
|
||||
|
||||
private readRecord(offset: bigint, length: number): Buffer {
|
||||
@@ -204,6 +205,10 @@ export class DataReader<T> {
|
||||
return this.indexReader.getHeader().latestSequence;
|
||||
}
|
||||
|
||||
getFlags(): number {
|
||||
return this.indexReader.getFlags();
|
||||
}
|
||||
|
||||
close(): void {
|
||||
if (this.fd !== null) {
|
||||
fs.closeSync(this.fd);
|
||||
|
||||
@@ -17,3 +17,7 @@ export interface DataFileOptions<T> {
|
||||
forceTruncate?: boolean;
|
||||
indexFileOpt: IndexFileOptions;
|
||||
}
|
||||
|
||||
export interface DataFileReaderOptions<T> {
|
||||
serializer: Serializer<T>;
|
||||
}
|
||||
|
||||
@@ -117,6 +117,7 @@ export class DataWriter<T> {
|
||||
this.currentOffset += BigInt(buf.length);
|
||||
++this.recordCount;
|
||||
|
||||
this.writeHeader();
|
||||
return this.latestSequence;
|
||||
}
|
||||
|
||||
@@ -149,8 +150,11 @@ export class DataWriter<T> {
|
||||
return this.latestSequence;
|
||||
}
|
||||
|
||||
getNextSequence(): number {
|
||||
return this.latestSequence + 1;
|
||||
writeHeader(): void {
|
||||
if (this.fd === null || !this.headerBuf) return;
|
||||
|
||||
DataProtocol.updateHeader(this.headerBuf, this.currentOffset, this.recordCount);
|
||||
fs.writeSync(this.fd, this.headerBuf, 0, DATA_HEADER_SIZE, 0);
|
||||
}
|
||||
|
||||
sync(): void {
|
||||
@@ -159,8 +163,6 @@ export class DataWriter<T> {
|
||||
DataProtocol.updateHeader(this.headerBuf, this.currentOffset, this.recordCount);
|
||||
fs.writeSync(this.fd, this.headerBuf, 0, DATA_HEADER_SIZE, 0);
|
||||
fs.fsyncSync(this.fd);
|
||||
|
||||
this.indexWriter.syncAll();
|
||||
}
|
||||
|
||||
close(): void {
|
||||
@@ -175,6 +177,10 @@ export class DataWriter<T> {
|
||||
this.headerBuf = null;
|
||||
}
|
||||
|
||||
writeFlags(flags: number): void {
|
||||
this.indexWriter.writeFlags(flags);
|
||||
}
|
||||
|
||||
getStats() {
|
||||
return {
|
||||
dataPath: this.dataPath,
|
||||
|
||||
Reference in New Issue
Block a user