cleanup

apache · May 13, 2018 · b52af25 · b52af25
1 parent 3187732
commit b52af25
Show file tree

Hide file tree

Showing 3 changed files with 16 additions and 11 deletions.
diff --git a/js/src/ipc/reader/binary.ts b/js/src/ipc/reader/binary.ts
@@ -171,13 +171,18 @@ function readFileSchema(bb: ByteBuffer) {
 
 function readFileMessages(footer: Footer) {
     return function* (bb: ByteBuffer) {
+        let message: RecordBatchMetadata | DictionaryBatch;
         for (let i = -1, batches = footer.dictionaryBatches, n = batches.length; ++i < n;) {
             bb.setPosition(batches[i].offset);
-            yield readMessage(bb, bb.readInt32(bb.position())) as DictionaryBatch;
+            if (message = readMessage(bb, bb.readInt32(bb.position())) as DictionaryBatch) {
+                yield message;
+            }
         }
         for (let i = -1, batches = footer.recordBatches, n = batches.length; ++i < n;) {
             bb.setPosition(batches[i].offset);
-            yield readMessage(bb, bb.readInt32(bb.position())) as RecordBatchMetadata;
+            if (message = readMessage(bb, bb.readInt32(bb.position())) as RecordBatchMetadata) {
+                yield message;
+            }
         }
     };
 }

diff --git a/js/src/ipc/reader/node.ts b/js/src/ipc/reader/node.ts
@@ -29,12 +29,12 @@ export async function* fromNodeStream(stream: NodeJS.ReadableStream) {
 
     for await (let chunk of (stream as any as AsyncIterable<Uint8Array | Buffer | string>)) {
 
-        const grown = new Uint8Array(bytes.length + chunk.length);
+        const grown = new Uint8Array(bytes.byteLength + chunk.length);
 
         if (typeof chunk !== 'string') {
-            grown.set(bytes, 0) || grown.set(chunk, bytes.length);
+            grown.set(bytes, 0) || grown.set(chunk, bytes.byteLength);
         } else {
-            for (let i = -1, j = bytes.length, n = chunk.length; ++i < n;) {
+            for (let i = -1, j = bytes.byteLength, n = chunk.length; ++i < n;) {
                 grown[i + j] = chunk.charCodeAt(i);
             }
         }
@@ -45,7 +45,7 @@ export async function* fromNodeStream(stream: NodeJS.ReadableStream) {
             messageLength = new DataView(bytes.buffer).getInt32(0, true);
         }
 
-        while (messageLength < bytes.length) {
+        while (messageLength < bytes.byteLength) {
             if (!message) {
                 (bb = new ByteBuffer(bytes)).setPosition(4);
                 if (message = _Message.getRootAsMessage(bb)) {
@@ -55,9 +55,9 @@ export async function* fromNodeStream(stream: NodeJS.ReadableStream) {
                 throw new Error(`Invalid message at position ${bytesRead}`);
             }
             bytesRead += messageLength + PADDING;
-            yield bytes.subarray(0, messageLength);
+            yield bytes.subarray(0, messageLength + PADDING);
             bytes = bytes.subarray(messageLength + PADDING);
-            messageLength = bytes.length <= 0 ? 0 :
+            messageLength = bytes.byteLength <= 0 ? 0 :
                 new DataView(bytes.buffer).getInt32(bytes.byteOffset, true);
             message = null;
         }

diff --git a/js/test/integration/validate-tests.ts b/js/test/integration/validate-tests.ts
@@ -120,8 +120,8 @@ describe(`Integration`, () => {
             testReaderIntegration(json, arrowBuffer);
             testTableFromBuffersIntegration(json, arrowBuffer);
             testTableToBuffersIntegration('json', 'file')(json, arrowBuffer);
-            testTableToBuffersIntegration('json', 'stream')(json, arrowBuffer);
             testTableToBuffersIntegration('binary', 'file')(json, arrowBuffer);
+            testTableToBuffersIntegration('json', 'stream')(json, arrowBuffer);
             testTableToBuffersIntegration('binary', 'stream')(json, arrowBuffer);
         });
     }
@@ -164,10 +164,10 @@ function testTableFromBuffersIntegration(jsonData: any, arrowBuffer: Uint8Array)
 }
 
 function testTableToBuffersIntegration(srcFormat: 'json' | 'binary', arrowFormat: 'stream' | 'file') {
+    const refFormat = srcFormat === `json` ? `binary` : `json`;
     return function testTableToBuffersIntegration(jsonData: any, arrowBuffer: Uint8Array) {
-        test(`serializing ${srcFormat} to a ${arrowFormat} reports the same values as the alternate format`, () => {
+        test(`serialized ${srcFormat} ${arrowFormat} reports the same values as the ${refFormat} ${arrowFormat}`, () => {
             expect.hasAssertions();
-            const refFormat = srcFormat === 'json' ? `binary` : `json`;
             const refTable = Table.from(refFormat === `json` ? jsonData : arrowBuffer);
             const srcTable = Table.from(srcFormat === `json` ? jsonData : arrowBuffer);
             const dstTable = Table.from(srcTable.serialize(`binary`, arrowFormat === `stream`));