Fix splitting BSON objects

Author: simolus3

packages/common/src/utils/stream_transform.ts

@@ -212,7 +212,7 @@ export function extractBsonObjects(source: SimpleAsyncIterator<Uint8Array>): Sim
             // We're in the middle of reading a BSON document.
             const bytesToRead = Math.min(availableInData, remainingLength);
             const copySource = new Uint8Array(chunk.buffer, chunk.byteOffset + i, bytesToRead);
-            objectBody.set(copySource, 4 + copySource.length - remainingLength);
+            objectBody.set(copySource, objectBody.length - remainingLength);
             i += bytesToRead;
             remainingLength -= bytesToRead;
 
@@ -242,10 +242,7 @@ export function extractBsonObjects(source: SimpleAsyncIterator<Uint8Array>): Sim
               }
 
               objectBody = new Uint8Array(length);
-              for (let j = 0; j < 4; j++) {
-                objectBody[j] = lengthBuffer.getUint8(j);
-                lengthBuffer.setUint8(j, 0);
-              }
+              new DataView(objectBody.buffer).setInt32(0, length, true);
             }
           }
         }

packages/common/tests/utils/stream_transform.test.ts

@@ -117,6 +117,24 @@ describe('bson objects', () => {
     expect(await stream.next()).toStrictEqual(doneResult);
   });
 
+  test('splits bson objects (2)', async () => {
+    async function* source() {
+      yield new Uint8Array([5, 0, 0, 0, 1]);
+
+      yield new Uint8Array([6, 0, 0, 0, 2]);
+      yield new Uint8Array([3]);
+    }
+
+    const stream = extractBsonObjects(source());
+    const a = (await stream.next()).value;
+    const b = (await stream.next()).value;
+    expect(a).toHaveLength(5);
+    expect(b).toHaveLength(6);
+    expect(Array.from(a)).toEqual([5, 0, 0, 0, 1]);
+    expect(Array.from(b)).toEqual([6, 0, 0, 0, 2, 3]);
+    expect(await stream.next()).toStrictEqual(doneResult);
+  });
+
   test('invalid bson size', async () => {
     async function* source() {
       yield new Uint8Array([3, 0, 0, 0]);