fix: addressing PR feedback

Author: alexforsyth

lib/storage/src/Upload.ts

@@ -58,19 +58,11 @@ export class Upload extends EventEmitter {
     this.leavePartsOnError = options.leavePartsOnError || this.leavePartsOnError;
     this.tags = options.tags || this.tags;
 
-    if (this.partSize < MIN_PART_SIZE) {
-      throw new Error(
-        `EntityTooSmall: Your proposed upload partsize [${this.partSize}] is smaller than the minimum allowed size [${MIN_PART_SIZE}] (5MB)`
-      );
-    }
-
-    if (this.queueSize < 1) {
-      throw new Error(`Queue size: Must have atleast one uploading queue.`);
-    }
-
     this.client = options.client;
     this.params = options.params;
 
+    this.__validateInput();
+    
     // set progress defaults
     this.totalBytes = byteLength(this.params.Body);
     this.bytesUploadedSoFar = 0;
@@ -94,7 +86,7 @@ export class Upload extends EventEmitter {
     super.on(event, listener);
   }
 
-  async __doConcurrentUpload(dataFeeder: AsyncGenerator<RawDataPart>): Promise<void> {
+  async __doConcurrentUpload(dataFeeder: AsyncGenerator<RawDataPart, void, undefined>): Promise<void> {
     for await (const dataPart of dataFeeder) {
       if (this.uploadedParts.length > this.MAX_PARTS) {
         throw new Error(
@@ -199,4 +191,24 @@ export class Upload extends EventEmitter {
       };
     });
   }
+
+  __validateInput() {
+    if (!this.params) {
+      throw new Error(`InputError: Upload requires params to be passed to upload.`);
+    }
+
+    if (!this.client) {
+      throw new Error(`InputError: Upload requires a AWS client to do uploads with.`);
+    }
+
+    if (this.partSize < MIN_PART_SIZE) {
+      throw new Error(
+        `EntityTooSmall: Your proposed upload partsize [${this.partSize}] is smaller than the minimum allowed size [${MIN_PART_SIZE}] (5MB)`
+      );
+    }
+
+    if (this.queueSize < 1) {
+      throw new Error(`Queue size: Must have at least one uploading queue.`);
+    }
+  }
 }

lib/storage/src/chunker.ts

@@ -2,25 +2,25 @@ import stream from "stream";
 import { Buffer } from "buffer";
 
 import { BodyDataTypes } from "./types";
-import { getChunk as getChunkBuffer } from "./chunks/getChunkBuffer";
-import { getChunk as getChunkStream } from "./chunks/getChunkStream";
-import { getNextData as getNextDataReadableStream } from "./chunks/getDataReadableStream";
-import { getNextData as getNextDataReadable } from "./chunks/getDataReadable";
+import { getChunkBuffer } from "./chunks/getChunkBuffer";
+import { getChunkStream } from "./chunks/getChunkStream";
+import { getDataReadableStream } from "./chunks/getDataReadableStream";
+import { getDataReadable } from "./chunks/getDataReadable";
 
 export const getChunk = (data: BodyDataTypes, partSize: number) => {
   if (data instanceof Buffer) {
     return getChunkBuffer(data, partSize);
   } else if (data instanceof stream.Readable) {
-    return getChunkStream<stream.Readable>(data, partSize, getNextDataReadable);
-  } else if (data instanceof String || typeof data == "string" || data instanceof Uint8Array) {
+    return getChunkStream<stream.Readable>(data, partSize, getDataReadable);
+  } else if (data instanceof String || typeof data === "string" || data instanceof Uint8Array) {
     // chunk Strings, Uint8Array.
     return getChunkBuffer(Buffer.from(data), partSize);
   }
   if (typeof (data as any).stream === "function") {
     // approximate support for Blobs.
-    return getChunkStream<ReadableStream>((data as any).stream(), partSize, getNextDataReadableStream);
+    return getChunkStream<ReadableStream>((data as any).stream(), partSize, getDataReadableStream);
   } else if (data instanceof ReadableStream) {
-    return getChunkStream<ReadableStream>(data, partSize, getNextDataReadableStream);
+    return getChunkStream<ReadableStream>(data, partSize, getDataReadableStream);
   } else {
     throw new Error(
       "Body Data is unsupported format, expected data to be one of: string | Uint8Array | Buffer | Readable | ReadableStream | Blob;."

lib/storage/src/chunks/getChunkBuffer.ts

@@ -1,6 +1,6 @@
 import { RawDataPart } from "../Upload";
 
-export async function* getChunk(data: Buffer, partSize: number): AsyncGenerator<RawDataPart> {
+export async function* getChunkBuffer(data: Buffer, partSize: number): AsyncGenerator<RawDataPart, void, undefined> {
   let partNumber = 1;
   let startByte = 0;
   let endByte = startByte + partSize;

lib/storage/src/chunks/getChunkStream.ts

@@ -6,11 +6,11 @@ interface Buffers {
   length: number;
 }
 
-export async function* getChunk<T>(
+export async function* getChunkStream<T>(
   data: T,
   partSize: number,
   getNextData: (data: T) => AsyncGenerator<Buffer>
-): AsyncGenerator<RawDataPart> {
+): AsyncGenerator<RawDataPart, void, undefined> {
   let partNumber = 1;
   const currentBuffer: Buffers = { chunks: [], length: 0 };
 
@@ -19,9 +19,13 @@ export async function* getChunk<T>(
     currentBuffer.length += datum.length;
 
     while (currentBuffer.length >= partSize) {
-      // Concat all the buffers together once.
       let dataChunk: Buffer = currentBuffer.chunks[0];
-      if(currentBuffer.chunks.length) {
+      
+      /**
+       * Concat all the buffers together once if there is more than one to concat. Attempt 
+       * to minimize concats as Buffer.Concat is an extremely expensive operation.
+       */
+      if (currentBuffer.chunks.length > 1) {
         dataChunk = Buffer.concat(currentBuffer.chunks);
       }
 

lib/storage/src/chunks/getDataReadable.ts

@@ -1,7 +1,7 @@
 import Stream from "stream";
 import { Buffer } from "buffer";
 
-export async function* getNextData(data: Stream.Readable): AsyncGenerator<Buffer> {
+export async function* getDataReadable(data: Stream.Readable): AsyncGenerator<Buffer> {
   for await (const chunk of data) {
     yield Buffer.from(chunk);
   }

lib/storage/src/chunks/getDataReadableStream.ts

@@ -1,6 +1,6 @@
 import { Buffer } from "buffer";
 
-export async function* getNextData(data: ReadableStream): AsyncGenerator<Buffer> {
+export async function* getDataReadableStream(data: ReadableStream): AsyncGenerator<Buffer> {
   // Get a lock on the stream.
   const reader = data.getReader();
 
@@ -13,6 +13,8 @@ export async function* getNextData(data: ReadableStream): AsyncGenerator<Buffer>
       // Else yield the chunk.
       yield Buffer.from(value);
     }
+  } catch(e) {
+    throw e;
   } finally {
     // release the lock for reading from this stream.
     reader.releaseLock();

lib/storage/test/data-chunk/buffer-chunk.spec.ts

@@ -1,15 +1,15 @@
 import { byteLength } from "../../src/bytelength";
-import { getChunk as getChunkFromBuffer } from "../../src/chunks/getChunkBuffer";
+import { getChunkBuffer } from "../../src/chunks/getChunkBuffer";
 
-describe.only(getChunkFromBuffer.name, () => {
+describe.only(getChunkBuffer.name, () => {
   const getBuffer = (size: number) => Buffer.from("#".repeat(size));
 
   describe("Buffer chunking", () => {
     it("should come back with small sub buffers", async (done) => {
       const chunklength = 100;
       const totalLength = 1000;
       const buffer = getBuffer(totalLength);
-      const chunker = getChunkFromBuffer(buffer, chunklength);
+      const chunker = getChunkBuffer(buffer, chunklength);
 
       let chunkNum = 0;
       for await (const chunk of chunker) {
@@ -27,7 +27,7 @@ describe.only(getChunkFromBuffer.name, () => {
       const totalLength = 2200;
       const buffer = getBuffer(totalLength);
 
-      const chunker = getChunkFromBuffer(buffer, chunklength);
+      const chunker = getChunkBuffer(buffer, chunklength);
       const chunks = [];
       for await (const chunk of chunker) {
         chunks.push(chunk);
@@ -45,7 +45,7 @@ describe.only(getChunkFromBuffer.name, () => {
       const totalLength = 200;
       const buffer = getBuffer(totalLength);
 
-      const chunker = getChunkFromBuffer(buffer, chunklength);
+      const chunker = getChunkBuffer(buffer, chunklength);
       const chunks = [];
       for await (const chunk of chunker) {
         chunks.push(chunk);

lib/storage/test/data-chunk/readable-chunk.spec.ts

@@ -1,7 +1,7 @@
 import { Readable } from "stream";
 import { byteLength } from "../../src/bytelength";
-import { getChunk as chunkFromReadable } from "../../src/chunks/getChunkStream";
-import { getNextData } from "../../src/chunks/getDataReadable";
+import { getChunkStream as chunkFromReadable } from "../../src/chunks/getChunkStream";
+import { getDataReadable } from "../../src/chunks/getDataReadable";
 import { RawDataPart as DataPart } from "../../src/Upload";
 
 const fs = require("fs");
@@ -23,7 +23,7 @@ describe(chunkFromReadable.name, () => {
   ): Promise<DataPart[]> => {
     const stream = Readable.from(getUnknownEnding(streamYieldSize, streamYieldCount));
     const chunks: DataPart[] = [];
-    const chunker = chunkFromReadable<Readable>(stream, partsize, getNextData);
+    const chunker = chunkFromReadable<Readable>(stream, partsize, getDataReadable);
 
     for await (const chunk of chunker) {
       chunks.push(chunk);
@@ -49,7 +49,7 @@ describe(chunkFromReadable.name, () => {
   it("should properly chunk a file", async (done) => {
     const fileStream = fs.createReadStream(__dirname + "/sample.file");
     const chunks: DataPart[] = [];
-    const chunker = chunkFromReadable<Readable>(fileStream, _6MB, getNextData);
+    const chunker = chunkFromReadable<Readable>(fileStream, _6MB, getDataReadable);
     for await (const chunk of chunker) {
       chunks.push(chunk);
     }

lib/storage/test/data-chunk/readable-stream-chunk.spec.ts

@@ -2,8 +2,8 @@
 import { ReadableStream } from "web-streams-polyfill";
 
 import { byteLength } from "../../src/bytelength";
-import { getChunk as chunkFromReadable } from "../../src/chunks/getChunkStream";
-import { getNextData } from "../../src/chunks/getDataReadableStream";
+import { getChunkStream as chunkFromReadable } from "../../src/chunks/getChunkStream";
+import { getDataReadableStream } from "../../src/chunks/getDataReadableStream";
 import { RawDataPart as DataPart } from "../../src/Upload";
 
 describe("chunkFromReadable.name", () => {
@@ -27,7 +27,7 @@ describe("chunkFromReadable.name", () => {
   ): Promise<DataPart[]> => {
     const stream = getStreamOfUnknownlength(streamYieldSize, streamYieldCount);
     const chunks: DataPart[] = [];
-    const chunker = chunkFromReadable<ReadableStream>(stream, partsize, getNextData);
+    const chunker = chunkFromReadable<ReadableStream>(stream, partsize, getDataReadableStream);
 
     for await (const chunk of chunker) {
       chunks.push(chunk);