helpy4/immich
1
0
Fork 0
mirror of https://github.com/immich-app/immich synced 2025-11-14 17:36:12 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

test interruption + abort

Browse source
This commit is contained in:
mertalev 2025-10-08 16:41:42 -04:00
parent ebda00fcf0
commit c295a48061
No known key found for this signature in database
GPG key ID: DF6ABC77AAD98C95
2 changed files with 110 additions and 31 deletions
Download patch file Download diff file Expand all files Collapse all files

33
server/src/services/asset-upload.service.ts
View file

@ -38,7 +38,7 @@ export class AssetUploadService extends BaseService {
// we can assume the previous request has already failed on the client end.
private activeRequests = new Map<string, { req: Readable; startTime: Date }>();
@OnEvent({ name: 'UploadAbort', workers: [ImmichWorker.Api] })
@OnEvent({ name: 'UploadAbort', workers: [ImmichWorker.Api], server: true })
onUploadAbort({ assetId, abortTime }: ArgOf<'UploadAbort'>) {
const entry = this.activeRequests.get(assetId);
if (!entry) {
@ -80,14 +80,13 @@ export class AssetUploadService extends BaseService {
this.addRequest(asset.id, req);
let checksumBuffer: Buffer | undefined;
const writeStream = this.storageRepository.createOrAppendWriteStream(asset.path);
const writeStream = this.pipe(req, asset.path, contentLength);
if (isComplete) {
const hash = createHash('sha1');
req.on('data', (data: Buffer) => hash.update(data));
writeStream.on('finish', () => (checksumBuffer = hash.digest()));
}
req.pipe(writeStream);
await new Promise((resolve, reject) => writeStream.on('finish', resolve).on('close', reject));
await new Promise((resolve, reject) => writeStream.on('close', resolve).on('error', reject));
this.setCompleteHeader(res, dto.version, isComplete);
if (!isComplete) {
res.status(201).set('Location', location).setHeader('Upload-Limit', 'min-size=0').send();
@ -139,9 +138,8 @@ export class AssetUploadService extends BaseService {
return;
}
const writeStream = this.storageRepository.createOrAppendWriteStream(path);
req.pipe(writeStream);
await new Promise((resolve, reject) => writeStream.on('finish', resolve).on('close', reject));
const writeStream = this.pipe(req, path, contentLength);
await new Promise((resolve, reject) => writeStream.on('close', resolve).on('error', reject));
this.setCompleteHeader(res, version, isComplete);
if (!isComplete) {
try {
@ -334,6 +332,27 @@ export class AssetUploadService extends BaseService {
}
}
private pipe(req: Readable, path: string, size: number) {
const writeStream = this.storageRepository.createOrAppendWriteStream(path);
let receivedLength = 0;
req.on('data', (data: Buffer) => {
receivedLength += data.length;
if (!writeStream.write(data)) {
req.pause();
writeStream.once('drain', () => req.resume());
}
});
req.on('close', () => {
if (receivedLength < size) {
writeStream.emit('error', new Error('Request closed before all data received'));
}
writeStream.end();
});
return writeStream;
}
private sendInterimResponse({ socket }: Response, location: string, interopVersion: number): void {
if (socket && !socket.destroyed) {
// Express doesn't understand interim responses, so write directly to socket