smartfile/ts/classes.streamfile.ts

154 lines
5.1 KiB
TypeScript
Raw Normal View History

2024-04-02 18:53:02 +00:00
import * as plugins from './plugins.js';
2023-11-04 19:14:20 +00:00
import * as smartfileFs from './fs.js';
import * as smartfileFsStream from './fsstream.js';
2023-11-04 19:07:43 +00:00
import { Readable } from 'stream';
2023-11-04 19:54:13 +00:00
type TStreamSource = (streamFile: StreamFile) => Promise<Readable>;
2023-11-04 19:07:43 +00:00
/**
* The StreamFile class represents a file as a stream.
* It allows creating streams from a file path, a URL, or a buffer.
*/
export class StreamFile {
// STATIC
public static async fromPath(filePath: string): Promise<StreamFile> {
2023-11-04 19:54:13 +00:00
const streamSource: TStreamSource = async (stremFileArg) => smartfileFsStream.createReadStream(filePath);
2023-11-04 19:43:54 +00:00
const streamFile = new StreamFile(streamSource, filePath);
streamFile.multiUse = true;
return streamFile;
2023-11-04 19:07:43 +00:00
}
public static async fromUrl(url: string): Promise<StreamFile> {
2023-11-04 19:54:13 +00:00
const streamSource: TStreamSource = async (streamFileArg) => plugins.smartrequest.getStream(url); // Replace with actual plugin method
2023-11-04 19:43:54 +00:00
const streamFile = new StreamFile(streamSource);
streamFile.multiUse = true;
return streamFile;
2023-11-04 19:07:43 +00:00
}
public static fromBuffer(buffer: Buffer, relativeFilePath?: string): StreamFile {
2023-11-04 19:54:13 +00:00
const streamSource: TStreamSource = async (streamFileArg) => {
2023-11-04 19:07:43 +00:00
const stream = new Readable();
stream.push(buffer);
stream.push(null); // End of stream
2023-11-04 19:54:13 +00:00
return stream;
2023-11-04 19:07:43 +00:00
};
2023-11-04 19:43:54 +00:00
const streamFile = new StreamFile(streamSource, relativeFilePath);
streamFile.multiUse = true;
return streamFile;
}
/**
* Creates a StreamFile from an existing Readable stream with an option for multiple uses.
* @param stream A Node.js Readable stream.
* @param relativeFilePath Optional file path for the stream.
* @param multiUse If true, the stream can be read multiple times, caching its content.
* @returns A StreamFile instance.
*/
public static fromStream(stream: Readable, relativeFilePath?: string, multiUse: boolean = false): StreamFile {
2023-11-04 19:54:13 +00:00
const streamSource: TStreamSource = (streamFileArg) => {
if (streamFileArg.multiUse) {
2023-11-04 19:43:54 +00:00
// If multi-use is enabled and we have cached content, create a new readable stream from the buffer
const bufferedStream = new Readable();
2023-11-04 19:54:13 +00:00
bufferedStream.push(streamFileArg.cachedStreamBuffer);
2023-11-04 19:43:54 +00:00
bufferedStream.push(null); // No more data to push
return Promise.resolve(bufferedStream);
} else {
return Promise.resolve(stream);
}
};
const streamFile = new StreamFile(streamSource, relativeFilePath);
streamFile.multiUse = multiUse;
// If multi-use is enabled, cache the stream when it's first read
if (multiUse) {
const chunks: Buffer[] = [];
stream.on('data', (chunk) => chunks.push(Buffer.from(chunk)));
stream.on('end', () => {
streamFile.cachedStreamBuffer = Buffer.concat(chunks);
});
// It's important to handle errors that may occur during streaming
stream.on('error', (err) => {
console.error('Error while caching stream:', err);
});
}
return streamFile;
2023-11-04 19:07:43 +00:00
}
2024-06-06 20:29:06 +00:00
// INSTANCE
relativeFilePath?: string;
private streamSource: TStreamSource;
// enable stream based multi use
private cachedStreamBuffer?: Buffer;
public multiUse: boolean;
public used: boolean = false;
private constructor(streamSource: TStreamSource, relativeFilePath?: string) {
this.streamSource = streamSource;
this.relativeFilePath = relativeFilePath;
}
2023-11-04 19:07:43 +00:00
// METHODS
2023-11-04 19:43:54 +00:00
private checkMultiUse() {
if (!this.multiUse && this.used) {
throw new Error('This stream can only be used once.');
}
this.used = true;
}
2023-11-04 19:07:43 +00:00
/**
* Creates a new readable stream from the source.
*/
public async createReadStream(): Promise<Readable> {
2023-11-04 19:54:13 +00:00
return this.streamSource(this);
2023-11-04 19:07:43 +00:00
}
/**
* Writes the stream to the disk at the specified path.
* @param filePathArg The file path where the stream should be written.
*/
public async writeToDisk(filePathArg: string): Promise<void> {
2023-11-04 19:43:54 +00:00
this.checkMultiUse();
2023-11-04 19:07:43 +00:00
const readStream = await this.createReadStream();
2023-11-04 19:14:20 +00:00
const writeStream = smartfileFsStream.createWriteStream(filePathArg);
2023-11-04 19:07:43 +00:00
return new Promise((resolve, reject) => {
readStream.pipe(writeStream);
readStream.on('error', reject);
writeStream.on('error', reject);
writeStream.on('finish', resolve);
});
}
public async writeToDir(dirPathArg: string) {
2023-11-04 19:43:54 +00:00
this.checkMultiUse();
2023-11-04 19:07:43 +00:00
const filePath = plugins.path.join(dirPathArg, this.relativeFilePath);
2023-11-04 19:14:20 +00:00
await smartfileFs.ensureDir(plugins.path.parse(filePath).dir);
2023-11-04 19:07:43 +00:00
return this.writeToDisk(filePath);
}
public async getContentAsBuffer() {
2023-11-04 19:43:54 +00:00
this.checkMultiUse();
2023-11-04 19:07:43 +00:00
const done = plugins.smartpromise.defer<Buffer>();
const readStream = await this.createReadStream();
const chunks: Buffer[] = [];
readStream.on('data', (chunk) => chunks.push(Buffer.from(chunk)));
readStream.on('error', done.reject);
readStream.on('end', () => {
const contentBuffer = Buffer.concat(chunks);
done.resolve(contentBuffer);
});
return done.promise;
}
public async getContentAsString(formatArg: 'utf8' | 'binary' = 'utf8') {
const contentBuffer = await this.getContentAsBuffer();
return contentBuffer.toString(formatArg);
}
}