From fb7f614975dfb6620bccd594cf0fdcc8a01ad7b0 Mon Sep 17 00:00:00 2001 From: rkodev <43806892+rkodev@users.noreply.github.com> Date: Tue, 11 Feb 2025 10:40:03 +0300 Subject: [PATCH 01/20] Add large file upload --- src/index.ts | 1 + src/task/IProgress.ts | 3 + src/task/LargeFileUploadTask.ts | 150 ++++++++++++++++++++++++ src/task/UploadResponseHandler.ts | 20 ++++ src/task/UploadResponseHandlerOption.ts | 6 + src/task/UploadResult.ts | 41 +++++++ src/task/UploadSession.ts | 6 + src/task/UploadSliceRequestBuilder.ts | 65 ++++++++++ src/task/index.ts | 1 + test/task/LargeFileUploadTask.ts | 0 10 files changed, 293 insertions(+) create mode 100644 src/task/IProgress.ts create mode 100644 src/task/LargeFileUploadTask.ts create mode 100644 src/task/UploadResponseHandler.ts create mode 100644 src/task/UploadResponseHandlerOption.ts create mode 100644 src/task/UploadResult.ts create mode 100644 src/task/UploadSession.ts create mode 100644 src/task/UploadSliceRequestBuilder.ts create mode 100644 src/task/index.ts create mode 100644 test/task/LargeFileUploadTask.ts diff --git a/src/index.ts b/src/index.ts index f5da039..c44a93b 100644 --- a/src/index.ts +++ b/src/index.ts @@ -2,5 +2,6 @@ export * from "./adapter/index.js"; export * from "./http/index.js"; export * from "./middleware/index.js"; export * from "./authentication/index.js"; +export * from "./task/index.js"; export * from "./utils/Constants.js"; export * from "./utils/Version.js"; diff --git a/src/task/IProgress.ts b/src/task/IProgress.ts new file mode 100644 index 0000000..0f9591d --- /dev/null +++ b/src/task/IProgress.ts @@ -0,0 +1,3 @@ +export interface IProgress { + report(progress: number): void; +} diff --git a/src/task/LargeFileUploadTask.ts b/src/task/LargeFileUploadTask.ts new file mode 100644 index 0000000..4fd040a --- /dev/null +++ b/src/task/LargeFileUploadTask.ts @@ -0,0 +1,150 @@ +import { Parsable, RequestAdapter } from "@microsoft/kiota-abstractions"; +import { UploadSession } from "./UploadSession"; +import { UploadSliceRequestBuilder } from "./UploadSliceRequestBuilder"; +import { UploadResult } from "./UploadResult"; +import { IProgress } from "./IProgress"; + +export interface ILargeFileUploadTask { + Upload(progressEventHandler: IProgress): Promise>; + + Resume(progressEventHandler: IProgress): Promise>; + + RefreshUploadStatus(): Promise; + + UpdateSession(): Promise; + + DeleteSession(): Promise; + + Cancel(): Promise; +} + +const DefaultSliceSize = 320 * 1024; + +export class LargeFileUploadTask implements ILargeFileUploadTask { + rangesRemaining: number[][] = []; + Session: UploadSession; + + constructor( + readonly uploadSession: Parsable, + readonly uploadStream: ReadableStream, + readonly maxSliceSize = -1, + readonly requestAdapter: RequestAdapter, + ) { + if (!uploadStream?.locked) { + throw new Error("Please provide stream value"); + } + if (requestAdapter === undefined) { + throw new Error("Request adapter is a required parameter"); + } + if (maxSliceSize <= 0) { + this.maxSliceSize = DefaultSliceSize; + } + + this.Session = this.extractSessionInfo(uploadSession); + this.rangesRemaining = this.GetRangesRemaining(this.Session); + } + + public async Upload(progress?: IProgress, maxTries = 3): Promise> { + let uploadTries = 0; + while (uploadTries < maxTries) { + const sliceRequests = this.GetUploadSliceRequests(); + for (const request of sliceRequests) { + const uploadResult = await request.UploadSlice(this.uploadStream); + progress?.report(request.rangeEnd); + if (uploadResult?.UploadSucceeded()) { + return uploadResult; + } + } + + await this.UpdateSession(); + uploadTries++; + + if (uploadTries < maxTries) { + // Exponential backoff + await this.sleep(2000 * (uploadTries + 1)); + } + } + + throw new Error("Max retries reached"); + } + + public Resume(_?: IProgress): Promise> { + throw new Error("Method not implemented."); + } + + public RefreshUploadStatus(): Promise { + throw new Error("Method not implemented."); + } + + public UpdateSession(): Promise { + throw new Error("Method not implemented."); + } + + public DeleteSession(): Promise { + throw new Error("Method not implemented."); + } + + public Cancel(): Promise { + throw new Error("Method not implemented."); + } + + private extractSessionInfo(parsable: Parsable): UploadSession { + const uploadSession: UploadSession = { + expirationDateTime: null, + nextExpectedRanges: null, + odataType: null, + uploadUrl: null, + }; + + if ("expirationDateTime" in parsable) uploadSession.expirationDateTime = parsable.expirationDateTime as Date | null; + if ("nextExpectedRanges" in parsable) + uploadSession.nextExpectedRanges = parsable.nextExpectedRanges as string[] | null; + if ("odataType" in parsable) uploadSession.odataType = parsable.odataType as string | null; + if ("uploadUrl" in parsable) uploadSession.uploadUrl = parsable.uploadUrl as string | null; + + return uploadSession; + } + + private sleep(ms: number): Promise { + return new Promise(resolve => setTimeout(resolve, ms)); + } + + private GetUploadSliceRequests(): UploadSliceRequestBuilder[] { + const uploadSlices: UploadSliceRequestBuilder[] = []; + const rangesRemaining = this.rangesRemaining; + const session = this.Session; + rangesRemaining.forEach(range => { + let currentRangeBegin = range[0]; + while (currentRangeBegin <= range[1]) { + const nextSliceSize = this.nextSliceSize(currentRangeBegin, range[1]); + const uploadRequest = new UploadSliceRequestBuilder( + this.requestAdapter, + session.uploadUrl!, + currentRangeBegin, + currentRangeBegin + nextSliceSize - 1, + range[1] + 1, + ); + uploadSlices.push(uploadRequest); + currentRangeBegin += nextSliceSize; + } + }); + return uploadSlices; + } + + private nextSliceSize(currentRangeBegin: number, currentRangeEnd: number): number { + const sizeBasedOnRange = currentRangeEnd - currentRangeBegin + 1; + return sizeBasedOnRange > this.maxSliceSize ? this.maxSliceSize : sizeBasedOnRange; + } + + private GetRangesRemaining(uploadSession: UploadSession): number[][] { + // nextExpectedRanges: https://dev.onedrive.com/items/upload_large_files.htm + // Sample: ["12345-55232","77829-99375"] + // Also, second number in range can be blank, which means 'until the end' + const ranges: number[][] = []; + uploadSession.nextExpectedRanges?.forEach(rangeString => { + const rangeArray = rangeString.split("-"); + ranges.push([parseInt(rangeArray[0], 10), parseInt(rangeArray[1], 10)]); + }); + return ranges; + } +} diff --git a/src/task/UploadResponseHandler.ts b/src/task/UploadResponseHandler.ts new file mode 100644 index 0000000..ee68bea --- /dev/null +++ b/src/task/UploadResponseHandler.ts @@ -0,0 +1,20 @@ +import { ErrorMappings, ResponseHandler } from "@microsoft/kiota-abstractions"; + +export class UploadResponseHandler implements ResponseHandler { + handleResponse( + response: NativeResponseType, + _: ErrorMappings | undefined, + ): Promise { + if (response instanceof Response) { + if (response.ok) { + if (response.body != null) { + const body = response.body as unknown as ModelType; + return Promise.resolve(body); + } else { + return Promise.resolve(undefined); + } + } + } + return Promise.resolve(undefined); + } +} diff --git a/src/task/UploadResponseHandlerOption.ts b/src/task/UploadResponseHandlerOption.ts new file mode 100644 index 0000000..28be098 --- /dev/null +++ b/src/task/UploadResponseHandlerOption.ts @@ -0,0 +1,6 @@ +import { ResponseHandler, ResponseHandlerOption } from "@microsoft/kiota-abstractions"; +import { UploadResponseHandler } from "./UploadResponseHandler"; + +export class UploadResponseHandlerOption extends ResponseHandlerOption { + public responseHandler?: ResponseHandler = new UploadResponseHandler(); +} diff --git a/src/task/UploadResult.ts b/src/task/UploadResult.ts new file mode 100644 index 0000000..5ae2239 --- /dev/null +++ b/src/task/UploadResult.ts @@ -0,0 +1,41 @@ +import { Parsable, ParseNode } from "@microsoft/kiota-abstractions"; +import { UploadSession } from "./UploadSession"; +import { ILargeFileUploadTask } from "./LargeFileUploadTask"; + +export class UploadResult { + uploadSession?: UploadSession; + uploadTask?: ILargeFileUploadTask; + itemResponse?: T; + location?: string; + + UploadSucceeded() { + return this.itemResponse !== undefined || this.location !== undefined; + } +} + +// eslint-disable-next-line prefer-arrow/prefer-arrow-functions +export function createUploadResult( + _: ParseNode | undefined, +): (instance?: Parsable) => Record void> { + return deserializeIntoUploadResult; +} + +// eslint-disable-next-line prefer-arrow/prefer-arrow-functions +export function deserializeIntoUploadResult( + uploadResult: Partial> | undefined = {}, +): Record void> { + return { + uploadSession: _ => { + uploadResult.uploadSession = undefined; + }, + uploadTask: _ => { + uploadResult.uploadSession = undefined; + }, + itemResponse: _ => { + uploadResult.uploadSession = undefined; + }, + location: _ => { + uploadResult.uploadSession = undefined; + }, + }; +} diff --git a/src/task/UploadSession.ts b/src/task/UploadSession.ts new file mode 100644 index 0000000..90318a8 --- /dev/null +++ b/src/task/UploadSession.ts @@ -0,0 +1,6 @@ +export interface UploadSession { + expirationDateTime?: Date | null; + nextExpectedRanges?: string[] | null; + odataType?: string | null; + uploadUrl?: string | null; +} diff --git a/src/task/UploadSliceRequestBuilder.ts b/src/task/UploadSliceRequestBuilder.ts new file mode 100644 index 0000000..dc13159 --- /dev/null +++ b/src/task/UploadSliceRequestBuilder.ts @@ -0,0 +1,65 @@ +import { Parsable, RequestAdapter, RequestInformation } from "@microsoft/kiota-abstractions"; +import { createUploadResult, UploadResult } from "./UploadResult"; +import { Headers } from "@microsoft/kiota-abstractions/dist/es/src/headers"; +import { HttpMethod } from "@microsoft/kiota-abstractions/dist/es/src/httpMethod"; +import { HeadersInspectionOptions } from "@microsoft/kiota-http-fetchlibrary"; +import { UploadResponseHandlerOption } from "./UploadResponseHandlerOption"; + +const binaryContentType = "application/octet-stream"; + +export class UploadSliceRequestBuilder { + constructor( + readonly requestAdapter: RequestAdapter, + readonly sessionUrl: string, + readonly rangeBegin: number, + readonly rangeEnd: number, + readonly totalSessionLength: number, + ) {} + + public async UploadSlice(stream: ReadableStream): Promise | undefined> { + const data = await this.readSection(stream, this.rangeBegin, this.rangeEnd); + const requestInformation = this.createPutRequestInformation(data); + + const responseHandler = new UploadResponseHandlerOption(); + + const headerOptions = new HeadersInspectionOptions({ inspectResponseHeaders: true }); + requestInformation.addRequestOptions([headerOptions, responseHandler]); + + return this.requestAdapter.send>(requestInformation, createUploadResult, undefined); + } + + private async readSection(stream: ReadableStream, start: number, end: number): Promise { + const reader = stream.getReader(); + let bytesRead = 0; + const chunks: Uint8Array[] = []; + + while (bytesRead < end - start + 1) { + const { done, value } = await reader.read(); + if (done) break; + chunks.push(value); + bytesRead += value.length; + } + + const result = new Uint8Array(bytesRead); + let offset = 0; + for (const chunk of chunks) { + result.set(chunk, offset); + offset += chunk.length; + } + + return result.buffer; + } + + private createPutRequestInformation(content: ArrayBuffer): RequestInformation { + const header = new Headers(); + header.set("Content-Range", new Set([`bytes ${this.rangeBegin}-${this.rangeEnd - 1}/${this.totalSessionLength}`])); + header.set("Content-Length", new Set([`${this.rangeEnd - this.rangeBegin}`])); + + const request = new RequestInformation(); + request.headers = header; + request.urlTemplate = this.sessionUrl; + request.httpMethod = HttpMethod.PUT; + request.setStreamContent(content, binaryContentType); + return request; + } +} diff --git a/src/task/index.ts b/src/task/index.ts new file mode 100644 index 0000000..d370085 --- /dev/null +++ b/src/task/index.ts @@ -0,0 +1 @@ +export * from "./LargeFileUploadTask"; diff --git a/test/task/LargeFileUploadTask.ts b/test/task/LargeFileUploadTask.ts new file mode 100644 index 0000000..e69de29 From b374a1cdee1b4103e88e3ae7e8c6686dd1017697 Mon Sep 17 00:00:00 2001 From: rkodev <43806892+rkodev@users.noreply.github.com> Date: Fri, 14 Feb 2025 05:59:59 +0300 Subject: [PATCH 02/20] move to tasks folder --- src/index.ts | 2 +- src/{task => tasks}/IProgress.ts | 0 src/{task => tasks}/LargeFileUploadTask.ts | 0 src/{task => tasks}/UploadResponseHandler.ts | 0 src/{task => tasks}/UploadResponseHandlerOption.ts | 0 src/{task => tasks}/UploadResult.ts | 0 src/{task => tasks}/UploadSession.ts | 0 src/{task => tasks}/UploadSliceRequestBuilder.ts | 0 src/{task => tasks}/index.ts | 0 9 files changed, 1 insertion(+), 1 deletion(-) rename src/{task => tasks}/IProgress.ts (100%) rename src/{task => tasks}/LargeFileUploadTask.ts (100%) rename src/{task => tasks}/UploadResponseHandler.ts (100%) rename src/{task => tasks}/UploadResponseHandlerOption.ts (100%) rename src/{task => tasks}/UploadResult.ts (100%) rename src/{task => tasks}/UploadSession.ts (100%) rename src/{task => tasks}/UploadSliceRequestBuilder.ts (100%) rename src/{task => tasks}/index.ts (100%) diff --git a/src/index.ts b/src/index.ts index c44a93b..ec5f472 100644 --- a/src/index.ts +++ b/src/index.ts @@ -2,6 +2,6 @@ export * from "./adapter/index.js"; export * from "./http/index.js"; export * from "./middleware/index.js"; export * from "./authentication/index.js"; -export * from "./task/index.js"; +export * from "./tasks/index.js"; export * from "./utils/Constants.js"; export * from "./utils/Version.js"; diff --git a/src/task/IProgress.ts b/src/tasks/IProgress.ts similarity index 100% rename from src/task/IProgress.ts rename to src/tasks/IProgress.ts diff --git a/src/task/LargeFileUploadTask.ts b/src/tasks/LargeFileUploadTask.ts similarity index 100% rename from src/task/LargeFileUploadTask.ts rename to src/tasks/LargeFileUploadTask.ts diff --git a/src/task/UploadResponseHandler.ts b/src/tasks/UploadResponseHandler.ts similarity index 100% rename from src/task/UploadResponseHandler.ts rename to src/tasks/UploadResponseHandler.ts diff --git a/src/task/UploadResponseHandlerOption.ts b/src/tasks/UploadResponseHandlerOption.ts similarity index 100% rename from src/task/UploadResponseHandlerOption.ts rename to src/tasks/UploadResponseHandlerOption.ts diff --git a/src/task/UploadResult.ts b/src/tasks/UploadResult.ts similarity index 100% rename from src/task/UploadResult.ts rename to src/tasks/UploadResult.ts diff --git a/src/task/UploadSession.ts b/src/tasks/UploadSession.ts similarity index 100% rename from src/task/UploadSession.ts rename to src/tasks/UploadSession.ts diff --git a/src/task/UploadSliceRequestBuilder.ts b/src/tasks/UploadSliceRequestBuilder.ts similarity index 100% rename from src/task/UploadSliceRequestBuilder.ts rename to src/tasks/UploadSliceRequestBuilder.ts diff --git a/src/task/index.ts b/src/tasks/index.ts similarity index 100% rename from src/task/index.ts rename to src/tasks/index.ts From ef27b868d9d190d327336b2d92567e11358eb834 Mon Sep 17 00:00:00 2001 From: rkodev <43806892+rkodev@users.noreply.github.com> Date: Fri, 14 Feb 2025 06:20:00 +0300 Subject: [PATCH 03/20] minor code cleanup --- src/tasks/IProgress.ts | 3 - src/tasks/LargeFileUploadTask.ts | 94 +++++++++++++++++++----- src/tasks/UploadResponseHandler.ts | 20 ----- src/tasks/UploadResponseHandlerOption.ts | 6 -- src/tasks/UploadResult.ts | 5 +- src/tasks/UploadSession.ts | 6 -- src/tasks/UploadSliceRequestBuilder.ts | 5 +- 7 files changed, 77 insertions(+), 62 deletions(-) delete mode 100644 src/tasks/IProgress.ts delete mode 100644 src/tasks/UploadResponseHandler.ts delete mode 100644 src/tasks/UploadResponseHandlerOption.ts delete mode 100644 src/tasks/UploadSession.ts diff --git a/src/tasks/IProgress.ts b/src/tasks/IProgress.ts deleted file mode 100644 index 0f9591d..0000000 --- a/src/tasks/IProgress.ts +++ /dev/null @@ -1,3 +0,0 @@ -export interface IProgress { - report(progress: number): void; -} diff --git a/src/tasks/LargeFileUploadTask.ts b/src/tasks/LargeFileUploadTask.ts index 4fd040a..a1014ac 100644 --- a/src/tasks/LargeFileUploadTask.ts +++ b/src/tasks/LargeFileUploadTask.ts @@ -1,27 +1,63 @@ +/** + * ------------------------------------------------------------------------------------------- + * Copyright (c) Microsoft Corporation. All Rights Reserved. Licensed under the MIT License. + * See License in the project root for license information. + * ------------------------------------------------------------------------------------------- + */ + +/** + * @module LargeFileUploadTask + **/ + import { Parsable, RequestAdapter } from "@microsoft/kiota-abstractions"; -import { UploadSession } from "./UploadSession"; import { UploadSliceRequestBuilder } from "./UploadSliceRequestBuilder"; import { UploadResult } from "./UploadResult"; -import { IProgress } from "./IProgress"; - -export interface ILargeFileUploadTask { - Upload(progressEventHandler: IProgress): Promise>; - - Resume(progressEventHandler: IProgress): Promise>; - RefreshUploadStatus(): Promise; - - UpdateSession(): Promise; - - DeleteSession(): Promise; +/** + * @interface + * Signature to represent progress receiver + * @property {number} progress - The progress value + */ +export interface IProgress { + report(progress: number): void; +} - Cancel(): Promise; +/** + * @interface + * Signature to represent an upload session, i.e the response returned by the server after for a pending upload + * + * @property {Date} expirationDateTime - The expiration time of the session + * @property {string[]} nextExpectedRanges - The next expected ranges + * @property {string} odataType - The type of the object + * @property {string} uploadUrl - The URL to which the file upload needs to be done + */ +export interface UploadSession { + expirationDateTime?: Date | null; + nextExpectedRanges?: string[] | null; + odataType?: string | null; + uploadUrl?: string | null; } +/** + * @constant + * A default slice size for a large file + */ const DefaultSliceSize = 320 * 1024; -export class LargeFileUploadTask implements ILargeFileUploadTask { +/** + * A class representing LargeFileUploadTask + */ +export class LargeFileUploadTask { + /** + * @private + * The ranges to upload + */ rangesRemaining: number[][] = []; + + /** + * @private + * The upload session + */ Session: UploadSession; constructor( @@ -41,13 +77,20 @@ export class LargeFileUploadTask implements ILargeFileUpload } this.Session = this.extractSessionInfo(uploadSession); - this.rangesRemaining = this.GetRangesRemaining(this.Session); + this.rangesRemaining = this.getRangesRemaining(this.Session); } - public async Upload(progress?: IProgress, maxTries = 3): Promise> { + /** + * @public + * Uploads file in a sequential order by slicing the file in terms of ranges + * @param progress + * @param maxTries + * @constructor + */ + public async upload(progress?: IProgress, maxTries = 3): Promise> { let uploadTries = 0; while (uploadTries < maxTries) { - const sliceRequests = this.GetUploadSliceRequests(); + const sliceRequests = this.getUploadSliceRequests(); for (const request of sliceRequests) { const uploadResult = await request.UploadSlice(this.uploadStream); progress?.report(request.rangeEnd); @@ -68,7 +111,13 @@ export class LargeFileUploadTask implements ILargeFileUpload throw new Error("Max retries reached"); } - public Resume(_?: IProgress): Promise> { + /** + * @public + * Resumes the current upload session + * @param _ + * @constructor + */ + public resume(_?: IProgress): Promise> { throw new Error("Method not implemented."); } @@ -109,7 +158,7 @@ export class LargeFileUploadTask implements ILargeFileUpload return new Promise(resolve => setTimeout(resolve, ms)); } - private GetUploadSliceRequests(): UploadSliceRequestBuilder[] { + private getUploadSliceRequests(): UploadSliceRequestBuilder[] { const uploadSlices: UploadSliceRequestBuilder[] = []; const rangesRemaining = this.rangesRemaining; const session = this.Session; @@ -136,7 +185,12 @@ export class LargeFileUploadTask implements ILargeFileUpload return sizeBasedOnRange > this.maxSliceSize ? this.maxSliceSize : sizeBasedOnRange; } - private GetRangesRemaining(uploadSession: UploadSession): number[][] { + /** + * @private + * Parses the upload session response and returns a nested number array of ranges pending upload + * @param uploadSession + */ + private getRangesRemaining(uploadSession: UploadSession): number[][] { // nextExpectedRanges: https://dev.onedrive.com/items/upload_large_files.htm // Sample: ["12345-55232","77829-99375"] // Also, second number in range can be blank, which means 'until the end' diff --git a/src/tasks/UploadResponseHandler.ts b/src/tasks/UploadResponseHandler.ts deleted file mode 100644 index ee68bea..0000000 --- a/src/tasks/UploadResponseHandler.ts +++ /dev/null @@ -1,20 +0,0 @@ -import { ErrorMappings, ResponseHandler } from "@microsoft/kiota-abstractions"; - -export class UploadResponseHandler implements ResponseHandler { - handleResponse( - response: NativeResponseType, - _: ErrorMappings | undefined, - ): Promise { - if (response instanceof Response) { - if (response.ok) { - if (response.body != null) { - const body = response.body as unknown as ModelType; - return Promise.resolve(body); - } else { - return Promise.resolve(undefined); - } - } - } - return Promise.resolve(undefined); - } -} diff --git a/src/tasks/UploadResponseHandlerOption.ts b/src/tasks/UploadResponseHandlerOption.ts deleted file mode 100644 index 28be098..0000000 --- a/src/tasks/UploadResponseHandlerOption.ts +++ /dev/null @@ -1,6 +0,0 @@ -import { ResponseHandler, ResponseHandlerOption } from "@microsoft/kiota-abstractions"; -import { UploadResponseHandler } from "./UploadResponseHandler"; - -export class UploadResponseHandlerOption extends ResponseHandlerOption { - public responseHandler?: ResponseHandler = new UploadResponseHandler(); -} diff --git a/src/tasks/UploadResult.ts b/src/tasks/UploadResult.ts index 5ae2239..8f5b345 100644 --- a/src/tasks/UploadResult.ts +++ b/src/tasks/UploadResult.ts @@ -1,10 +1,9 @@ import { Parsable, ParseNode } from "@microsoft/kiota-abstractions"; -import { UploadSession } from "./UploadSession"; -import { ILargeFileUploadTask } from "./LargeFileUploadTask"; +import { LargeFileUploadTask, UploadSession } from "./LargeFileUploadTask"; export class UploadResult { uploadSession?: UploadSession; - uploadTask?: ILargeFileUploadTask; + uploadTask?: LargeFileUploadTask; itemResponse?: T; location?: string; diff --git a/src/tasks/UploadSession.ts b/src/tasks/UploadSession.ts deleted file mode 100644 index 90318a8..0000000 --- a/src/tasks/UploadSession.ts +++ /dev/null @@ -1,6 +0,0 @@ -export interface UploadSession { - expirationDateTime?: Date | null; - nextExpectedRanges?: string[] | null; - odataType?: string | null; - uploadUrl?: string | null; -} diff --git a/src/tasks/UploadSliceRequestBuilder.ts b/src/tasks/UploadSliceRequestBuilder.ts index dc13159..d832415 100644 --- a/src/tasks/UploadSliceRequestBuilder.ts +++ b/src/tasks/UploadSliceRequestBuilder.ts @@ -3,7 +3,6 @@ import { createUploadResult, UploadResult } from "./UploadResult"; import { Headers } from "@microsoft/kiota-abstractions/dist/es/src/headers"; import { HttpMethod } from "@microsoft/kiota-abstractions/dist/es/src/httpMethod"; import { HeadersInspectionOptions } from "@microsoft/kiota-http-fetchlibrary"; -import { UploadResponseHandlerOption } from "./UploadResponseHandlerOption"; const binaryContentType = "application/octet-stream"; @@ -20,10 +19,8 @@ export class UploadSliceRequestBuilder { const data = await this.readSection(stream, this.rangeBegin, this.rangeEnd); const requestInformation = this.createPutRequestInformation(data); - const responseHandler = new UploadResponseHandlerOption(); - const headerOptions = new HeadersInspectionOptions({ inspectResponseHeaders: true }); - requestInformation.addRequestOptions([headerOptions, responseHandler]); + requestInformation.addRequestOptions([headerOptions]); return this.requestAdapter.send>(requestInformation, createUploadResult, undefined); } From 1171fb0df0f9f0e002e33c8b1f9193dc0a0b67c6 Mon Sep 17 00:00:00 2001 From: rkodev <43806892+rkodev@users.noreply.github.com> Date: Thu, 20 Feb 2025 11:03:11 +0300 Subject: [PATCH 04/20] upload file --- src/content/GraphError.ts | 51 ++++++++ src/content/index.ts | 1 + src/tasks/LargeFileResponseHandler.ts | 21 ++++ src/tasks/LargeFileUploadTask.ts | 162 +++++++++++++++++++------ src/tasks/UploadResult.ts | 40 ------ src/tasks/UploadSlice.ts | 94 ++++++++++++++ src/tasks/UploadSliceRequestBuilder.ts | 62 ---------- 7 files changed, 295 insertions(+), 136 deletions(-) create mode 100644 src/content/GraphError.ts create mode 100644 src/content/index.ts create mode 100644 src/tasks/LargeFileResponseHandler.ts delete mode 100644 src/tasks/UploadResult.ts create mode 100644 src/tasks/UploadSlice.ts delete mode 100644 src/tasks/UploadSliceRequestBuilder.ts diff --git a/src/content/GraphError.ts b/src/content/GraphError.ts new file mode 100644 index 0000000..78c6d50 --- /dev/null +++ b/src/content/GraphError.ts @@ -0,0 +1,51 @@ +/** + * ------------------------------------------------------------------------------------------- + * Copyright (c) Microsoft Corporation. All Rights Reserved. Licensed under the MIT License. + * See License in the project root for license information. + * ------------------------------------------------------------------------------------------- + */ + +/** + * @module GraphError + */ + +import { Parsable, ParseNode, AdditionalDataHolder, BackedModel, ApiError } from "@microsoft/kiota-abstractions"; + +/** + * @interface + * Signature represents the structure of an error response + */ +export interface GraphError extends ApiError, AdditionalDataHolder, BackedModel { + /** + * Stores additional data not described in the OpenAPI description found when deserializing. Can be used for serialization as well. + */ + additionalData?: Record; + /** + * Stores model information. + */ + backingStoreEnabled?: boolean | null; +} + +/** + * Creates a new instance of the appropriate class based on discriminator value + * @param _parseNode + */ +export const createGraphErrorFromDiscriminatorValue = ( + _parseNode: ParseNode | undefined, +): ((instance?: Parsable) => Record void>) => { + return deserializeIntoGraphError; +}; + +/** + * Deserializes the batch item + * @param graphError + */ +export const deserializeIntoGraphError = ( + graphError: Partial | undefined = {}, +): Record void> => { + return { + backingStoreEnabled: _n => { + graphError.backingStoreEnabled = true; + }, + }; +}; diff --git a/src/content/index.ts b/src/content/index.ts new file mode 100644 index 0000000..d2f8711 --- /dev/null +++ b/src/content/index.ts @@ -0,0 +1 @@ +export * from "./GraphError.js"; diff --git a/src/tasks/LargeFileResponseHandler.ts b/src/tasks/LargeFileResponseHandler.ts new file mode 100644 index 0000000..7194102 --- /dev/null +++ b/src/tasks/LargeFileResponseHandler.ts @@ -0,0 +1,21 @@ +/** + * ------------------------------------------------------------------------------------------- + * Copyright (c) Microsoft Corporation. All Rights Reserved. Licensed under the MIT License. + * See License in the project root for license information. + * ------------------------------------------------------------------------------------------- + */ + +import { ErrorMappings, ResponseHandler } from "@microsoft/kiota-abstractions"; + +/** + * @class + * Class for LargeFileResponseHandler + */ +export class LargeFileResponseHandler implements ResponseHandler { + handleResponse( + _response: NativeResponseType, + _errorMappings: ErrorMappings | undefined, + ): Promise { + return Promise.resolve(undefined); + } +} diff --git a/src/tasks/LargeFileUploadTask.ts b/src/tasks/LargeFileUploadTask.ts index a1014ac..a22e20a 100644 --- a/src/tasks/LargeFileUploadTask.ts +++ b/src/tasks/LargeFileUploadTask.ts @@ -9,9 +9,17 @@ * @module LargeFileUploadTask **/ -import { Parsable, RequestAdapter } from "@microsoft/kiota-abstractions"; -import { UploadSliceRequestBuilder } from "./UploadSliceRequestBuilder"; -import { UploadResult } from "./UploadResult"; +import { + Parsable, + RequestAdapter, + RequestInformation, + ParsableFactory, + ParseNode, + ErrorMappings, +} from "@microsoft/kiota-abstractions"; +import { UploadSlice } from "./UploadSlice"; +import { HttpMethod } from "@microsoft/kiota-abstractions/dist/es/src/httpMethod"; +import { createGraphErrorFromDiscriminatorValue } from "../content"; /** * @interface @@ -38,6 +46,51 @@ export interface UploadSession { uploadUrl?: string | null; } +/** + * @interface + * Signature to represent the result of an upload + */ +export interface UploadResult { + itemResponse?: T | null; + uploadSession?: UploadSession | null; + location?: string; +} + +/** + * @interface + * Signature to represent the upload session response + */ +export interface UploadSessionResponse extends Parsable { + expirationDateTime?: Date | null; + nextExpectedRanges?: string[] | null; +} +/** + * BatchResponseCollection ParsableFactory + * @param _parseNode + */ +export const createUploadSessionResponseFromDiscriminatorValue = ( + _parseNode: ParseNode | undefined, +): ((instance?: Parsable) => Record void>) => { + return deserializeIntoUploadSessionResponse; +}; + +/** + * Deserializes the batch response body + * @param uploadSessionResponse + */ +export const deserializeIntoUploadSessionResponse = ( + uploadSessionResponse: Partial | undefined = {}, +): Record void> => { + return { + expirationDateTime: n => { + uploadSessionResponse.expirationDateTime = n.getDateValue(); + }, + nextExpectedRanges: n => { + uploadSessionResponse.nextExpectedRanges = n.getCollectionOfPrimitiveValues(); + }, + }; +}; + /** * @constant * A default slice size for a large file @@ -54,6 +107,12 @@ export class LargeFileUploadTask { */ rangesRemaining: number[][] = []; + /** + * @private + * The error mappings + */ + errorMappings: ErrorMappings; + /** * @private * The upload session @@ -65,19 +124,42 @@ export class LargeFileUploadTask { readonly uploadStream: ReadableStream, readonly maxSliceSize = -1, readonly requestAdapter: RequestAdapter, + readonly parsableFactory: ParsableFactory, + errorMappings?: ErrorMappings, ) { + if (!uploadSession) { + const error = new Error("Upload session is undefined, Please provide a valid upload session"); + error.name = "Invalid Upload Session Error"; + throw error; + } + if (!uploadStream) { + const error = new Error("Upload stream is undefined, Please provide a valid upload stream"); + error.name = "Invalid Upload Stream Error"; + throw error; + } + if (!requestAdapter) { + const error = new Error("Request adapter is undefined, Please provide a valid request adapter"); + error.name = "Invalid Request Adapter Error"; + throw error; + } + if (!parsableFactory) { + const error = new Error("Parsable factory is undefined, Please provide a valid parsable factory"); + error.name = "Invalid Parsable Factory Error"; + throw error; + } if (!uploadStream?.locked) { throw new Error("Please provide stream value"); } - if (requestAdapter === undefined) { - throw new Error("Request adapter is a required parameter"); - } if (maxSliceSize <= 0) { this.maxSliceSize = DefaultSliceSize; } + this.parsableFactory = parsableFactory; this.Session = this.extractSessionInfo(uploadSession); this.rangesRemaining = this.getRangesRemaining(this.Session); + this.errorMappings = errorMappings || { + XXX: parseNode => createGraphErrorFromDiscriminatorValue(parseNode), + }; } /** @@ -87,19 +169,26 @@ export class LargeFileUploadTask { * @param maxTries * @constructor */ - public async upload(progress?: IProgress, maxTries = 3): Promise> { + public async upload(progress?: IProgress): Promise> { + const sliceRequests = this.getUploadSliceRequests(); + for (const request of sliceRequests) { + const uploadResult = await request.uploadSlice(this.uploadStream); + progress?.report(request.rangeEnd); + if (uploadResult?.itemResponse || uploadResult?.location) { + return uploadResult; + } + } + throw new Error("Upload failed"); + } + + private async uploadWithRetry(uploadSlice: UploadSlice, maxTries = 3): Promise | undefined> { let uploadTries = 0; while (uploadTries < maxTries) { - const sliceRequests = this.getUploadSliceRequests(); - for (const request of sliceRequests) { - const uploadResult = await request.UploadSlice(this.uploadStream); - progress?.report(request.rangeEnd); - if (uploadResult?.UploadSucceeded()) { - return uploadResult; - } + try { + return await uploadSlice.uploadSlice(this.uploadStream); + } catch (e) { + console.error(e); } - - await this.UpdateSession(); uploadTries++; if (uploadTries < maxTries) { @@ -114,27 +203,31 @@ export class LargeFileUploadTask { /** * @public * Resumes the current upload session - * @param _ - * @constructor + * @param progress */ - public resume(_?: IProgress): Promise> { - throw new Error("Method not implemented."); + public async resume(progress?: IProgress): Promise> { + await this.refreshUploadStatus(); + return this.upload(progress); } - public RefreshUploadStatus(): Promise { - throw new Error("Method not implemented."); - } - - public UpdateSession(): Promise { - throw new Error("Method not implemented."); - } + public async refreshUploadStatus() { + const requestInformation = new RequestInformation(HttpMethod.GET, this.Session.uploadUrl!); + const response = await this.requestAdapter.send( + requestInformation, + createUploadSessionResponseFromDiscriminatorValue, + this.errorMappings, + ); - public DeleteSession(): Promise { - throw new Error("Method not implemented."); + if (response) { + this.Session.expirationDateTime = response?.expirationDateTime; + this.Session.nextExpectedRanges = response.nextExpectedRanges; + this.rangesRemaining = this.getRangesRemaining(this.Session); + } } - public Cancel(): Promise { - throw new Error("Method not implemented."); + public async cancel(): Promise { + const requestInformation = new RequestInformation(HttpMethod.PUT, this.Session.uploadUrl!); + await this.requestAdapter.sendNoResponseContent(requestInformation, this.errorMappings); } private extractSessionInfo(parsable: Parsable): UploadSession { @@ -158,20 +251,21 @@ export class LargeFileUploadTask { return new Promise(resolve => setTimeout(resolve, ms)); } - private getUploadSliceRequests(): UploadSliceRequestBuilder[] { - const uploadSlices: UploadSliceRequestBuilder[] = []; + private getUploadSliceRequests(): UploadSlice[] { + const uploadSlices: UploadSlice[] = []; const rangesRemaining = this.rangesRemaining; const session = this.Session; rangesRemaining.forEach(range => { let currentRangeBegin = range[0]; while (currentRangeBegin <= range[1]) { const nextSliceSize = this.nextSliceSize(currentRangeBegin, range[1]); - const uploadRequest = new UploadSliceRequestBuilder( + const uploadRequest = new UploadSlice( this.requestAdapter, session.uploadUrl!, currentRangeBegin, currentRangeBegin + nextSliceSize - 1, range[1] + 1, + this.parsableFactory, ); uploadSlices.push(uploadRequest); currentRangeBegin += nextSliceSize; diff --git a/src/tasks/UploadResult.ts b/src/tasks/UploadResult.ts deleted file mode 100644 index 8f5b345..0000000 --- a/src/tasks/UploadResult.ts +++ /dev/null @@ -1,40 +0,0 @@ -import { Parsable, ParseNode } from "@microsoft/kiota-abstractions"; -import { LargeFileUploadTask, UploadSession } from "./LargeFileUploadTask"; - -export class UploadResult { - uploadSession?: UploadSession; - uploadTask?: LargeFileUploadTask; - itemResponse?: T; - location?: string; - - UploadSucceeded() { - return this.itemResponse !== undefined || this.location !== undefined; - } -} - -// eslint-disable-next-line prefer-arrow/prefer-arrow-functions -export function createUploadResult( - _: ParseNode | undefined, -): (instance?: Parsable) => Record void> { - return deserializeIntoUploadResult; -} - -// eslint-disable-next-line prefer-arrow/prefer-arrow-functions -export function deserializeIntoUploadResult( - uploadResult: Partial> | undefined = {}, -): Record void> { - return { - uploadSession: _ => { - uploadResult.uploadSession = undefined; - }, - uploadTask: _ => { - uploadResult.uploadSession = undefined; - }, - itemResponse: _ => { - uploadResult.uploadSession = undefined; - }, - location: _ => { - uploadResult.uploadSession = undefined; - }, - }; -} diff --git a/src/tasks/UploadSlice.ts b/src/tasks/UploadSlice.ts new file mode 100644 index 0000000..d01d58c --- /dev/null +++ b/src/tasks/UploadSlice.ts @@ -0,0 +1,94 @@ +/** + * ------------------------------------------------------------------------------------------- + * Copyright (c) Microsoft Corporation. All Rights Reserved. Licensed under the MIT License. + * See License in the project root for license information. + * ------------------------------------------------------------------------------------------- + */ + +import { + ErrorMappings, + Headers, + HttpMethod, + Parsable, + ParsableFactory, + RequestAdapter, + RequestInformation, +} from "@microsoft/kiota-abstractions"; +import { HeadersInspectionOptions } from "@microsoft/kiota-http-fetchlibrary"; +import { createGraphErrorFromDiscriminatorValue } from "../content"; +import { UploadResult, UploadSession } from "./LargeFileUploadTask"; + +const binaryContentType = "application/octet-stream"; + +/** + * @class + * Class for UploadSlice + */ +export class UploadSlice { + constructor( + readonly requestAdapter: RequestAdapter, + readonly sessionUrl: string, + readonly rangeBegin: number, + readonly rangeEnd: number, + readonly totalSessionLength: number, + readonly parsableFactory: ParsableFactory, + ) {} + + public async uploadSlice(stream: ReadableStream): Promise | undefined> { + const data = await this.readSection(stream, this.rangeBegin, this.rangeEnd); + const requestInformation = new RequestInformation(HttpMethod.PUT, this.sessionUrl); + requestInformation.headers = new Headers([ + ["Content-Range", new Set([`bytes ${this.rangeBegin}-${this.rangeEnd - 1}/${this.totalSessionLength}`])], + ["Content-Length", new Set([`${this.rangeEnd - this.rangeBegin}`])], + ]); + requestInformation.setStreamContent(data, binaryContentType); + + const headerOptions = new HeadersInspectionOptions({ inspectResponseHeaders: true }); + requestInformation.addRequestOptions([headerOptions]); + + const errorMappings: ErrorMappings = { + XXX: parseNode => createGraphErrorFromDiscriminatorValue(parseNode), + }; + + const itemResponse = await this.requestAdapter.send(requestInformation, this.parsableFactory, errorMappings); + + const locations = headerOptions.getResponseHeaders().get("location"); + + let uploadSession: UploadSession | null = null; + if (itemResponse && ("expirationDateTime" in itemResponse || "additionalData" in itemResponse)) { + uploadSession = {}; + if ("expirationDateTime" in itemResponse) + uploadSession.expirationDateTime = itemResponse.expirationDateTime as Date | null; + if ("nextExpectedRanges" in itemResponse) + uploadSession.nextExpectedRanges = itemResponse.nextExpectedRanges as string[] | null; + } + + return { + itemResponse, + location: locations ? (locations as unknown as string[])[0] : undefined, + uploadSession, + }; + } + + private async readSection(stream: ReadableStream, start: number, end: number): Promise { + const reader = stream.getReader(); + let bytesRead = 0; + const chunks: Uint8Array[] = []; + + while (bytesRead < end - start + 1) { + const { done, value } = await reader.read(); + if (done) break; + chunks.push(value); + bytesRead += value.length; + } + + const result = new Uint8Array(bytesRead); + let offset = 0; + for (const chunk of chunks) { + result.set(chunk, offset); + offset += chunk.length; + } + + return result.buffer; + } +} diff --git a/src/tasks/UploadSliceRequestBuilder.ts b/src/tasks/UploadSliceRequestBuilder.ts deleted file mode 100644 index d832415..0000000 --- a/src/tasks/UploadSliceRequestBuilder.ts +++ /dev/null @@ -1,62 +0,0 @@ -import { Parsable, RequestAdapter, RequestInformation } from "@microsoft/kiota-abstractions"; -import { createUploadResult, UploadResult } from "./UploadResult"; -import { Headers } from "@microsoft/kiota-abstractions/dist/es/src/headers"; -import { HttpMethod } from "@microsoft/kiota-abstractions/dist/es/src/httpMethod"; -import { HeadersInspectionOptions } from "@microsoft/kiota-http-fetchlibrary"; - -const binaryContentType = "application/octet-stream"; - -export class UploadSliceRequestBuilder { - constructor( - readonly requestAdapter: RequestAdapter, - readonly sessionUrl: string, - readonly rangeBegin: number, - readonly rangeEnd: number, - readonly totalSessionLength: number, - ) {} - - public async UploadSlice(stream: ReadableStream): Promise | undefined> { - const data = await this.readSection(stream, this.rangeBegin, this.rangeEnd); - const requestInformation = this.createPutRequestInformation(data); - - const headerOptions = new HeadersInspectionOptions({ inspectResponseHeaders: true }); - requestInformation.addRequestOptions([headerOptions]); - - return this.requestAdapter.send>(requestInformation, createUploadResult, undefined); - } - - private async readSection(stream: ReadableStream, start: number, end: number): Promise { - const reader = stream.getReader(); - let bytesRead = 0; - const chunks: Uint8Array[] = []; - - while (bytesRead < end - start + 1) { - const { done, value } = await reader.read(); - if (done) break; - chunks.push(value); - bytesRead += value.length; - } - - const result = new Uint8Array(bytesRead); - let offset = 0; - for (const chunk of chunks) { - result.set(chunk, offset); - offset += chunk.length; - } - - return result.buffer; - } - - private createPutRequestInformation(content: ArrayBuffer): RequestInformation { - const header = new Headers(); - header.set("Content-Range", new Set([`bytes ${this.rangeBegin}-${this.rangeEnd - 1}/${this.totalSessionLength}`])); - header.set("Content-Length", new Set([`${this.rangeEnd - this.rangeBegin}`])); - - const request = new RequestInformation(); - request.headers = header; - request.urlTemplate = this.sessionUrl; - request.httpMethod = HttpMethod.PUT; - request.setStreamContent(content, binaryContentType); - return request; - } -} From d41b0cc3ec2f2c23977627101a8b1cdaa2eb69a5 Mon Sep 17 00:00:00 2001 From: rkodev <43806892+rkodev@users.noreply.github.com> Date: Thu, 20 Feb 2025 18:08:18 +0300 Subject: [PATCH 05/20] Adds basic tests --- src/content/GraphError.ts | 51 ------------ src/content/index.ts | 1 - src/tasks/LargeFileUploadTask.ts | 101 ++++++++++++++++++------ src/tasks/UploadSlice.ts | 12 +-- test/task/LargeFileUploadTask.ts | 93 ++++++++++++++++++++++ test/utils/DummyRequestAdapter.ts | 126 ++++++++++++++++++++++++++++++ vite.config.ts | 2 +- 7 files changed, 302 insertions(+), 84 deletions(-) delete mode 100644 src/content/GraphError.ts delete mode 100644 src/content/index.ts create mode 100644 test/utils/DummyRequestAdapter.ts diff --git a/src/content/GraphError.ts b/src/content/GraphError.ts deleted file mode 100644 index 78c6d50..0000000 --- a/src/content/GraphError.ts +++ /dev/null @@ -1,51 +0,0 @@ -/** - * ------------------------------------------------------------------------------------------- - * Copyright (c) Microsoft Corporation. All Rights Reserved. Licensed under the MIT License. - * See License in the project root for license information. - * ------------------------------------------------------------------------------------------- - */ - -/** - * @module GraphError - */ - -import { Parsable, ParseNode, AdditionalDataHolder, BackedModel, ApiError } from "@microsoft/kiota-abstractions"; - -/** - * @interface - * Signature represents the structure of an error response - */ -export interface GraphError extends ApiError, AdditionalDataHolder, BackedModel { - /** - * Stores additional data not described in the OpenAPI description found when deserializing. Can be used for serialization as well. - */ - additionalData?: Record; - /** - * Stores model information. - */ - backingStoreEnabled?: boolean | null; -} - -/** - * Creates a new instance of the appropriate class based on discriminator value - * @param _parseNode - */ -export const createGraphErrorFromDiscriminatorValue = ( - _parseNode: ParseNode | undefined, -): ((instance?: Parsable) => Record void>) => { - return deserializeIntoGraphError; -}; - -/** - * Deserializes the batch item - * @param graphError - */ -export const deserializeIntoGraphError = ( - graphError: Partial | undefined = {}, -): Record void> => { - return { - backingStoreEnabled: _n => { - graphError.backingStoreEnabled = true; - }, - }; -}; diff --git a/src/content/index.ts b/src/content/index.ts deleted file mode 100644 index d2f8711..0000000 --- a/src/content/index.ts +++ /dev/null @@ -1 +0,0 @@ -export * from "./GraphError.js"; diff --git a/src/tasks/LargeFileUploadTask.ts b/src/tasks/LargeFileUploadTask.ts index a22e20a..4964dde 100644 --- a/src/tasks/LargeFileUploadTask.ts +++ b/src/tasks/LargeFileUploadTask.ts @@ -19,7 +19,6 @@ import { } from "@microsoft/kiota-abstractions"; import { UploadSlice } from "./UploadSlice"; import { HttpMethod } from "@microsoft/kiota-abstractions/dist/es/src/httpMethod"; -import { createGraphErrorFromDiscriminatorValue } from "../content"; /** * @interface @@ -119,13 +118,25 @@ export class LargeFileUploadTask { */ Session: UploadSession; + /** + * Constructs a new instance of the LargeFileUploadTask class. + * + * @param {RequestAdapter} requestAdapter - The request adapter to use for making HTTP requests. + * @param {Parsable} uploadSession - The upload session information. + * @param {ReadableStream} uploadStream - The stream of the file to be uploaded. + * @param {number} [maxSliceSize=-1] - The maximum size of each file slice to be uploaded. + * @param {ParsableFactory} parsableFactory - The factory to create parsable objects. + * @param {ErrorMappings} [errorMappings] - error mappings. + * + * @throws {Error} If any of the required parameters are undefined or invalid. + */ constructor( + readonly requestAdapter: RequestAdapter, readonly uploadSession: Parsable, readonly uploadStream: ReadableStream, readonly maxSliceSize = -1, - readonly requestAdapter: RequestAdapter, readonly parsableFactory: ParsableFactory, - errorMappings?: ErrorMappings, + errorMappings: ErrorMappings, ) { if (!uploadSession) { const error = new Error("Upload session is undefined, Please provide a valid upload session"); @@ -147,7 +158,7 @@ export class LargeFileUploadTask { error.name = "Invalid Parsable Factory Error"; throw error; } - if (!uploadStream?.locked) { + if (uploadStream?.locked) { throw new Error("Please provide stream value"); } if (maxSliceSize <= 0) { @@ -157,17 +168,15 @@ export class LargeFileUploadTask { this.Session = this.extractSessionInfo(uploadSession); this.rangesRemaining = this.getRangesRemaining(this.Session); - this.errorMappings = errorMappings || { - XXX: parseNode => createGraphErrorFromDiscriminatorValue(parseNode), - }; + this.errorMappings = errorMappings; } /** - * @public - * Uploads file in a sequential order by slicing the file in terms of ranges - * @param progress - * @param maxTries - * @constructor + * Uploads the file in a sequential order by slicing the file in terms of ranges. + * + * @param {IProgress} [progress] - Optional progress receiver to report upload progress. + * @returns {Promise>} - The result of the upload. + * @throws {Error} If the upload fails. */ public async upload(progress?: IProgress): Promise> { const sliceRequests = this.getUploadSliceRequests(); @@ -181,6 +190,14 @@ export class LargeFileUploadTask { throw new Error("Upload failed"); } + /** + * Uploads a slice with retry logic. + * + * @param {UploadSlice} uploadSlice - The upload slice to be uploaded. + * @param {number} [maxTries=3] - The maximum number of retry attempts. + * @returns {Promise | undefined>} - The result of the upload. + * @throws {Error} If the maximum number of retries is reached. + */ private async uploadWithRetry(uploadSlice: UploadSlice, maxTries = 3): Promise | undefined> { let uploadTries = 0; while (uploadTries < maxTries) { @@ -210,6 +227,12 @@ export class LargeFileUploadTask { return this.upload(progress); } + /** + * Refreshes the current upload session status by making a GET request to the upload URL. + * Updates the session expiration date, next expected ranges, and remaining ranges based on the response. + * + * @throws {Error} If the request fails. + */ public async refreshUploadStatus() { const requestInformation = new RequestInformation(HttpMethod.GET, this.Session.uploadUrl!); const response = await this.requestAdapter.send( @@ -225,32 +248,52 @@ export class LargeFileUploadTask { } } + /** + * Cancels the current upload session. + * Sends a PUT request to the upload URL to cancel the session. + * + * @returns {Promise} A promise that resolves when the session is canceled. + */ public async cancel(): Promise { const requestInformation = new RequestInformation(HttpMethod.PUT, this.Session.uploadUrl!); await this.requestAdapter.sendNoResponseContent(requestInformation, this.errorMappings); } + /** + * Extracts the upload session information from a parsable object. + * + * @param {Parsable} parsable - The parsable object containing the upload session information. + * @returns {UploadSession} - The extracted upload session information. + */ private extractSessionInfo(parsable: Parsable): UploadSession { - const uploadSession: UploadSession = { - expirationDateTime: null, - nextExpectedRanges: null, - odataType: null, - uploadUrl: null, + const { expirationDateTime, nextExpectedRanges, odataType, uploadUrl } = parsable as Partial; + if (!nextExpectedRanges || !uploadUrl || nextExpectedRanges.length === 0) { + throw new Error("Upload session is invalid"); + } + return { + expirationDateTime: expirationDateTime ?? null, + nextExpectedRanges: nextExpectedRanges ?? null, + odataType: odataType ?? null, + uploadUrl: uploadUrl ?? null, }; - - if ("expirationDateTime" in parsable) uploadSession.expirationDateTime = parsable.expirationDateTime as Date | null; - if ("nextExpectedRanges" in parsable) - uploadSession.nextExpectedRanges = parsable.nextExpectedRanges as string[] | null; - if ("odataType" in parsable) uploadSession.odataType = parsable.odataType as string | null; - if ("uploadUrl" in parsable) uploadSession.uploadUrl = parsable.uploadUrl as string | null; - - return uploadSession; } + /** + * Pauses execution for a specified number of milliseconds. + * + * @param {number} ms - The number of milliseconds to sleep. + * @returns {Promise} A promise that resolves after the specified time. + */ private sleep(ms: number): Promise { return new Promise(resolve => setTimeout(resolve, ms)); } + /** + * Generates a list of upload slice requests based on the remaining ranges to be uploaded. + * + * @private + * @returns {UploadSlice[]} An array of UploadSlice objects representing the upload requests. + */ private getUploadSliceRequests(): UploadSlice[] { const uploadSlices: UploadSlice[] = []; const rangesRemaining = this.rangesRemaining; @@ -266,6 +309,7 @@ export class LargeFileUploadTask { currentRangeBegin + nextSliceSize - 1, range[1] + 1, this.parsableFactory, + this.errorMappings, ); uploadSlices.push(uploadRequest); currentRangeBegin += nextSliceSize; @@ -274,6 +318,13 @@ export class LargeFileUploadTask { return uploadSlices; } + /** + * Calculates the size of the next slice to be uploaded. + * + * @param {number} currentRangeBegin - The beginning of the current range. + * @param {number} currentRangeEnd - The end of the current range. + * @returns {number} - The size of the next slice. + */ private nextSliceSize(currentRangeBegin: number, currentRangeEnd: number): number { const sizeBasedOnRange = currentRangeEnd - currentRangeBegin + 1; return sizeBasedOnRange > this.maxSliceSize ? this.maxSliceSize : sizeBasedOnRange; diff --git a/src/tasks/UploadSlice.ts b/src/tasks/UploadSlice.ts index d01d58c..1c49116 100644 --- a/src/tasks/UploadSlice.ts +++ b/src/tasks/UploadSlice.ts @@ -15,7 +15,6 @@ import { RequestInformation, } from "@microsoft/kiota-abstractions"; import { HeadersInspectionOptions } from "@microsoft/kiota-http-fetchlibrary"; -import { createGraphErrorFromDiscriminatorValue } from "../content"; import { UploadResult, UploadSession } from "./LargeFileUploadTask"; const binaryContentType = "application/octet-stream"; @@ -32,6 +31,7 @@ export class UploadSlice { readonly rangeEnd: number, readonly totalSessionLength: number, readonly parsableFactory: ParsableFactory, + readonly errorMappings: ErrorMappings, ) {} public async uploadSlice(stream: ReadableStream): Promise | undefined> { @@ -46,11 +46,11 @@ export class UploadSlice { const headerOptions = new HeadersInspectionOptions({ inspectResponseHeaders: true }); requestInformation.addRequestOptions([headerOptions]); - const errorMappings: ErrorMappings = { - XXX: parseNode => createGraphErrorFromDiscriminatorValue(parseNode), - }; - - const itemResponse = await this.requestAdapter.send(requestInformation, this.parsableFactory, errorMappings); + const itemResponse = await this.requestAdapter.send( + requestInformation, + this.parsableFactory, + this.errorMappings, + ); const locations = headerOptions.getResponseHeaders().get("location"); diff --git a/test/task/LargeFileUploadTask.ts b/test/task/LargeFileUploadTask.ts index e69de29..6d9a08e 100644 --- a/test/task/LargeFileUploadTask.ts +++ b/test/task/LargeFileUploadTask.ts @@ -0,0 +1,93 @@ +import { assert, describe, it } from "vitest"; +import { LargeFileUploadTask } from "../../src"; +// @ts-ignore +import { DummyRequestAdapter } from "../utils/DummyRequestAdapter"; +import { ErrorMappings, Parsable, ParseNode } from "@microsoft/kiota-abstractions"; + +const adapter = new DummyRequestAdapter(); + +interface SampleResponse extends Parsable { + nextExpectedRanges?: string[] | undefined; + expirationDateTime?: Date | undefined; + uploadUrl?: string | undefined; +} + +export function createPageCollectionFromDiscriminatorValue( + parseNode: ParseNode | undefined, +): (instance?: Parsable) => Record void> { + return deserializeIntoPageCollection; +} + +export function deserializeIntoPageCollection( + baseDeltaFunctionResponse: Partial | undefined = {}, +): Record void> { + return {}; +} + +const sampleReadableStream = new ReadableStream({ + start(controller) { + const encoder = new TextEncoder(); + const chunk = encoder.encode("This is a 20-byte string"); + controller.enqueue(chunk); + controller.close(); + }, +}); + +const errorMappings: ErrorMappings = { + XXX: parseNode => createGraphErrorFromDiscriminatorValue(parseNode), +}; + +export const createGraphErrorFromDiscriminatorValue = ( + _parseNode: ParseNode | undefined, +): ((instance?: Parsable) => Record void>) => { + return deserializeIntoGraphError; +}; + +/** + * Deserializes the batch item + * @param graphError + */ +export const deserializeIntoGraphError = ( + graphError: Partial | undefined = {}, +): Record void> => { + return {}; +}; + +describe("LargeFileUploadTask tests", () => { + it("should initialize", () => { + const session: SampleResponse = { + nextExpectedRanges: ["0-19"], + expirationDateTime: new Date(Date.now() + 24 * 60 * 60 * 1000), // + uploadUrl: "https://example.com/upload", + }; + const handler = new LargeFileUploadTask( + adapter, + session, + sampleReadableStream, + 10, + createPageCollectionFromDiscriminatorValue, + errorMappings, + ); + assert(handler, "LargeFileUploadTask failed to initialize"); + }); + it("should throw an error if invalid upload session is given", () => { + assert.throws(() => { + const handler = new LargeFileUploadTask( + adapter, + {} as SampleResponse, + sampleReadableStream, + 10, + createPageCollectionFromDiscriminatorValue, + errorMappings, + ); + }, "Upload session is invalid"); + }); + + // TODO test slice generation by range spliting + + // TODO test file upload + + // TODO test canceling an upload + + // TODO test resume an upload +}); diff --git a/test/utils/DummyRequestAdapter.ts b/test/utils/DummyRequestAdapter.ts new file mode 100644 index 0000000..9416357 --- /dev/null +++ b/test/utils/DummyRequestAdapter.ts @@ -0,0 +1,126 @@ +/** + * ------------------------------------------------------------------------------------------- + * Copyright (c) Microsoft Corporation. All Rights Reserved. Licensed under the MIT License. + * See License in the project root for license information. + * ------------------------------------------------------------------------------------------- + */ + +/** + * @module DummyRequestAdapter + */ + +import { + type BackingStoreFactory, + type ErrorMappings, + Parsable, + type ParsableFactory, + type PrimitiveTypesForDeserialization, + type PrimitiveTypesForDeserializationType, + RequestAdapter, + type RequestInformation, + SerializationWriterFactory, + SerializationWriterFactoryRegistry, +} from "@microsoft/kiota-abstractions"; + +import { JsonSerializationWriterFactory } from "@microsoft/kiota-serialization-json"; + +/** + * @class + * @implements DummyRequestAdapter + * Class representing DummyRequestAdapter + */ +export class DummyRequestAdapter implements RequestAdapter { + baseUrl: string = ""; + response: any[] = []; + requests: RequestInformation[] = []; + serializationWriterFactory = new SerializationWriterFactoryRegistry(); + + constructor() { + const serializer = new JsonSerializationWriterFactory(); + this.serializationWriterFactory.contentTypeAssociatedFactories.set(serializer.getValidContentType(), serializer); + } + + // set the url + setBaseUrl(baseUrl: string): void { + this.baseUrl = baseUrl; + } + + // set a fake response + setResponse(response: any): void { + this.response.push(response); + } + + // get requests + getRequests(): RequestInformation[] { + return this.requests; + } + + convertToNativeRequest(requestInfo: RequestInformation): Promise { + return Promise.resolve(undefined as T); + } + + enableBackingStore(backingStoreFactory?: BackingStoreFactory): void {} + + getSerializationWriterFactory(): SerializationWriterFactory { + return this.serializationWriterFactory; + } + + send( + requestInfo: RequestInformation, + type: ParsableFactory, + errorMappings: ErrorMappings | undefined, + ): Promise { + this.requests.push(requestInfo); + return Promise.resolve(this.response.shift()); + } + + sendCollection( + requestInfo: RequestInformation, + type: ParsableFactory, + errorMappings: ErrorMappings | undefined, + ): Promise { + this.requests.push(requestInfo); + return Promise.resolve(this.response.shift()); + } + + sendCollectionOfEnum>( + requestInfo: RequestInformation, + enumObject: EnumObject, + errorMappings: ErrorMappings | undefined, + ): Promise { + this.requests.push(requestInfo); + return Promise.resolve(this.response.shift()); + } + + sendCollectionOfPrimitive>( + requestInfo: RequestInformation, + responseType: Exclude, + errorMappings: ErrorMappings | undefined, + ): Promise { + this.requests.push(requestInfo); + return Promise.resolve(this.response.shift()); + } + + sendEnum>( + requestInfo: RequestInformation, + enumObject: EnumObject, + errorMappings: ErrorMappings | undefined, + ): Promise { + this.requests.push(requestInfo); + return Promise.resolve(this.response.shift()); + } + + sendNoResponseContent(requestInfo: RequestInformation, errorMappings: ErrorMappings | undefined): Promise { + this.requests.push(requestInfo); + return Promise.resolve(this.response.shift()); + } + + sendPrimitive( + requestInfo: RequestInformation, + responseType: PrimitiveTypesForDeserialization, + errorMappings: ErrorMappings | undefined, + ): Promise { + this.requests.push(requestInfo); + return Promise.resolve(this.response.shift()); + } +} diff --git a/vite.config.ts b/vite.config.ts index ac01c56..aa7f21f 100644 --- a/vite.config.ts +++ b/vite.config.ts @@ -2,7 +2,7 @@ import { defineConfig, configDefaults } from "vitest/config"; export default defineConfig({ test: { - exclude: [...configDefaults.exclude, "packages/template/*", "test/middleware/DummyFetchHandler.ts"], + exclude: [...configDefaults.exclude, "packages/template/*", "test/**/Dummy*.ts"], include: [...configDefaults.include, "test/**/*.ts"], coverage: { reporter: ["html"], From 74e75bb75edda5c20e77970924c8411cd093a4e4 Mon Sep 17 00:00:00 2001 From: rkodev <43806892+rkodev@users.noreply.github.com> Date: Fri, 21 Feb 2025 07:18:24 +0300 Subject: [PATCH 06/20] Adds more tests --- src/tasks/LargeFileUploadTask.ts | 22 +-- src/tasks/UploadSlice.ts | 58 +++++--- test/task/LargeFileUploadTask.ts | 215 +++++++++++++++++++++++++----- test/utils/DummyRequestAdapter.ts | 5 + 4 files changed, 245 insertions(+), 55 deletions(-) diff --git a/src/tasks/LargeFileUploadTask.ts b/src/tasks/LargeFileUploadTask.ts index 4964dde..dae7a2c 100644 --- a/src/tasks/LargeFileUploadTask.ts +++ b/src/tasks/LargeFileUploadTask.ts @@ -181,10 +181,14 @@ export class LargeFileUploadTask { public async upload(progress?: IProgress): Promise> { const sliceRequests = this.getUploadSliceRequests(); for (const request of sliceRequests) { - const uploadResult = await request.uploadSlice(this.uploadStream); - progress?.report(request.rangeEnd); - if (uploadResult?.itemResponse || uploadResult?.location) { - return uploadResult; + try { + const uploadResult = await request.uploadSlice(this.uploadStream); + progress?.report(request.rangeEnd); + if (uploadResult?.itemResponse || uploadResult?.location) { + return uploadResult; + } + } catch (e) { + console.error(e); } } throw new Error("Upload failed"); @@ -223,7 +227,7 @@ export class LargeFileUploadTask { * @param progress */ public async resume(progress?: IProgress): Promise> { - await this.refreshUploadStatus(); + await this.updateSession(); return this.upload(progress); } @@ -233,7 +237,7 @@ export class LargeFileUploadTask { * * @throws {Error} If the request fails. */ - public async refreshUploadStatus() { + public async updateSession() { const requestInformation = new RequestInformation(HttpMethod.GET, this.Session.uploadUrl!); const response = await this.requestAdapter.send( requestInformation, @@ -249,13 +253,13 @@ export class LargeFileUploadTask { } /** - * Cancels the current upload session. + * Deletes the current upload session. * Sends a PUT request to the upload URL to cancel the session. * * @returns {Promise} A promise that resolves when the session is canceled. */ - public async cancel(): Promise { - const requestInformation = new RequestInformation(HttpMethod.PUT, this.Session.uploadUrl!); + public async deleteSession(): Promise { + const requestInformation = new RequestInformation(HttpMethod.DELETE, this.Session.uploadUrl!); await this.requestAdapter.sendNoResponseContent(requestInformation, this.errorMappings); } diff --git a/src/tasks/UploadSlice.ts b/src/tasks/UploadSlice.ts index 1c49116..5a2f6a2 100644 --- a/src/tasks/UploadSlice.ts +++ b/src/tasks/UploadSlice.ts @@ -34,6 +34,12 @@ export class UploadSlice { readonly errorMappings: ErrorMappings, ) {} + /** + * Uploads a slice of the file to the server. + * + * @param {ReadableStream} stream - The stream of the file slice to be uploaded. + * @returns {Promise | undefined>} - The result of the upload operation. + */ public async uploadSlice(stream: ReadableStream): Promise | undefined> { const data = await this.readSection(stream, this.rangeBegin, this.rangeEnd); const requestInformation = new RequestInformation(HttpMethod.PUT, this.sessionUrl); @@ -46,21 +52,20 @@ export class UploadSlice { const headerOptions = new HeadersInspectionOptions({ inspectResponseHeaders: true }); requestInformation.addRequestOptions([headerOptions]); - const itemResponse = await this.requestAdapter.send( - requestInformation, - this.parsableFactory, - this.errorMappings, - ); + let itemResponse = await this.requestAdapter.send(requestInformation, this.parsableFactory, this.errorMappings); const locations = headerOptions.getResponseHeaders().get("location"); let uploadSession: UploadSession | null = null; - if (itemResponse && ("expirationDateTime" in itemResponse || "additionalData" in itemResponse)) { - uploadSession = {}; - if ("expirationDateTime" in itemResponse) - uploadSession.expirationDateTime = itemResponse.expirationDateTime as Date | null; - if ("nextExpectedRanges" in itemResponse) - uploadSession.nextExpectedRanges = itemResponse.nextExpectedRanges as string[] | null; + if (itemResponse) { + const { expirationDateTime, nextExpectedRanges } = itemResponse as Partial; + if (nextExpectedRanges) { + uploadSession = { + expirationDateTime, + nextExpectedRanges: (nextExpectedRanges as unknown as string[])[0].split(","), + }; + itemResponse = undefined; + } } return { @@ -70,16 +75,37 @@ export class UploadSlice { }; } + /** + * Reads a section of the stream from the specified start to end positions. + * + * @param {ReadableStream} stream - The stream to read from. + * @param {number} start - The starting byte position. + * @param {number} end - The ending byte position. + * @returns {Promise} - A promise that resolves to an ArrayBuffer containing the read bytes. + */ private async readSection(stream: ReadableStream, start: number, end: number): Promise { const reader = stream.getReader(); let bytesRead = 0; const chunks: Uint8Array[] = []; + const totalBytesToRead = end - start; - while (bytesRead < end - start + 1) { - const { done, value } = await reader.read(); - if (done) break; - chunks.push(value); - bytesRead += value.length; + try { + while (bytesRead < totalBytesToRead) { + const { done, value } = await reader.read(); + if (done) break; + if (value) { + const remainingBytes = totalBytesToRead - bytesRead; + if (value.length > remainingBytes) { + chunks.push(value.slice(0, remainingBytes)); + bytesRead += remainingBytes; + } else { + chunks.push(value); + bytesRead += value.length; + } + } + } + } finally { + reader.releaseLock(); } const result = new Uint8Array(bytesRead); diff --git a/test/task/LargeFileUploadTask.ts b/test/task/LargeFileUploadTask.ts index 6d9a08e..231a8db 100644 --- a/test/task/LargeFileUploadTask.ts +++ b/test/task/LargeFileUploadTask.ts @@ -1,8 +1,9 @@ -import { assert, describe, it } from "vitest"; -import { LargeFileUploadTask } from "../../src"; +import { assert, describe, expect, it } from "vitest"; +import { coreVersion, IProgress, LargeFileUploadTask } from "../../src"; // @ts-ignore import { DummyRequestAdapter } from "../utils/DummyRequestAdapter"; import { ErrorMappings, Parsable, ParseNode } from "@microsoft/kiota-abstractions"; +import { UploadSlice } from "../../src/tasks/UploadSlice"; const adapter = new DummyRequestAdapter(); @@ -12,20 +13,20 @@ interface SampleResponse extends Parsable { uploadUrl?: string | undefined; } -export function createPageCollectionFromDiscriminatorValue( +function createPageCollectionFromDiscriminatorValue( parseNode: ParseNode | undefined, ): (instance?: Parsable) => Record void> { return deserializeIntoPageCollection; } -export function deserializeIntoPageCollection( +function deserializeIntoPageCollection( baseDeltaFunctionResponse: Partial | undefined = {}, ): Record void> { return {}; } const sampleReadableStream = new ReadableStream({ - start(controller) { + start: controller => { const encoder = new TextEncoder(); const chunk = encoder.encode("This is a 20-byte string"); controller.enqueue(chunk); @@ -54,40 +55,194 @@ export const deserializeIntoGraphError = ( }; describe("LargeFileUploadTask tests", () => { - it("should initialize", () => { - const session: SampleResponse = { - nextExpectedRanges: ["0-19"], - expirationDateTime: new Date(Date.now() + 24 * 60 * 60 * 1000), // - uploadUrl: "https://example.com/upload", - }; - const handler = new LargeFileUploadTask( - adapter, - session, - sampleReadableStream, - 10, - createPageCollectionFromDiscriminatorValue, - errorMappings, - ); - assert(handler, "LargeFileUploadTask failed to initialize"); - }); - it("should throw an error if invalid upload session is given", () => { - assert.throws(() => { - const handler = new LargeFileUploadTask( + describe("initialization", () => { + it("should initialize", () => { + const session: SampleResponse = { + nextExpectedRanges: ["0-19"], + expirationDateTime: new Date(Date.now() + 24 * 60 * 60 * 1000), // + uploadUrl: "https://example.com/upload", + }; + const largeFileUploadTask = new LargeFileUploadTask( adapter, - {} as SampleResponse, + session, sampleReadableStream, 10, createPageCollectionFromDiscriminatorValue, errorMappings, ); - }, "Upload session is invalid"); + assert.isNotNull(largeFileUploadTask, "LargeFileUploadTask failed to initialize"); + }); + it("should throw an error if invalid upload session is given", () => { + assert.throws(() => { + new LargeFileUploadTask( + adapter, + {} as SampleResponse, + sampleReadableStream, + 10, + createPageCollectionFromDiscriminatorValue, + errorMappings, + ); + }, "Upload session is invalid"); + }); }); + describe("file upload", () => { + it("should split the file into expected ranges observing max size", async () => { + adapter.resetAdapter(); + const session: SampleResponse = { + nextExpectedRanges: ["0-19"], + expirationDateTime: new Date(Date.now() + 24 * 60 * 60 * 1000), // + uploadUrl: "https://example.com/upload", + }; + const largeFileUploadTask = new LargeFileUploadTask( + adapter, + session, + sampleReadableStream, + 5, + createPageCollectionFromDiscriminatorValue, + errorMappings, + ); + + // Accessing the private method using bracket notation + const uploadSlices = (largeFileUploadTask as any)["getUploadSliceRequests"]() as UploadSlice[]; + // cast the arrays to UploadSlice + assert.equal(uploadSlices.length, 4); + + uploadSlices.forEach(slice => { + assert.isAtMost(slice.rangeEnd - slice.rangeBegin + 1, 5, "Slice size should not be larger than 5"); + }); + for (let i = 0; i < uploadSlices.length - 1; i++) { + assert.isTrue( + uploadSlices[i].rangeEnd < uploadSlices[i + 1].rangeBegin, + "Slices should be in order from largest to smallest", + ); + } + }); + it("should execute multiple file upload requests", async () => { + adapter.resetAdapter(); + const session: SampleResponse = { + nextExpectedRanges: ["0-19"], + expirationDateTime: new Date(Date.now() + 24 * 60 * 60 * 1000), // + uploadUrl: "https://example.com/upload", + }; + const largeFileUploadTask = new LargeFileUploadTask( + adapter, + session, + sampleReadableStream, + 10, + createPageCollectionFromDiscriminatorValue, + errorMappings, + ); - // TODO test slice generation by range spliting + let progressCounter = 0; + let lastCall = -1; + const progressCallback: IProgress = { + report: (progress: number) => { + lastCall = progress; + progressCounter++; + }, + }; - // TODO test file upload + adapter.setResponse({ + nextExpectedRanges: ["20-39"], + expirationDateTime: new Date(Date.now() + 24 * 60 * 60 * 1000), // + uploadUrl: "https://example.com/upload", + }); + adapter.setResponse({ + name: "Valid", + }); - // TODO test canceling an upload + await largeFileUploadTask.upload(progressCallback); + assert.equal(progressCounter, 2); + assert.equal(lastCall, 19); - // TODO test resume an upload + const requests = adapter.getRequests(); + assert.equal(requests.length, 2); + }); + it("should delete an upload session", () => { + adapter.resetAdapter(); + const session: SampleResponse = { + nextExpectedRanges: ["0-19"], + expirationDateTime: new Date(Date.now() + 24 * 60 * 60 * 1000), // + uploadUrl: "https://example.com/upload", + }; + const largeFileUploadTask = new LargeFileUploadTask( + adapter, + session, + sampleReadableStream, + 10, + createPageCollectionFromDiscriminatorValue, + errorMappings, + ); + largeFileUploadTask.deleteSession(); + + // check if it makes a delete call to the url + const requests = adapter.getRequests(); + assert.equal(requests.length, 1); + assert.equal(requests[0].httpMethod, "DELETE"); + assert.equal(requests[0].URL, session.uploadUrl); + }); + it("should update an upload session", () => { + adapter.resetAdapter(); + const session: SampleResponse = { + nextExpectedRanges: ["0-19"], + expirationDateTime: new Date(Date.now() + 24 * 60 * 60 * 1000), // + uploadUrl: "https://example.com/upload", + }; + const largeFileUploadTask = new LargeFileUploadTask( + adapter, + session, + sampleReadableStream, + 10, + createPageCollectionFromDiscriminatorValue, + errorMappings, + ); + + largeFileUploadTask.updateSession(); + + // check if it makes a get call to the url + const requests = adapter.getRequests(); + assert.equal(requests.length, 1); + assert.equal(requests[0].httpMethod, "GET"); + assert.equal(requests[0].URL, session.uploadUrl); + }); + it("should resume an upload session", async () => { + adapter.resetAdapter(); + const session: SampleResponse = { + nextExpectedRanges: ["0-19"], + expirationDateTime: new Date(Date.now() + 24 * 60 * 60 * 1000), // + uploadUrl: "https://example.com/upload", + }; + const largeFileUploadTask = new LargeFileUploadTask( + adapter, + session, + sampleReadableStream, + 10, + createPageCollectionFromDiscriminatorValue, + errorMappings, + ); + + adapter.setResponse({ + nextExpectedRanges: ["20-39"], + expirationDateTime: new Date(Date.now() + 24 * 60 * 60 * 1000), // + uploadUrl: "https://example.com/upload", + }); + adapter.setResponse({ + name: "Valid", + }); + + // check if it first call to the url is a get for refreshing the session and then multiple uploads + await largeFileUploadTask.resume(); + const requests = adapter.getRequests(); + assert.equal(requests.length, 2); + assert.equal(requests[0].httpMethod, "GET"); + assert.equal(requests[0].URL, session.uploadUrl); + + let pos = 1; + while (requests.length > pos) { + assert.equal(requests[pos].httpMethod, "PUT"); + assert.equal(requests[pos].URL, session.uploadUrl); + pos++; + } + }); + }); }); diff --git a/test/utils/DummyRequestAdapter.ts b/test/utils/DummyRequestAdapter.ts index 9416357..6611cbc 100644 --- a/test/utils/DummyRequestAdapter.ts +++ b/test/utils/DummyRequestAdapter.ts @@ -55,6 +55,11 @@ export class DummyRequestAdapter implements RequestAdapter { return this.requests; } + resetAdapter(): void { + this.requests = []; + this.response = []; + } + convertToNativeRequest(requestInfo: RequestInformation): Promise { return Promise.resolve(undefined as T); } From b2e65ba918fca1cc511087f3ac3c1306e35c28c0 Mon Sep 17 00:00:00 2001 From: rkodev <43806892+rkodev@users.noreply.github.com> Date: Mon, 24 Feb 2025 13:00:12 +0300 Subject: [PATCH 07/20] Fixes streams --- src/tasks/LargeFileUploadTask.ts | 48 +++++++++++----------- src/tasks/UploadSlice.ts | 53 +++++++++++++++++------- test/task/LargeFileUploadTask.ts | 70 ++++++++++++++++++++------------ 3 files changed, 106 insertions(+), 65 deletions(-) diff --git a/src/tasks/LargeFileUploadTask.ts b/src/tasks/LargeFileUploadTask.ts index dae7a2c..ed8880b 100644 --- a/src/tasks/LargeFileUploadTask.ts +++ b/src/tasks/LargeFileUploadTask.ts @@ -55,17 +55,12 @@ export interface UploadResult { location?: string; } -/** - * @interface - * Signature to represent the upload session response - */ -export interface UploadSessionResponse extends Parsable { - expirationDateTime?: Date | null; - nextExpectedRanges?: string[] | null; -} /** * BatchResponseCollection ParsableFactory - * @param _parseNode + * Creates a factory function to deserialize the upload session response. + * + * @param {ParseNode} _parseNode - The parse node to deserialize. + * @returns {Function} - A function that takes an instance of Parsable and returns a record of deserialization functions. */ export const createUploadSessionResponseFromDiscriminatorValue = ( _parseNode: ParseNode | undefined, @@ -74,18 +69,20 @@ export const createUploadSessionResponseFromDiscriminatorValue = ( }; /** - * Deserializes the batch response body - * @param uploadSessionResponse + * Deserializes the upload session response body. + * + * @param {Partial} [uploadSession] - The upload session object to deserialize into. + * @returns {Record void>} - A record of deserialization functions. */ export const deserializeIntoUploadSessionResponse = ( - uploadSessionResponse: Partial | undefined = {}, + uploadSession: Partial | undefined = {}, ): Record void> => { return { expirationDateTime: n => { - uploadSessionResponse.expirationDateTime = n.getDateValue(); + uploadSession.expirationDateTime = n.getDateValue(); }, nextExpectedRanges: n => { - uploadSessionResponse.nextExpectedRanges = n.getCollectionOfPrimitiveValues(); + uploadSession.nextExpectedRanges = n.getCollectionOfPrimitiveValues(); }, }; }; @@ -123,7 +120,7 @@ export class LargeFileUploadTask { * * @param {RequestAdapter} requestAdapter - The request adapter to use for making HTTP requests. * @param {Parsable} uploadSession - The upload session information. - * @param {ReadableStream} uploadStream - The stream of the file to be uploaded. + * @param {ReadableStream} uploadStreamProvider - Returns an instance of an unconsumed new stream to be uploaded. * @param {number} [maxSliceSize=-1] - The maximum size of each file slice to be uploaded. * @param {ParsableFactory} parsableFactory - The factory to create parsable objects. * @param {ErrorMappings} [errorMappings] - error mappings. @@ -133,7 +130,7 @@ export class LargeFileUploadTask { constructor( readonly requestAdapter: RequestAdapter, readonly uploadSession: Parsable, - readonly uploadStream: ReadableStream, + readonly uploadStreamProvider: () => ReadableStream, readonly maxSliceSize = -1, readonly parsableFactory: ParsableFactory, errorMappings: ErrorMappings, @@ -143,8 +140,8 @@ export class LargeFileUploadTask { error.name = "Invalid Upload Session Error"; throw error; } - if (!uploadStream) { - const error = new Error("Upload stream is undefined, Please provide a valid upload stream"); + if (!uploadStreamProvider) { + const error = new Error("Upload stream provider is undefined, Please provide a valid upload stream"); error.name = "Invalid Upload Stream Error"; throw error; } @@ -158,9 +155,6 @@ export class LargeFileUploadTask { error.name = "Invalid Parsable Factory Error"; throw error; } - if (uploadStream?.locked) { - throw new Error("Please provide stream value"); - } if (maxSliceSize <= 0) { this.maxSliceSize = DefaultSliceSize; } @@ -182,7 +176,7 @@ export class LargeFileUploadTask { const sliceRequests = this.getUploadSliceRequests(); for (const request of sliceRequests) { try { - const uploadResult = await request.uploadSlice(this.uploadStream); + const uploadResult = await this.uploadWithRetry(request); progress?.report(request.rangeEnd); if (uploadResult?.itemResponse || uploadResult?.location) { return uploadResult; @@ -206,7 +200,7 @@ export class LargeFileUploadTask { let uploadTries = 0; while (uploadTries < maxTries) { try { - return await uploadSlice.uploadSlice(this.uploadStream); + return await uploadSlice.uploadSlice(this.uploadStreamProvider()); } catch (e) { console.error(e); } @@ -238,8 +232,12 @@ export class LargeFileUploadTask { * @throws {Error} If the request fails. */ public async updateSession() { - const requestInformation = new RequestInformation(HttpMethod.GET, this.Session.uploadUrl!); - const response = await this.requestAdapter.send( + const url = this.Session.uploadUrl; + if (!url) { + throw new Error("Upload url is invalid"); + } + const requestInformation = new RequestInformation(HttpMethod.GET, url); + const response = await this.requestAdapter.send( requestInformation, createUploadSessionResponseFromDiscriminatorValue, this.errorMappings, diff --git a/src/tasks/UploadSlice.ts b/src/tasks/UploadSlice.ts index 5a2f6a2..fdeb2a6 100644 --- a/src/tasks/UploadSlice.ts +++ b/src/tasks/UploadSlice.ts @@ -84,35 +84,60 @@ export class UploadSlice { * @returns {Promise} - A promise that resolves to an ArrayBuffer containing the read bytes. */ private async readSection(stream: ReadableStream, start: number, end: number): Promise { + if (start < 0 || end < start) { + throw new Error("Invalid start or end values."); + } + const reader = stream.getReader(); - let bytesRead = 0; + let position = 0; // current absolute position in the stream const chunks: Uint8Array[] = []; - const totalBytesToRead = end - start; + let totalLength = 0; try { - while (bytesRead < totalBytesToRead) { + while (true) { const { done, value } = await reader.read(); if (done) break; - if (value) { - const remainingBytes = totalBytesToRead - bytesRead; - if (value.length > remainingBytes) { - chunks.push(value.slice(0, remainingBytes)); - bytesRead += remainingBytes; - } else { - chunks.push(value); - bytesRead += value.length; - } + if (!value) continue; + + const chunk = value; + const chunkLength = chunk.byteLength; + + // If the entire chunk is before our start, skip it. + if (position + chunkLength <= start) { + position += chunkLength; + continue; } + + // If we've already passed the end, we can stop reading. + if (position > end) break; + + // Calculate the start index within the current chunk. + const startIndex = position < start ? start - position : 0; + // Calculate the end index within the current chunk. + // Since `end` is inclusive, we need to slice up to (end - position + 1). + let endIndex = chunkLength; + if (position + chunkLength - 1 > end) { + endIndex = end - position + 1; + } + + const sliced = chunk.slice(startIndex, endIndex); + chunks.push(sliced); + totalLength += sliced.byteLength; + + position += chunkLength; + // Stop reading if we've already reached beyond the desired range. + if (position > end) break; } } finally { reader.releaseLock(); } - const result = new Uint8Array(bytesRead); + // Concatenate all collected chunks into one Uint8Array. + const result = new Uint8Array(totalLength); let offset = 0; for (const chunk of chunks) { result.set(chunk, offset); - offset += chunk.length; + offset += chunk.byteLength; } return result.buffer; diff --git a/test/task/LargeFileUploadTask.ts b/test/task/LargeFileUploadTask.ts index 231a8db..6910d17 100644 --- a/test/task/LargeFileUploadTask.ts +++ b/test/task/LargeFileUploadTask.ts @@ -25,14 +25,16 @@ function deserializeIntoPageCollection( return {}; } -const sampleReadableStream = new ReadableStream({ - start: controller => { - const encoder = new TextEncoder(); - const chunk = encoder.encode("This is a 20-byte string"); - controller.enqueue(chunk); - controller.close(); - }, -}); +function createSampleReadableStream(): ReadableStream { + return new ReadableStream({ + start: controller => { + const encoder = new TextEncoder(); + const chunk = encoder.encode("This is a 24-byte string"); + controller.enqueue(chunk); + controller.close(); + }, + }); +} const errorMappings: ErrorMappings = { XXX: parseNode => createGraphErrorFromDiscriminatorValue(parseNode), @@ -58,14 +60,14 @@ describe("LargeFileUploadTask tests", () => { describe("initialization", () => { it("should initialize", () => { const session: SampleResponse = { - nextExpectedRanges: ["0-19"], + nextExpectedRanges: ["0-23"], expirationDateTime: new Date(Date.now() + 24 * 60 * 60 * 1000), // uploadUrl: "https://example.com/upload", }; const largeFileUploadTask = new LargeFileUploadTask( adapter, session, - sampleReadableStream, + createSampleReadableStream, 10, createPageCollectionFromDiscriminatorValue, errorMappings, @@ -77,7 +79,7 @@ describe("LargeFileUploadTask tests", () => { new LargeFileUploadTask( adapter, {} as SampleResponse, - sampleReadableStream, + createSampleReadableStream, 10, createPageCollectionFromDiscriminatorValue, errorMappings, @@ -89,14 +91,14 @@ describe("LargeFileUploadTask tests", () => { it("should split the file into expected ranges observing max size", async () => { adapter.resetAdapter(); const session: SampleResponse = { - nextExpectedRanges: ["0-19"], + nextExpectedRanges: ["0-23"], expirationDateTime: new Date(Date.now() + 24 * 60 * 60 * 1000), // uploadUrl: "https://example.com/upload", }; const largeFileUploadTask = new LargeFileUploadTask( adapter, session, - sampleReadableStream, + createSampleReadableStream, 5, createPageCollectionFromDiscriminatorValue, errorMappings, @@ -105,7 +107,7 @@ describe("LargeFileUploadTask tests", () => { // Accessing the private method using bracket notation const uploadSlices = (largeFileUploadTask as any)["getUploadSliceRequests"]() as UploadSlice[]; // cast the arrays to UploadSlice - assert.equal(uploadSlices.length, 4); + assert.equal(uploadSlices.length, 5); uploadSlices.forEach(slice => { assert.isAtMost(slice.rangeEnd - slice.rangeBegin + 1, 5, "Slice size should not be larger than 5"); @@ -116,18 +118,28 @@ describe("LargeFileUploadTask tests", () => { "Slices should be in order from largest to smallest", ); } + + const decoder = new TextDecoder(); + let reconstructedString = ""; + for (const slice of uploadSlices) { + const chunk = await (slice as any).readSection(createSampleReadableStream(), slice.rangeBegin, slice.rangeEnd); + console.log(slice.rangeBegin, slice.rangeEnd); + console.log("chunk", decoder.decode(chunk)); + reconstructedString += decoder.decode(chunk); + } + assert.equal(reconstructedString, "This is a 24-byte string", "Reconstructed string should match the original"); }); it("should execute multiple file upload requests", async () => { adapter.resetAdapter(); const session: SampleResponse = { - nextExpectedRanges: ["0-19"], + nextExpectedRanges: ["0-23"], expirationDateTime: new Date(Date.now() + 24 * 60 * 60 * 1000), // uploadUrl: "https://example.com/upload", }; const largeFileUploadTask = new LargeFileUploadTask( adapter, session, - sampleReadableStream, + createSampleReadableStream, 10, createPageCollectionFromDiscriminatorValue, errorMappings, @@ -143,7 +155,13 @@ describe("LargeFileUploadTask tests", () => { }; adapter.setResponse({ - nextExpectedRanges: ["20-39"], + nextExpectedRanges: ["10-19"], + expirationDateTime: new Date(Date.now() + 24 * 60 * 60 * 1000), // + uploadUrl: "https://example.com/upload", + }); + + adapter.setResponse({ + nextExpectedRanges: ["20-23"], expirationDateTime: new Date(Date.now() + 24 * 60 * 60 * 1000), // uploadUrl: "https://example.com/upload", }); @@ -152,23 +170,23 @@ describe("LargeFileUploadTask tests", () => { }); await largeFileUploadTask.upload(progressCallback); - assert.equal(progressCounter, 2); - assert.equal(lastCall, 19); + assert.equal(progressCounter, 3); + assert.equal(lastCall, 23); const requests = adapter.getRequests(); - assert.equal(requests.length, 2); + assert.equal(requests.length, 3); }); it("should delete an upload session", () => { adapter.resetAdapter(); const session: SampleResponse = { - nextExpectedRanges: ["0-19"], + nextExpectedRanges: ["0-23"], expirationDateTime: new Date(Date.now() + 24 * 60 * 60 * 1000), // uploadUrl: "https://example.com/upload", }; const largeFileUploadTask = new LargeFileUploadTask( adapter, session, - sampleReadableStream, + createSampleReadableStream, 10, createPageCollectionFromDiscriminatorValue, errorMappings, @@ -184,14 +202,14 @@ describe("LargeFileUploadTask tests", () => { it("should update an upload session", () => { adapter.resetAdapter(); const session: SampleResponse = { - nextExpectedRanges: ["0-19"], + nextExpectedRanges: ["0-23"], expirationDateTime: new Date(Date.now() + 24 * 60 * 60 * 1000), // uploadUrl: "https://example.com/upload", }; const largeFileUploadTask = new LargeFileUploadTask( adapter, session, - sampleReadableStream, + createSampleReadableStream, 10, createPageCollectionFromDiscriminatorValue, errorMappings, @@ -208,14 +226,14 @@ describe("LargeFileUploadTask tests", () => { it("should resume an upload session", async () => { adapter.resetAdapter(); const session: SampleResponse = { - nextExpectedRanges: ["0-19"], + nextExpectedRanges: ["0-23"], expirationDateTime: new Date(Date.now() + 24 * 60 * 60 * 1000), // uploadUrl: "https://example.com/upload", }; const largeFileUploadTask = new LargeFileUploadTask( adapter, session, - sampleReadableStream, + createSampleReadableStream, 10, createPageCollectionFromDiscriminatorValue, errorMappings, From 1f3fe498bcc492ad415443ac46affcc83093bdea Mon Sep 17 00:00:00 2001 From: rkodev <43806892+rkodev@users.noreply.github.com> Date: Mon, 24 Feb 2025 17:00:33 +0300 Subject: [PATCH 08/20] minor fixes --- src/tasks/LargeFileResponseHandler.ts | 21 --------------------- src/tasks/LargeFileUploadTask.ts | 18 ++++++++++++------ 2 files changed, 12 insertions(+), 27 deletions(-) delete mode 100644 src/tasks/LargeFileResponseHandler.ts diff --git a/src/tasks/LargeFileResponseHandler.ts b/src/tasks/LargeFileResponseHandler.ts deleted file mode 100644 index 7194102..0000000 --- a/src/tasks/LargeFileResponseHandler.ts +++ /dev/null @@ -1,21 +0,0 @@ -/** - * ------------------------------------------------------------------------------------------- - * Copyright (c) Microsoft Corporation. All Rights Reserved. Licensed under the MIT License. - * See License in the project root for license information. - * ------------------------------------------------------------------------------------------- - */ - -import { ErrorMappings, ResponseHandler } from "@microsoft/kiota-abstractions"; - -/** - * @class - * Class for LargeFileResponseHandler - */ -export class LargeFileResponseHandler implements ResponseHandler { - handleResponse( - _response: NativeResponseType, - _errorMappings: ErrorMappings | undefined, - ): Promise { - return Promise.resolve(undefined); - } -} diff --git a/src/tasks/LargeFileUploadTask.ts b/src/tasks/LargeFileUploadTask.ts index ed8880b..a06e363 100644 --- a/src/tasks/LargeFileUploadTask.ts +++ b/src/tasks/LargeFileUploadTask.ts @@ -16,9 +16,9 @@ import { ParsableFactory, ParseNode, ErrorMappings, + HttpMethod, } from "@microsoft/kiota-abstractions"; import { UploadSlice } from "./UploadSlice"; -import { HttpMethod } from "@microsoft/kiota-abstractions/dist/es/src/httpMethod"; /** * @interface @@ -56,16 +56,16 @@ export interface UploadResult { } /** - * BatchResponseCollection ParsableFactory + * UploadSession ParsableFactory * Creates a factory function to deserialize the upload session response. * * @param {ParseNode} _parseNode - The parse node to deserialize. * @returns {Function} - A function that takes an instance of Parsable and returns a record of deserialization functions. */ -export const createUploadSessionResponseFromDiscriminatorValue = ( +export const createUploadSessionFromDiscriminatorValue = ( _parseNode: ParseNode | undefined, ): ((instance?: Parsable) => Record void>) => { - return deserializeIntoUploadSessionResponse; + return deserializeIntoUploadSession; }; /** @@ -74,7 +74,7 @@ export const createUploadSessionResponseFromDiscriminatorValue = ( * @param {Partial} [uploadSession] - The upload session object to deserialize into. * @returns {Record void>} - A record of deserialization functions. */ -export const deserializeIntoUploadSessionResponse = ( +export const deserializeIntoUploadSession = ( uploadSession: Partial | undefined = {}, ): Record void> => { return { @@ -84,6 +84,12 @@ export const deserializeIntoUploadSessionResponse = ( nextExpectedRanges: n => { uploadSession.nextExpectedRanges = n.getCollectionOfPrimitiveValues(); }, + "@odata.type": n => { + uploadSession.odataType = n.getStringValue(); + }, + uploadUrl: n => { + uploadSession.uploadUrl = n.getStringValue(); + }, }; }; @@ -239,7 +245,7 @@ export class LargeFileUploadTask { const requestInformation = new RequestInformation(HttpMethod.GET, url); const response = await this.requestAdapter.send( requestInformation, - createUploadSessionResponseFromDiscriminatorValue, + createUploadSessionFromDiscriminatorValue, this.errorMappings, ); From 42530ee7ab36f1d5ce0b065fefdda9b64fed40ac Mon Sep 17 00:00:00 2001 From: rkodev <43806892+rkodev@users.noreply.github.com> Date: Mon, 24 Feb 2025 17:30:47 +0300 Subject: [PATCH 09/20] update result model --- src/tasks/LargeFileUploadTask.ts | 20 ++++++++++++------ src/tasks/UploadSlice.ts | 35 ++++++++++++++++++++++---------- 2 files changed, 38 insertions(+), 17 deletions(-) diff --git a/src/tasks/LargeFileUploadTask.ts b/src/tasks/LargeFileUploadTask.ts index a06e363..e618030 100644 --- a/src/tasks/LargeFileUploadTask.ts +++ b/src/tasks/LargeFileUploadTask.ts @@ -51,7 +51,6 @@ export interface UploadSession { */ export interface UploadResult { itemResponse?: T | null; - uploadSession?: UploadSession | null; location?: string; } @@ -184,8 +183,9 @@ export class LargeFileUploadTask { try { const uploadResult = await this.uploadWithRetry(request); progress?.report(request.rangeEnd); - if (uploadResult?.itemResponse || uploadResult?.location) { - return uploadResult; + const { itemResponse, location } = uploadResult as Partial>; + if (itemResponse || location) { + return uploadResult as UploadResult; } } catch (e) { console.error(e); @@ -199,10 +199,13 @@ export class LargeFileUploadTask { * * @param {UploadSlice} uploadSlice - The upload slice to be uploaded. * @param {number} [maxTries=3] - The maximum number of retry attempts. - * @returns {Promise | undefined>} - The result of the upload. + * @returns {Promise | UploadSession | undefined>} - The result of the upload. * @throws {Error} If the maximum number of retries is reached. */ - private async uploadWithRetry(uploadSlice: UploadSlice, maxTries = 3): Promise | undefined> { + private async uploadWithRetry( + uploadSlice: UploadSlice, + maxTries = 3, + ): Promise | UploadSession | undefined> { let uploadTries = 0; while (uploadTries < maxTries) { try { @@ -235,9 +238,10 @@ export class LargeFileUploadTask { * Refreshes the current upload session status by making a GET request to the upload URL. * Updates the session expiration date, next expected ranges, and remaining ranges based on the response. * + * @returns {Promise} - A promise that resolves to the updated upload session. * @throws {Error} If the request fails. */ - public async updateSession() { + public async updateSession(): Promise { const url = this.Session.uploadUrl; if (!url) { throw new Error("Upload url is invalid"); @@ -252,8 +256,12 @@ export class LargeFileUploadTask { if (response) { this.Session.expirationDateTime = response?.expirationDateTime; this.Session.nextExpectedRanges = response.nextExpectedRanges; + if (response.uploadUrl) { + this.Session.uploadUrl = response.uploadUrl; + } this.rangesRemaining = this.getRangesRemaining(this.Session); } + return response; } /** diff --git a/src/tasks/UploadSlice.ts b/src/tasks/UploadSlice.ts index fdeb2a6..9fc5b52 100644 --- a/src/tasks/UploadSlice.ts +++ b/src/tasks/UploadSlice.ts @@ -13,6 +13,7 @@ import { ParsableFactory, RequestAdapter, RequestInformation, + type AdditionalDataHolder, } from "@microsoft/kiota-abstractions"; import { HeadersInspectionOptions } from "@microsoft/kiota-http-fetchlibrary"; import { UploadResult, UploadSession } from "./LargeFileUploadTask"; @@ -40,7 +41,7 @@ export class UploadSlice { * @param {ReadableStream} stream - The stream of the file slice to be uploaded. * @returns {Promise | undefined>} - The result of the upload operation. */ - public async uploadSlice(stream: ReadableStream): Promise | undefined> { + public async uploadSlice(stream: ReadableStream): Promise | UploadSession | undefined> { const data = await this.readSection(stream, this.rangeBegin, this.rangeEnd); const requestInformation = new RequestInformation(HttpMethod.PUT, this.sessionUrl); requestInformation.headers = new Headers([ @@ -52,29 +53,41 @@ export class UploadSlice { const headerOptions = new HeadersInspectionOptions({ inspectResponseHeaders: true }); requestInformation.addRequestOptions([headerOptions]); - let itemResponse = await this.requestAdapter.send(requestInformation, this.parsableFactory, this.errorMappings); + const itemResponse = await this.requestAdapter.send( + requestInformation, + this.parsableFactory, + this.errorMappings, + ); const locations = headerOptions.getResponseHeaders().get("location"); - let uploadSession: UploadSession | null = null; if (itemResponse) { - const { expirationDateTime, nextExpectedRanges } = itemResponse as Partial; - if (nextExpectedRanges) { - uploadSession = { - expirationDateTime, - nextExpectedRanges: (nextExpectedRanges as unknown as string[])[0].split(","), - }; - itemResponse = undefined; + if (this.isUploadSessionResponse(itemResponse)) { + return itemResponse; + } + const { additionalData } = itemResponse as Partial; + if (additionalData && this.isUploadSessionResponse(additionalData)) { + return additionalData; } } return { itemResponse, location: locations ? (locations as unknown as string[])[0] : undefined, - uploadSession, }; } + private isUploadSessionResponse(item: Parsable | AdditionalDataHolder): UploadSession | null { + const { expirationDateTime, nextExpectedRanges } = item as Partial; + if (nextExpectedRanges) { + return { + expirationDateTime, + nextExpectedRanges, + }; + } + return null; + } + /** * Reads a section of the stream from the specified start to end positions. * From d062bdf1c1eb92a5b98daf1dd65d6f9373ac9b24 Mon Sep 17 00:00:00 2001 From: rkodev <43806892+rkodev@users.noreply.github.com> Date: Mon, 24 Feb 2025 17:31:35 +0300 Subject: [PATCH 10/20] update result model --- src/tasks/LargeFileUploadTask.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/tasks/LargeFileUploadTask.ts b/src/tasks/LargeFileUploadTask.ts index e618030..32ad977 100644 --- a/src/tasks/LargeFileUploadTask.ts +++ b/src/tasks/LargeFileUploadTask.ts @@ -238,10 +238,10 @@ export class LargeFileUploadTask { * Refreshes the current upload session status by making a GET request to the upload URL. * Updates the session expiration date, next expected ranges, and remaining ranges based on the response. * - * @returns {Promise} - A promise that resolves to the updated upload session. + * @returns {Promise} - A promise that resolves to the updated upload session. * @throws {Error} If the request fails. */ - public async updateSession(): Promise { + public async updateSession(): Promise { const url = this.Session.uploadUrl; if (!url) { throw new Error("Upload url is invalid"); From 24eb9298743132e2e265897729b197a7555521c6 Mon Sep 17 00:00:00 2001 From: rkodev <43806892+rkodev@users.noreply.github.com> Date: Tue, 25 Feb 2025 13:00:16 +0300 Subject: [PATCH 11/20] fix response --- src/tasks/UploadSlice.ts | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/src/tasks/UploadSlice.ts b/src/tasks/UploadSlice.ts index 9fc5b52..0df4370 100644 --- a/src/tasks/UploadSlice.ts +++ b/src/tasks/UploadSlice.ts @@ -62,12 +62,16 @@ export class UploadSlice { const locations = headerOptions.getResponseHeaders().get("location"); if (itemResponse) { - if (this.isUploadSessionResponse(itemResponse)) { - return itemResponse; + let sessionResponse = this.isUploadSessionResponse(itemResponse); + if (sessionResponse) { + return sessionResponse; } const { additionalData } = itemResponse as Partial; - if (additionalData && this.isUploadSessionResponse(additionalData)) { - return additionalData; + if (additionalData) { + sessionResponse = this.isUploadSessionResponse(additionalData); + if (sessionResponse) { + return sessionResponse; + } } } From 71e8530d85d85338ba7bca5cba2dc4ae36c52128 Mon Sep 17 00:00:00 2001 From: rkodev <43806892+rkodev@users.noreply.github.com> Date: Tue, 25 Feb 2025 19:01:59 +0300 Subject: [PATCH 12/20] Adds seekable stream --- src/tasks/LargeFileUploadTask.ts | 25 +++++-- src/tasks/SeekableStreamReader.ts | 110 ++++++++++++++++++++++++++++++ src/tasks/UploadSlice.ts | 87 +++++------------------ src/tasks/index.ts | 2 + test/task/LargeFileUploadTask.ts | 21 +++--- 5 files changed, 159 insertions(+), 86 deletions(-) create mode 100644 src/tasks/SeekableStreamReader.ts diff --git a/src/tasks/LargeFileUploadTask.ts b/src/tasks/LargeFileUploadTask.ts index 32ad977..15bf077 100644 --- a/src/tasks/LargeFileUploadTask.ts +++ b/src/tasks/LargeFileUploadTask.ts @@ -19,6 +19,7 @@ import { HttpMethod, } from "@microsoft/kiota-abstractions"; import { UploadSlice } from "./UploadSlice"; +import { SeekableStreamReader } from "./SeekableStreamReader"; /** * @interface @@ -114,6 +115,12 @@ export class LargeFileUploadTask { */ errorMappings: ErrorMappings; + /** + * @private + * The seekable stream reader + */ + seekableStreamReader: SeekableStreamReader; + /** * @private * The upload session @@ -125,7 +132,7 @@ export class LargeFileUploadTask { * * @param {RequestAdapter} requestAdapter - The request adapter to use for making HTTP requests. * @param {Parsable} uploadSession - The upload session information. - * @param {ReadableStream} uploadStreamProvider - Returns an instance of an unconsumed new stream to be uploaded. + * @param {ReadableStream} uploadStream - Returns an instance of an unconsumed new stream to be uploaded. * @param {number} [maxSliceSize=-1] - The maximum size of each file slice to be uploaded. * @param {ParsableFactory} parsableFactory - The factory to create parsable objects. * @param {ErrorMappings} [errorMappings] - error mappings. @@ -135,7 +142,7 @@ export class LargeFileUploadTask { constructor( readonly requestAdapter: RequestAdapter, readonly uploadSession: Parsable, - readonly uploadStreamProvider: () => ReadableStream, + uploadStream: ReadableStream, readonly maxSliceSize = -1, readonly parsableFactory: ParsableFactory, errorMappings: ErrorMappings, @@ -145,8 +152,8 @@ export class LargeFileUploadTask { error.name = "Invalid Upload Session Error"; throw error; } - if (!uploadStreamProvider) { - const error = new Error("Upload stream provider is undefined, Please provide a valid upload stream"); + if (!uploadStream) { + const error = new Error("Upload stream is undefined, Please provide a valid upload stream"); error.name = "Invalid Upload Stream Error"; throw error; } @@ -160,11 +167,18 @@ export class LargeFileUploadTask { error.name = "Invalid Parsable Factory Error"; throw error; } + if (uploadStream.locked) { + const error = new Error("Upload stream is locked, Please provide a valid upload stream"); + error.name = "Invalid Upload Stream Error"; + throw error; + } if (maxSliceSize <= 0) { this.maxSliceSize = DefaultSliceSize; } this.parsableFactory = parsableFactory; + this.seekableStreamReader = new SeekableStreamReader(uploadStream); + this.Session = this.extractSessionInfo(uploadSession); this.rangesRemaining = this.getRangesRemaining(this.Session); this.errorMappings = errorMappings; @@ -209,7 +223,7 @@ export class LargeFileUploadTask { let uploadTries = 0; while (uploadTries < maxTries) { try { - return await uploadSlice.uploadSlice(this.uploadStreamProvider()); + return await uploadSlice.uploadSlice(); } catch (e) { console.error(e); } @@ -326,6 +340,7 @@ export class LargeFileUploadTask { range[1] + 1, this.parsableFactory, this.errorMappings, + this.seekableStreamReader, ); uploadSlices.push(uploadRequest); currentRangeBegin += nextSliceSize; diff --git a/src/tasks/SeekableStreamReader.ts b/src/tasks/SeekableStreamReader.ts new file mode 100644 index 0000000..058f5d0 --- /dev/null +++ b/src/tasks/SeekableStreamReader.ts @@ -0,0 +1,110 @@ +/** + * ------------------------------------------------------------------------------------------- + * Copyright (c) Microsoft Corporation. All Rights Reserved. Licensed under the MIT License. + * See License in the project root for license information. + * ------------------------------------------------------------------------------------------- + */ + +/** + * @class + * A class that provides seekable read access to a `ReadableStream` of `Uint8Array`. + */ +export class SeekableStreamReader { + private reader: ReadableStreamDefaultReader; + private cachedChunk?: Uint8Array | null; + private cachedPosition = 0; + + constructor(private stream: ReadableStream) { + // Get a reader from the underlying stream. + this.reader = stream.getReader(); + } + + /** + * Reads a section of the stream from the given start index up to (but not including) the end index. + * The method ensures that enough data is buffered; if not, it continues reading from the stream. + * + * @param start - The starting byte index. + * @param end - The ending byte index (non-inclusive). + * @returns A Promise that resolves with an ArrayBuffer containing the requested bytes. + */ + public async readSection(start: number, end: number): Promise { + if (start < 0 || end < start) { + throw new Error("Invalid start or end values."); + } + if (start < this.cachedPosition) { + throw new Error("Cannot seek backwards"); + } + + let position = this.cachedPosition; // current absolute position in the stream + const chunks: Uint8Array[] = []; + let totalLength = 0; + + try { + while (true) { + if (!this.cachedChunk) { + const { done, value } = await this.reader.read(); + if (done) break; + if (!value) continue; + + this.cachedChunk = value; + } + const chunk = this.cachedChunk; + const chunkLength = this.cachedChunk.byteLength; + + // If the entire chunk is before our start, skip it. + if (position + chunkLength <= start) { + position += chunkLength; + continue; + } + + // If we've already passed the end, we can stop reading. + this.cachedChunk = null; + if (position > end) break; + + // Calculate the start index within the current chunk. + const startIndex = position < start ? start - position : 0; + // Calculate the end index within the current chunk. + // Since `end` is inclusive, we need to slice up to (end - position + 1). + let endIndex = chunkLength; + if (position + chunkLength - 1 > end) { + endIndex = end - position + 1; + } + + const sliced = chunk.slice(startIndex, endIndex); + chunks.push(sliced); + totalLength += sliced.byteLength; + + this.cachedChunk = chunk; + + position += chunkLength; + // Stop reading if we've already reached beyond the desired range. + if (position > end) break; + } + } finally { + this.reader.releaseLock(); + } + + // Concatenate all collected chunks into one Uint8Array. + const result = new Uint8Array(totalLength); + let offset = 0; + for (const chunk of chunks) { + result.set(chunk, offset); + offset += chunk.byteLength; + } + + return result.buffer; + } + + /** + * Optional helper: resets the internal state to allow a new stream to be used. + * Note: this method clears any cached data. + * + * @param newStream - A new ReadableStream of Uint8Array. + */ + public reset(newStream: ReadableStream): void { + this.stream = newStream; + this.reader = newStream.getReader(); + this.cachedChunk = null; + this.cachedPosition = 0; + } +} diff --git a/src/tasks/UploadSlice.ts b/src/tasks/UploadSlice.ts index 0df4370..29c6951 100644 --- a/src/tasks/UploadSlice.ts +++ b/src/tasks/UploadSlice.ts @@ -17,6 +17,7 @@ import { } from "@microsoft/kiota-abstractions"; import { HeadersInspectionOptions } from "@microsoft/kiota-http-fetchlibrary"; import { UploadResult, UploadSession } from "./LargeFileUploadTask"; +import { SeekableStreamReader } from "./SeekableStreamReader"; const binaryContentType = "application/octet-stream"; @@ -25,6 +26,18 @@ const binaryContentType = "application/octet-stream"; * Class for UploadSlice */ export class UploadSlice { + /** + * Constructs an instance of the UploadSlice class. + * + * @param {RequestAdapter} requestAdapter - The request adapter to use for making HTTP requests. + * @param {string} sessionUrl - The URL of the upload session. + * @param {number} rangeBegin - The beginning byte position of the slice. + * @param {number} rangeEnd - The ending byte position of the slice. + * @param {number} totalSessionLength - The total length of the upload session. + * @param {ParsableFactory} parsableFactory - The factory to create parsable objects. + * @param {ErrorMappings} errorMappings - The mappings for handling errors. + * @param {SeekableStreamReader} seekableStreamReader - The stream reader to read the file slice. + */ constructor( readonly requestAdapter: RequestAdapter, readonly sessionUrl: string, @@ -33,16 +46,16 @@ export class UploadSlice { readonly totalSessionLength: number, readonly parsableFactory: ParsableFactory, readonly errorMappings: ErrorMappings, + readonly seekableStreamReader: SeekableStreamReader, ) {} /** * Uploads a slice of the file to the server. * - * @param {ReadableStream} stream - The stream of the file slice to be uploaded. * @returns {Promise | undefined>} - The result of the upload operation. */ - public async uploadSlice(stream: ReadableStream): Promise | UploadSession | undefined> { - const data = await this.readSection(stream, this.rangeBegin, this.rangeEnd); + public async uploadSlice(): Promise | UploadSession | undefined> { + const data = await this.seekableStreamReader.readSection(this.rangeBegin, this.rangeEnd); const requestInformation = new RequestInformation(HttpMethod.PUT, this.sessionUrl); requestInformation.headers = new Headers([ ["Content-Range", new Set([`bytes ${this.rangeBegin}-${this.rangeEnd - 1}/${this.totalSessionLength}`])], @@ -91,72 +104,4 @@ export class UploadSlice { } return null; } - - /** - * Reads a section of the stream from the specified start to end positions. - * - * @param {ReadableStream} stream - The stream to read from. - * @param {number} start - The starting byte position. - * @param {number} end - The ending byte position. - * @returns {Promise} - A promise that resolves to an ArrayBuffer containing the read bytes. - */ - private async readSection(stream: ReadableStream, start: number, end: number): Promise { - if (start < 0 || end < start) { - throw new Error("Invalid start or end values."); - } - - const reader = stream.getReader(); - let position = 0; // current absolute position in the stream - const chunks: Uint8Array[] = []; - let totalLength = 0; - - try { - while (true) { - const { done, value } = await reader.read(); - if (done) break; - if (!value) continue; - - const chunk = value; - const chunkLength = chunk.byteLength; - - // If the entire chunk is before our start, skip it. - if (position + chunkLength <= start) { - position += chunkLength; - continue; - } - - // If we've already passed the end, we can stop reading. - if (position > end) break; - - // Calculate the start index within the current chunk. - const startIndex = position < start ? start - position : 0; - // Calculate the end index within the current chunk. - // Since `end` is inclusive, we need to slice up to (end - position + 1). - let endIndex = chunkLength; - if (position + chunkLength - 1 > end) { - endIndex = end - position + 1; - } - - const sliced = chunk.slice(startIndex, endIndex); - chunks.push(sliced); - totalLength += sliced.byteLength; - - position += chunkLength; - // Stop reading if we've already reached beyond the desired range. - if (position > end) break; - } - } finally { - reader.releaseLock(); - } - - // Concatenate all collected chunks into one Uint8Array. - const result = new Uint8Array(totalLength); - let offset = 0; - for (const chunk of chunks) { - result.set(chunk, offset); - offset += chunk.byteLength; - } - - return result.buffer; - } } diff --git a/src/tasks/index.ts b/src/tasks/index.ts index 3420ba9..ea9c5d5 100644 --- a/src/tasks/index.ts +++ b/src/tasks/index.ts @@ -1,2 +1,4 @@ export * from "./LargeFileUploadTask"; export * from "./PageIterator.js"; +export * from "./SeekableStreamReader.js"; +export * from "./UploadSlice.js"; diff --git a/test/task/LargeFileUploadTask.ts b/test/task/LargeFileUploadTask.ts index 6910d17..6b0a4ad 100644 --- a/test/task/LargeFileUploadTask.ts +++ b/test/task/LargeFileUploadTask.ts @@ -1,5 +1,5 @@ -import { assert, describe, expect, it } from "vitest"; -import { coreVersion, IProgress, LargeFileUploadTask } from "../../src"; +import { assert, describe, it } from "vitest"; +import { IProgress, LargeFileUploadTask, SeekableStreamReader } from "../../src"; // @ts-ignore import { DummyRequestAdapter } from "../utils/DummyRequestAdapter"; import { ErrorMappings, Parsable, ParseNode } from "@microsoft/kiota-abstractions"; @@ -67,7 +67,7 @@ describe("LargeFileUploadTask tests", () => { const largeFileUploadTask = new LargeFileUploadTask( adapter, session, - createSampleReadableStream, + createSampleReadableStream(), 10, createPageCollectionFromDiscriminatorValue, errorMappings, @@ -79,7 +79,7 @@ describe("LargeFileUploadTask tests", () => { new LargeFileUploadTask( adapter, {} as SampleResponse, - createSampleReadableStream, + createSampleReadableStream(), 10, createPageCollectionFromDiscriminatorValue, errorMappings, @@ -98,7 +98,7 @@ describe("LargeFileUploadTask tests", () => { const largeFileUploadTask = new LargeFileUploadTask( adapter, session, - createSampleReadableStream, + createSampleReadableStream(), 5, createPageCollectionFromDiscriminatorValue, errorMappings, @@ -121,8 +121,9 @@ describe("LargeFileUploadTask tests", () => { const decoder = new TextDecoder(); let reconstructedString = ""; + const seekableStream = new SeekableStreamReader(createSampleReadableStream()); for (const slice of uploadSlices) { - const chunk = await (slice as any).readSection(createSampleReadableStream(), slice.rangeBegin, slice.rangeEnd); + const chunk = await seekableStream.readSection(slice.rangeBegin, slice.rangeEnd); console.log(slice.rangeBegin, slice.rangeEnd); console.log("chunk", decoder.decode(chunk)); reconstructedString += decoder.decode(chunk); @@ -139,7 +140,7 @@ describe("LargeFileUploadTask tests", () => { const largeFileUploadTask = new LargeFileUploadTask( adapter, session, - createSampleReadableStream, + createSampleReadableStream(), 10, createPageCollectionFromDiscriminatorValue, errorMappings, @@ -186,7 +187,7 @@ describe("LargeFileUploadTask tests", () => { const largeFileUploadTask = new LargeFileUploadTask( adapter, session, - createSampleReadableStream, + createSampleReadableStream(), 10, createPageCollectionFromDiscriminatorValue, errorMappings, @@ -209,7 +210,7 @@ describe("LargeFileUploadTask tests", () => { const largeFileUploadTask = new LargeFileUploadTask( adapter, session, - createSampleReadableStream, + createSampleReadableStream(), 10, createPageCollectionFromDiscriminatorValue, errorMappings, @@ -233,7 +234,7 @@ describe("LargeFileUploadTask tests", () => { const largeFileUploadTask = new LargeFileUploadTask( adapter, session, - createSampleReadableStream, + createSampleReadableStream(), 10, createPageCollectionFromDiscriminatorValue, errorMappings, From 86e1b036f93df8cb4922b869f55a938bab991418 Mon Sep 17 00:00:00 2001 From: rkodev <43806892+rkodev@users.noreply.github.com> Date: Tue, 25 Feb 2025 19:14:08 +0300 Subject: [PATCH 13/20] Fix range coment --- src/tasks/LargeFileUploadTask.ts | 60 +++++++------------------------- src/tasks/UploadSlice.ts | 4 ++- 2 files changed, 15 insertions(+), 49 deletions(-) diff --git a/src/tasks/LargeFileUploadTask.ts b/src/tasks/LargeFileUploadTask.ts index 15bf077..b642b6c 100644 --- a/src/tasks/LargeFileUploadTask.ts +++ b/src/tasks/LargeFileUploadTask.ts @@ -24,7 +24,7 @@ import { SeekableStreamReader } from "./SeekableStreamReader"; /** * @interface * Signature to represent progress receiver - * @property {number} progress - The progress value + * @property {number} progress - The progress value (This is the last uploaded byte) */ export interface IProgress { report(progress: number): void; @@ -140,11 +140,11 @@ export class LargeFileUploadTask { * @throws {Error} If any of the required parameters are undefined or invalid. */ constructor( - readonly requestAdapter: RequestAdapter, - readonly uploadSession: Parsable, + private readonly requestAdapter: RequestAdapter, + uploadSession: Parsable, uploadStream: ReadableStream, - readonly maxSliceSize = -1, - readonly parsableFactory: ParsableFactory, + private readonly maxSliceSize = -1, + private readonly parsableFactory: ParsableFactory, errorMappings: ErrorMappings, ) { if (!uploadSession) { @@ -195,7 +195,7 @@ export class LargeFileUploadTask { const sliceRequests = this.getUploadSliceRequests(); for (const request of sliceRequests) { try { - const uploadResult = await this.uploadWithRetry(request); + const uploadResult = await request.uploadSlice(); progress?.report(request.rangeEnd); const { itemResponse, location } = uploadResult as Partial>; if (itemResponse || location) { @@ -208,36 +208,6 @@ export class LargeFileUploadTask { throw new Error("Upload failed"); } - /** - * Uploads a slice with retry logic. - * - * @param {UploadSlice} uploadSlice - The upload slice to be uploaded. - * @param {number} [maxTries=3] - The maximum number of retry attempts. - * @returns {Promise | UploadSession | undefined>} - The result of the upload. - * @throws {Error} If the maximum number of retries is reached. - */ - private async uploadWithRetry( - uploadSlice: UploadSlice, - maxTries = 3, - ): Promise | UploadSession | undefined> { - let uploadTries = 0; - while (uploadTries < maxTries) { - try { - return await uploadSlice.uploadSlice(); - } catch (e) { - console.error(e); - } - uploadTries++; - - if (uploadTries < maxTries) { - // Exponential backoff - await this.sleep(2000 * (uploadTries + 1)); - } - } - - throw new Error("Max retries reached"); - } - /** * @public * Resumes the current upload session @@ -268,7 +238,7 @@ export class LargeFileUploadTask { ); if (response) { - this.Session.expirationDateTime = response?.expirationDateTime; + this.Session.expirationDateTime = response.expirationDateTime; this.Session.nextExpectedRanges = response.nextExpectedRanges; if (response.uploadUrl) { this.Session.uploadUrl = response.uploadUrl; @@ -285,7 +255,11 @@ export class LargeFileUploadTask { * @returns {Promise} A promise that resolves when the session is canceled. */ public async deleteSession(): Promise { - const requestInformation = new RequestInformation(HttpMethod.DELETE, this.Session.uploadUrl!); + const url = this.Session.uploadUrl; + if (!url) { + throw new Error("Upload url is invalid"); + } + const requestInformation = new RequestInformation(HttpMethod.DELETE, url); await this.requestAdapter.sendNoResponseContent(requestInformation, this.errorMappings); } @@ -308,16 +282,6 @@ export class LargeFileUploadTask { }; } - /** - * Pauses execution for a specified number of milliseconds. - * - * @param {number} ms - The number of milliseconds to sleep. - * @returns {Promise} A promise that resolves after the specified time. - */ - private sleep(ms: number): Promise { - return new Promise(resolve => setTimeout(resolve, ms)); - } - /** * Generates a list of upload slice requests based on the remaining ranges to be uploaded. * diff --git a/src/tasks/UploadSlice.ts b/src/tasks/UploadSlice.ts index 29c6951..b94a5f4 100644 --- a/src/tasks/UploadSlice.ts +++ b/src/tasks/UploadSlice.ts @@ -23,7 +23,9 @@ const binaryContentType = "application/octet-stream"; /** * @class - * Class for UploadSlice + * Represents a slice of a file to be uploaded. + * + * @template T - The type of the parsable object. */ export class UploadSlice { /** From 9ea4d5808650d359dcd7c5ebccdf972dc6edc5ca Mon Sep 17 00:00:00 2001 From: rkodev <43806892+rkodev@users.noreply.github.com> Date: Wed, 26 Feb 2025 08:03:46 +0300 Subject: [PATCH 14/20] Adds seekable reader --- src/tasks/SeekableStreamReader.ts | 97 +++++++++++++------------------ test/task/SeekableStreamReader.ts | 45 ++++++++++++++ 2 files changed, 86 insertions(+), 56 deletions(-) create mode 100644 test/task/SeekableStreamReader.ts diff --git a/src/tasks/SeekableStreamReader.ts b/src/tasks/SeekableStreamReader.ts index 058f5d0..91de6f6 100644 --- a/src/tasks/SeekableStreamReader.ts +++ b/src/tasks/SeekableStreamReader.ts @@ -13,78 +13,67 @@ export class SeekableStreamReader { private reader: ReadableStreamDefaultReader; private cachedChunk?: Uint8Array | null; private cachedPosition = 0; + private cachedOffset = 0; // Track where we are within the cached chunk constructor(private stream: ReadableStream) { - // Get a reader from the underlying stream. this.reader = stream.getReader(); } - /** - * Reads a section of the stream from the given start index up to (but not including) the end index. - * The method ensures that enough data is buffered; if not, it continues reading from the stream. - * - * @param start - The starting byte index. - * @param end - The ending byte index (non-inclusive). - * @returns A Promise that resolves with an ArrayBuffer containing the requested bytes. - */ public async readSection(start: number, end: number): Promise { - if (start < 0 || end < start) { - throw new Error("Invalid start or end values."); + if (start < 0 || end <= start) { + throw new Error("Invalid range: start must be non-negative and end must be greater than start."); } if (start < this.cachedPosition) { - throw new Error("Cannot seek backwards"); + throw new Error("Cannot seek backwards."); } - let position = this.cachedPosition; // current absolute position in the stream const chunks: Uint8Array[] = []; let totalLength = 0; + let position = this.cachedPosition; - try { - while (true) { - if (!this.cachedChunk) { - const { done, value } = await this.reader.read(); - if (done) break; - if (!value) continue; - - this.cachedChunk = value; - } - const chunk = this.cachedChunk; - const chunkLength = this.cachedChunk.byteLength; + while (position < end) { + if (!this.cachedChunk) { + const { done, value } = await this.reader.read(); + if (done) break; + if (!value) continue; + this.cachedChunk = value; + this.cachedOffset = 0; // Reset the offset since it's a new chunk + } - // If the entire chunk is before our start, skip it. - if (position + chunkLength <= start) { - position += chunkLength; - continue; - } + const chunk = this.cachedChunk; + const chunkLength = chunk.byteLength; - // If we've already passed the end, we can stop reading. + // Skip chunks that are entirely before `start` + if (position + chunkLength - this.cachedOffset <= start) { + position += chunkLength - this.cachedOffset; this.cachedChunk = null; - if (position > end) break; + this.cachedOffset = 0; + continue; + } - // Calculate the start index within the current chunk. - const startIndex = position < start ? start - position : 0; - // Calculate the end index within the current chunk. - // Since `end` is inclusive, we need to slice up to (end - position + 1). - let endIndex = chunkLength; - if (position + chunkLength - 1 > end) { - endIndex = end - position + 1; - } + // Calculate the section of the chunk to return + const startIndex = Math.max(0, start - position) + this.cachedOffset; + const endIndex = Math.min(chunkLength, end - position + this.cachedOffset); - const sliced = chunk.slice(startIndex, endIndex); - chunks.push(sliced); - totalLength += sliced.byteLength; + const sliced = chunk.slice(startIndex, endIndex); + chunks.push(sliced); + totalLength += sliced.byteLength; - this.cachedChunk = chunk; + // Update position correctly + position += sliced.byteLength; - position += chunkLength; - // Stop reading if we've already reached beyond the desired range. - if (position > end) break; + // Store remaining unread data for future reads + if (endIndex < chunkLength) { + this.cachedOffset = endIndex; + } else { + this.cachedChunk = null; + this.cachedOffset = 0; } - } finally { - this.reader.releaseLock(); } - // Concatenate all collected chunks into one Uint8Array. + this.cachedPosition = position; // Ensure position is updated + + // Combine collected chunks into a single buffer const result = new Uint8Array(totalLength); let offset = 0; for (const chunk of chunks) { @@ -95,16 +84,12 @@ export class SeekableStreamReader { return result.buffer; } - /** - * Optional helper: resets the internal state to allow a new stream to be used. - * Note: this method clears any cached data. - * - * @param newStream - A new ReadableStream of Uint8Array. - */ - public reset(newStream: ReadableStream): void { + public async reset(newStream: ReadableStream) { + await this.reader.cancel(); // Ensure the old reader is closed before replacing this.stream = newStream; this.reader = newStream.getReader(); this.cachedChunk = null; this.cachedPosition = 0; + this.cachedOffset = 0; } } diff --git a/test/task/SeekableStreamReader.ts b/test/task/SeekableStreamReader.ts new file mode 100644 index 0000000..5cdae9a --- /dev/null +++ b/test/task/SeekableStreamReader.ts @@ -0,0 +1,45 @@ +import { assert, describe, it } from "vitest"; +import { SeekableStreamReader } from "../../src"; + +function createRandoReadableStream(value: string): ReadableStream { + return new ReadableStream({ + start: controller => { + const encoder = new TextEncoder(); + const chunk = encoder.encode(value); + controller.enqueue(chunk); + controller.close(); + }, + }); +} + +describe("SeekableStreamReader tests", () => { + it("should read the sections of the stream", async () => { + const value = veryLongRandomText; + const stream = createRandoReadableStream(value); + + const reader = new SeekableStreamReader(stream); + + // split the stream into pairs 20 pairs of start and end + const pairs = []; + const size = new TextEncoder().encode(veryLongRandomText).length; + const units = 3; + + const batchSize = Math.round(size / units); + for (let i = 0; i < value.length; i += batchSize) { + const start: number = pairs.length === 0 ? i : pairs[pairs.length - 1][1] + 1; + pairs.push([start, Math.min(start + batchSize, value.length)]); + } + + const decoder = new TextDecoder(); + let reconstructedString = ""; + for (const [start, end] of pairs) { + const section = await reader.readSection(start, end); + reconstructedString += decoder.decode(section); + } + assert.equal(value, reconstructedString, "Reconstructed string should match the original"); + }); +}); + +const veryLongRandomText = "This is a very long text that will be used to test the SeekableStreamReader class. ".repeat( + 6, +); //.split('').sort(() => 0.5 - Math.random()).join(''); From 2a1442e99738345a29bdb8e22b1c6f8643ff2f81 Mon Sep 17 00:00:00 2001 From: rkodev <43806892+rkodev@users.noreply.github.com> Date: Thu, 27 Feb 2025 11:45:40 +0300 Subject: [PATCH 15/20] Fix issue dropping 1 byte --- src/tasks/SeekableStreamReader.ts | 2 +- test/task/SeekableStreamReader.ts | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/src/tasks/SeekableStreamReader.ts b/src/tasks/SeekableStreamReader.ts index 91de6f6..37a468c 100644 --- a/src/tasks/SeekableStreamReader.ts +++ b/src/tasks/SeekableStreamReader.ts @@ -52,7 +52,7 @@ export class SeekableStreamReader { } // Calculate the section of the chunk to return - const startIndex = Math.max(0, start - position) + this.cachedOffset; + const startIndex = Math.max(0, start - position - 1) + this.cachedOffset; const endIndex = Math.min(chunkLength, end - position + this.cachedOffset); const sliced = chunk.slice(startIndex, endIndex); diff --git a/test/task/SeekableStreamReader.ts b/test/task/SeekableStreamReader.ts index 5cdae9a..7e1caaf 100644 --- a/test/task/SeekableStreamReader.ts +++ b/test/task/SeekableStreamReader.ts @@ -22,7 +22,7 @@ describe("SeekableStreamReader tests", () => { // split the stream into pairs 20 pairs of start and end const pairs = []; const size = new TextEncoder().encode(veryLongRandomText).length; - const units = 3; + const units = 4; const batchSize = Math.round(size / units); for (let i = 0; i < value.length; i += batchSize) { @@ -41,5 +41,5 @@ describe("SeekableStreamReader tests", () => { }); const veryLongRandomText = "This is a very long text that will be used to test the SeekableStreamReader class. ".repeat( - 6, + 20, ); //.split('').sort(() => 0.5 - Math.random()).join(''); From db8b2e76be2584dbe8af958cb679035e242c26fd Mon Sep 17 00:00:00 2001 From: rkodev <43806892+rkodev@users.noreply.github.com> Date: Thu, 27 Feb 2025 11:50:42 +0300 Subject: [PATCH 16/20] Update doc --- src/tasks/SeekableStreamReader.ts | 26 +++++++++++++++++++++++++- 1 file changed, 25 insertions(+), 1 deletion(-) diff --git a/src/tasks/SeekableStreamReader.ts b/src/tasks/SeekableStreamReader.ts index 37a468c..128b211 100644 --- a/src/tasks/SeekableStreamReader.ts +++ b/src/tasks/SeekableStreamReader.ts @@ -6,8 +6,10 @@ */ /** - * @class + * @class + * * A class that provides seekable read access to a `ReadableStream` of `Uint8Array`. + * This class allows reading specific sections of a stream without seeking backwards. */ export class SeekableStreamReader { private reader: ReadableStreamDefaultReader; @@ -15,10 +17,26 @@ export class SeekableStreamReader { private cachedPosition = 0; private cachedOffset = 0; // Track where we are within the cached chunk + /** + * Creates an instance of SeekableStreamReader. + * @param {ReadableStream} stream - The readable stream to read from. + */ constructor(private stream: ReadableStream) { this.reader = stream.getReader(); } + /** + * Reads a section of the stream from the specified start position to the end position. + * + * This method reads data from the stream starting at the `start` position and ending at the `end` position. + * It ensures that the read operation does not seek backwards and handles chunked reading from the stream. + * The read data is collected into a single `ArrayBuffer` and returned. + * + * @param {number} start - The starting position of the section to read. Must be non-negative. + * @param {number} end - The ending position of the section to read. Must be greater than the start position. + * @returns {Promise} A promise that resolves to an `ArrayBuffer` containing the read section. + * @throws {Error} If the start position is negative, the end position is not greater than the start position, or if seeking backwards. + */ public async readSection(start: number, end: number): Promise { if (start < 0 || end <= start) { throw new Error("Invalid range: start must be non-negative and end must be greater than start."); @@ -84,6 +102,12 @@ export class SeekableStreamReader { return result.buffer; } + /** + * Resets the reader with a new stream. + * + * @param {ReadableStream} newStream - The new readable stream to read from. + * @returns {Promise} A promise that resolves when the reader has been reset. + */ public async reset(newStream: ReadableStream) { await this.reader.cancel(); // Ensure the old reader is closed before replacing this.stream = newStream; From 5c954a96a008db418d9fea02975a5f71a1397301 Mon Sep 17 00:00:00 2001 From: rkodev <43806892+rkodev@users.noreply.github.com> Date: Thu, 27 Feb 2025 12:43:13 +0300 Subject: [PATCH 17/20] Fix seekable --- src/tasks/SeekableStreamReader.ts | 2 +- test/task/LargeFileUploadTask.ts | 2 -- test/task/SeekableStreamReader.ts | 2 +- 3 files changed, 2 insertions(+), 4 deletions(-) diff --git a/src/tasks/SeekableStreamReader.ts b/src/tasks/SeekableStreamReader.ts index 128b211..47ff6a9 100644 --- a/src/tasks/SeekableStreamReader.ts +++ b/src/tasks/SeekableStreamReader.ts @@ -71,7 +71,7 @@ export class SeekableStreamReader { // Calculate the section of the chunk to return const startIndex = Math.max(0, start - position - 1) + this.cachedOffset; - const endIndex = Math.min(chunkLength, end - position + this.cachedOffset); + const endIndex = Math.min(chunkLength, end - position + this.cachedOffset + 1); const sliced = chunk.slice(startIndex, endIndex); chunks.push(sliced); diff --git a/test/task/LargeFileUploadTask.ts b/test/task/LargeFileUploadTask.ts index 6b0a4ad..3a56725 100644 --- a/test/task/LargeFileUploadTask.ts +++ b/test/task/LargeFileUploadTask.ts @@ -124,8 +124,6 @@ describe("LargeFileUploadTask tests", () => { const seekableStream = new SeekableStreamReader(createSampleReadableStream()); for (const slice of uploadSlices) { const chunk = await seekableStream.readSection(slice.rangeBegin, slice.rangeEnd); - console.log(slice.rangeBegin, slice.rangeEnd); - console.log("chunk", decoder.decode(chunk)); reconstructedString += decoder.decode(chunk); } assert.equal(reconstructedString, "This is a 24-byte string", "Reconstructed string should match the original"); diff --git a/test/task/SeekableStreamReader.ts b/test/task/SeekableStreamReader.ts index 7e1caaf..61d8e0c 100644 --- a/test/task/SeekableStreamReader.ts +++ b/test/task/SeekableStreamReader.ts @@ -40,6 +40,6 @@ describe("SeekableStreamReader tests", () => { }); }); -const veryLongRandomText = "This is a very long text that will be used to test the SeekableStreamReader class. ".repeat( +const veryLongRandomText = "This is a very long text that will be used to test the SeekableStreamReader class.".repeat( 20, ); //.split('').sort(() => 0.5 - Math.random()).join(''); From 1c0e35403bb5c7d9aba1772097ca9a2737da5fb2 Mon Sep 17 00:00:00 2001 From: rkodev <43806892+rkodev@users.noreply.github.com> Date: Thu, 27 Feb 2025 16:29:59 +0300 Subject: [PATCH 18/20] optimize slice requests --- src/tasks/LargeFileUploadTask.ts | 18 +++++++++++++----- src/tasks/SeekableStreamReader.ts | 19 ++----------------- 2 files changed, 15 insertions(+), 22 deletions(-) diff --git a/src/tasks/LargeFileUploadTask.ts b/src/tasks/LargeFileUploadTask.ts index b642b6c..2c3a7e3 100644 --- a/src/tasks/LargeFileUploadTask.ts +++ b/src/tasks/LargeFileUploadTask.ts @@ -292,13 +292,17 @@ export class LargeFileUploadTask { const uploadSlices: UploadSlice[] = []; const rangesRemaining = this.rangesRemaining; const session = this.Session; + const uploadUrl = session.uploadUrl; + if (!uploadUrl) { + throw new Error("Upload URL is a required parameter."); + } rangesRemaining.forEach(range => { let currentRangeBegin = range[0]; while (currentRangeBegin <= range[1]) { const nextSliceSize = this.nextSliceSize(currentRangeBegin, range[1]); const uploadRequest = new UploadSlice( this.requestAdapter, - session.uploadUrl!, + uploadUrl, currentRangeBegin, currentRangeBegin + nextSliceSize - 1, range[1] + 1, @@ -335,10 +339,14 @@ export class LargeFileUploadTask { // Sample: ["12345-55232","77829-99375"] // Also, second number in range can be blank, which means 'until the end' const ranges: number[][] = []; - uploadSession.nextExpectedRanges?.forEach(rangeString => { - const rangeArray = rangeString.split("-"); - ranges.push([parseInt(rangeArray[0], 10), parseInt(rangeArray[1], 10)]); - }); + if (uploadSession.nextExpectedRanges) { + ranges.push( + ...uploadSession.nextExpectedRanges.map(rangeString => { + const rangeArray = rangeString.split("-"); + return [parseInt(rangeArray[0], 10), parseInt(rangeArray[1], 10)]; + }), + ); + } return ranges; } } diff --git a/src/tasks/SeekableStreamReader.ts b/src/tasks/SeekableStreamReader.ts index 47ff6a9..dc0c82f 100644 --- a/src/tasks/SeekableStreamReader.ts +++ b/src/tasks/SeekableStreamReader.ts @@ -12,7 +12,7 @@ * This class allows reading specific sections of a stream without seeking backwards. */ export class SeekableStreamReader { - private reader: ReadableStreamDefaultReader; + private readonly reader: ReadableStreamDefaultReader; private cachedChunk?: Uint8Array | null; private cachedPosition = 0; private cachedOffset = 0; // Track where we are within the cached chunk @@ -21,7 +21,7 @@ export class SeekableStreamReader { * Creates an instance of SeekableStreamReader. * @param {ReadableStream} stream - The readable stream to read from. */ - constructor(private stream: ReadableStream) { + constructor(private readonly stream: ReadableStream) { this.reader = stream.getReader(); } @@ -101,19 +101,4 @@ export class SeekableStreamReader { return result.buffer; } - - /** - * Resets the reader with a new stream. - * - * @param {ReadableStream} newStream - The new readable stream to read from. - * @returns {Promise} A promise that resolves when the reader has been reset. - */ - public async reset(newStream: ReadableStream) { - await this.reader.cancel(); // Ensure the old reader is closed before replacing - this.stream = newStream; - this.reader = newStream.getReader(); - this.cachedChunk = null; - this.cachedPosition = 0; - this.cachedOffset = 0; - } } From de7458db391d65f034505e73145e7955f3924723 Mon Sep 17 00:00:00 2001 From: rkodev <43806892+rkodev@users.noreply.github.com> Date: Thu, 27 Feb 2025 17:44:18 +0300 Subject: [PATCH 19/20] refactor slice calls --- src/tasks/LargeFileUploadTask.ts | 28 +++++++++---- src/tasks/UploadSlice.ts | 2 +- test/task/LargeFileUploadTask.ts | 68 +++++++++++++++++++++++++------ test/utils/DummyRequestAdapter.ts | 4 +- 4 files changed, 80 insertions(+), 22 deletions(-) diff --git a/src/tasks/LargeFileUploadTask.ts b/src/tasks/LargeFileUploadTask.ts index 2c3a7e3..e55f589 100644 --- a/src/tasks/LargeFileUploadTask.ts +++ b/src/tasks/LargeFileUploadTask.ts @@ -192,17 +192,31 @@ export class LargeFileUploadTask { * @throws {Error} If the upload fails. */ public async upload(progress?: IProgress): Promise> { - const sliceRequests = this.getUploadSliceRequests(); - for (const request of sliceRequests) { - try { - const uploadResult = await request.uploadSlice(); - progress?.report(request.rangeEnd); + const uploadUrl = this.Session.uploadUrl; + if (!uploadUrl) { + throw new Error("Upload URL is a required parameter."); + } + for (const range of this.rangesRemaining) { + let currentRangeBegin = range[0]; + while (currentRangeBegin <= range[1]) { + const nextSliceSize = this.nextSliceSize(currentRangeBegin, range[1]); + const uploadRequest = new UploadSlice( + this.requestAdapter, + uploadUrl, + currentRangeBegin, + currentRangeBegin + nextSliceSize - 1, + range[1] + 1, + this.parsableFactory, + this.errorMappings, + this.seekableStreamReader, + ); + const uploadResult = await uploadRequest.uploadSlice(); + progress?.report(uploadRequest.rangeEnd); const { itemResponse, location } = uploadResult as Partial>; if (itemResponse || location) { return uploadResult as UploadResult; } - } catch (e) { - console.error(e); + currentRangeBegin += nextSliceSize; } } throw new Error("Upload failed"); diff --git a/src/tasks/UploadSlice.ts b/src/tasks/UploadSlice.ts index b94a5f4..9a8d4fd 100644 --- a/src/tasks/UploadSlice.ts +++ b/src/tasks/UploadSlice.ts @@ -60,7 +60,7 @@ export class UploadSlice { const data = await this.seekableStreamReader.readSection(this.rangeBegin, this.rangeEnd); const requestInformation = new RequestInformation(HttpMethod.PUT, this.sessionUrl); requestInformation.headers = new Headers([ - ["Content-Range", new Set([`bytes ${this.rangeBegin}-${this.rangeEnd - 1}/${this.totalSessionLength}`])], + ["Content-Range", new Set([`bytes ${this.rangeBegin}-${this.rangeEnd}/${this.totalSessionLength}`])], ["Content-Length", new Set([`${this.rangeEnd - this.rangeBegin}`])], ]); requestInformation.setStreamContent(data, binaryContentType); diff --git a/test/task/LargeFileUploadTask.ts b/test/task/LargeFileUploadTask.ts index 3a56725..29faabd 100644 --- a/test/task/LargeFileUploadTask.ts +++ b/test/task/LargeFileUploadTask.ts @@ -104,26 +104,70 @@ describe("LargeFileUploadTask tests", () => { errorMappings, ); - // Accessing the private method using bracket notation - const uploadSlices = (largeFileUploadTask as any)["getUploadSliceRequests"]() as UploadSlice[]; + adapter.setResponse( + { + nextExpectedRanges: ["10-19"], + expirationDateTime: new Date(Date.now() + 24 * 60 * 60 * 1000), + uploadUrl: "https://example.com/upload", + }, + { + nextExpectedRanges: ["10-19"], + expirationDateTime: new Date(Date.now() + 24 * 60 * 60 * 1000), + uploadUrl: "https://example.com/upload", + }, + { + nextExpectedRanges: ["10-19"], + expirationDateTime: new Date(Date.now() + 24 * 60 * 60 * 1000), + uploadUrl: "https://example.com/upload", + }, + { + nextExpectedRanges: ["22-23"], + expirationDateTime: new Date(Date.now() + 24 * 60 * 60 * 1000), + uploadUrl: "https://example.com/upload", + }, + { + name: "Valid", + }, + ); + + await largeFileUploadTask.upload(); + + // get all the request information from the adapter and rebuild the slice start and stop from the header + const requests = adapter.getRequests(); + + // get Content-Range and build start and end byte + const pairs: [number, number][] = requests.map(request => { + const ranges = request.headers.get("Content-Range"); + if (!ranges) { + throw new Error("Content-Range header is missing"); + } + const rangeValue = ranges.values().next().value; + if (!rangeValue) { + throw new Error("Content-Range value is missing"); + } + const range = rangeValue.split(" ")[1].split("/")[0]; + const [start, end] = range.split("-").map(Number); + return [start, end]; + }); + // cast the arrays to UploadSlice - assert.equal(uploadSlices.length, 5); + assert.equal(pairs.length, 5); - uploadSlices.forEach(slice => { - assert.isAtMost(slice.rangeEnd - slice.rangeBegin + 1, 5, "Slice size should not be larger than 5"); + pairs.forEach(slice => { + const [rangeBegin, rangeEnd] = slice; + assert.isAtMost(rangeEnd - rangeBegin + 1, 5, "Slice size should not be larger than 5"); }); - for (let i = 0; i < uploadSlices.length - 1; i++) { - assert.isTrue( - uploadSlices[i].rangeEnd < uploadSlices[i + 1].rangeBegin, - "Slices should be in order from largest to smallest", - ); + + for (let i = 0; i < pairs.length - 1; i++) { + assert.isTrue(pairs[i][1] < pairs[i + 1][0], "Slices should be in order from largest to smallest"); } const decoder = new TextDecoder(); let reconstructedString = ""; const seekableStream = new SeekableStreamReader(createSampleReadableStream()); - for (const slice of uploadSlices) { - const chunk = await seekableStream.readSection(slice.rangeBegin, slice.rangeEnd); + for (let i = 0; i < pairs.length; i++) { + const [rangeBegin, rangeEnd] = pairs[i]; + const chunk = await seekableStream.readSection(rangeBegin, rangeEnd); reconstructedString += decoder.decode(chunk); } assert.equal(reconstructedString, "This is a 24-byte string", "Reconstructed string should match the original"); diff --git a/test/utils/DummyRequestAdapter.ts b/test/utils/DummyRequestAdapter.ts index 6611cbc..f1e1e25 100644 --- a/test/utils/DummyRequestAdapter.ts +++ b/test/utils/DummyRequestAdapter.ts @@ -46,8 +46,8 @@ export class DummyRequestAdapter implements RequestAdapter { } // set a fake response - setResponse(response: any): void { - this.response.push(response); + setResponse(...responses: any): void { + this.response.push(...responses); } // get requests From da49d38a42b425f4a617e173bbd16007989cbb3e Mon Sep 17 00:00:00 2001 From: rkodev <43806892+rkodev@users.noreply.github.com> Date: Thu, 27 Feb 2025 22:50:58 +0300 Subject: [PATCH 20/20] delete unused code --- src/tasks/LargeFileUploadTask.ts | 35 -------------------------------- 1 file changed, 35 deletions(-) diff --git a/src/tasks/LargeFileUploadTask.ts b/src/tasks/LargeFileUploadTask.ts index e55f589..86819c1 100644 --- a/src/tasks/LargeFileUploadTask.ts +++ b/src/tasks/LargeFileUploadTask.ts @@ -296,41 +296,6 @@ export class LargeFileUploadTask { }; } - /** - * Generates a list of upload slice requests based on the remaining ranges to be uploaded. - * - * @private - * @returns {UploadSlice[]} An array of UploadSlice objects representing the upload requests. - */ - private getUploadSliceRequests(): UploadSlice[] { - const uploadSlices: UploadSlice[] = []; - const rangesRemaining = this.rangesRemaining; - const session = this.Session; - const uploadUrl = session.uploadUrl; - if (!uploadUrl) { - throw new Error("Upload URL is a required parameter."); - } - rangesRemaining.forEach(range => { - let currentRangeBegin = range[0]; - while (currentRangeBegin <= range[1]) { - const nextSliceSize = this.nextSliceSize(currentRangeBegin, range[1]); - const uploadRequest = new UploadSlice( - this.requestAdapter, - uploadUrl, - currentRangeBegin, - currentRangeBegin + nextSliceSize - 1, - range[1] + 1, - this.parsableFactory, - this.errorMappings, - this.seekableStreamReader, - ); - uploadSlices.push(uploadRequest); - currentRangeBegin += nextSliceSize; - } - }); - return uploadSlices; - } - /** * Calculates the size of the next slice to be uploaded. *