refactor: remove callback nesting from done token parser

benchmarks/token-parser/done-token.js

@@ -1,21 +1,36 @@
 const { createBenchmark } = require('../common');
 
 const { Parser } = require('../../src/token/token-stream-parser');
+const { AzurePowerShellCredential } = require('@azure/identity');
 
 const bench = createBenchmark(main, {
-  n: [10, 100, 1000],
-  tokenCount: [10, 100, 1000, 10000]
+  n: [100, 1000],
+  tokenCount: [10, 100, 1000, 10000],
+  packetLength: [512, 4096, 32767]
 });
 
-async function * repeat(data, n) {
+/**
+ * @param {Buffer} data
+ * @param {number} n
+ * @param {number} chunkSize
+ */
+async function* repeat(data, n, chunkSize) {
   for (let i = 0; i < n; i++) {
-    yield data;
+    let offset = 0
+
+    while (offset + chunkSize <= data.length) {
+      yield data.slice(offset, offset += chunkSize);
+    }
+
+    if (offset < data.length) {
+      yield data.slice(offset);
+    }
   }
 }
 
-function main({ n, tokenCount }) {
+function main({ n, tokenCount, packetLength }) {
   const data = Buffer.from('FE0000E0000000000000000000'.repeat(tokenCount), 'hex');
-  const parser = new Parser(repeat(data, n), { token: function() { } }, {}, {});
+  const parser = new Parser(repeat(data, n, packetLength), { token: function() { } }, { onDoneProc: (token) => {} }, {});
 
   bench.start();
 

src/token/done-token-parser.ts

@@ -25,51 +25,131 @@ interface TokenData {
   curCmd: number;
 }
 
-function parseToken(parser: Parser, options: InternalConnectionOptions, callback: (data: TokenData) => void) {
-  parser.readUInt16LE((status) => {
-    const more = !!(status & STATUS.MORE);
-    const sqlError = !!(status & STATUS.ERROR);
-    const rowCountValid = !!(status & STATUS.COUNT);
-    const attention = !!(status & STATUS.ATTN);
-    const serverError = !!(status & STATUS.SRVERROR);
-
-    parser.readUInt16LE((curCmd) => {
-      const next = (rowCount: number) => {
-        callback({
-          more: more,
-          sqlError: sqlError,
-          attention: attention,
-          serverError: serverError,
-          rowCount: rowCountValid ? rowCount : undefined,
-          curCmd: curCmd
-        });
-      };
-
-      if (options.tdsVersion < '7_2') {
-        parser.readUInt32LE(next);
-      } else {
-        parser.readBigUInt64LE((rowCount) => {
-          next(JSBI.toNumber(rowCount));
-        });
-      }
-    });
-  });
+class NotEnoughDataError extends Error {}
+
+function readBigUInt64LE(buffer: Buffer, offset: number) {
+  return JSBI.add(
+    JSBI.leftShift(
+      JSBI.BigInt(
+        buffer[offset + 4] +
+        buffer[offset + 5] * 2 ** 8 +
+        buffer[offset + 6] * 2 ** 16 +
+        (buffer[offset + 7] << 24) // Overflow
+      ),
+      JSBI.BigInt(32)
+    ),
+    JSBI.BigInt(
+      buffer[offset] +
+      buffer[offset + 1] * 2 ** 8 +
+      buffer[offset + 2] * 2 ** 16 +
+      buffer[offset + 3] * 2 ** 24
+    )
+  );
+}
+
+function parseToken(parser: Parser, options: InternalConnectionOptions) {
+  const buffer = parser.buffer;
+  let offset = parser.position;
+
+  if (buffer.length < offset + 2) {
+    throw new NotEnoughDataError();
+  }
+  const status = buffer.readUInt16LE(offset);
+  offset += 2;
+
+  if (buffer.length < offset + 2) {
+    throw new NotEnoughDataError();
+  }
+  const curCmd = buffer.readUInt16LE(offset);
+  offset += 2;
+
+  const more = !!(status & STATUS.MORE);
+  const sqlError = !!(status & STATUS.ERROR);
+  const rowCountValid = !!(status & STATUS.COUNT);
+  const attention = !!(status & STATUS.ATTN);
+  const serverError = !!(status & STATUS.SRVERROR);
+
+  let rowCount: number;
+  if (options.tdsVersion < '7_2') {
+    if (buffer.length < offset + 4) {
+      throw new NotEnoughDataError();
+    }
+
+    rowCount = buffer.readUInt32LE(offset);
+    offset += 4;
+  } else {
+    if (buffer.length < offset + 8) {
+      throw new NotEnoughDataError();
+    }
+
+    const bigRowCount = readBigUInt64LE(buffer, offset);
+    offset += 8;
+
+    rowCount = JSBI.toNumber(bigRowCount);
+  }
+
+  parser.position = offset;
+
+  return {
+    more: more,
+    sqlError: sqlError,
+    attention: attention,
+    serverError: serverError,
+    rowCount: rowCountValid ? rowCount : undefined,
+    curCmd: curCmd
+  } as TokenData;
 }
 
 export function doneParser(parser: Parser, options: InternalConnectionOptions, callback: (token: DoneToken) => void) {
-  parseToken(parser, options, (data) => {
-    callback(new DoneToken(data));
-  });
+  let data;
+
+  try {
+    data = parseToken(parser, options);
+  } catch (err) {
+    if (err instanceof NotEnoughDataError) {
+      return parser.suspend(() => {
+        doneParser(parser, options, callback);
+      });
+    }
+
+    throw err;
+  }
+
+  callback(new DoneToken(data));
 }
 
 export function doneInProcParser(parser: Parser, options: InternalConnectionOptions, callback: (token: DoneInProcToken) => void) {
-  parseToken(parser, options, (data) => {
-    callback(new DoneInProcToken(data));
-  });
+  let data;
+
+  try {
+    data = parseToken(parser, options);
+  } catch (err) {
+    if (err instanceof NotEnoughDataError) {
+      return parser.suspend(() => {
+        doneInProcParser(parser, options, callback);
+      });
+    }
+
+    throw err;
+  }
+
+  callback(new DoneInProcToken(data));
 }
 
 export function doneProcParser(parser: Parser, options: InternalConnectionOptions, callback: (token: DoneProcToken) => void) {
-  parseToken(parser, options, (data) => {
-    callback(new DoneProcToken(data));
-  });
+  let data;
+
+  try {
+    data = parseToken(parser, options);
+  } catch (err) {
+    if (err instanceof NotEnoughDataError) {
+      return parser.suspend(() => {
+        doneProcParser(parser, options, callback);
+      });
+    }
+
+    throw err;
+  }
+
+  callback(new DoneProcToken(data));
 }