refactor: remove callback nesting from done token parser

tediousjs · Jun 19, 2022 · d0c2768 · d0c2768
1 parent 7ebd236
commit d0c2768
Show file tree

Hide file tree

Showing 2 changed files with 119 additions and 39 deletions.
diff --git a/benchmarks/token-parser/done-token.js b/benchmarks/token-parser/done-token.js
@@ -15,7 +15,7 @@ async function * repeat(data, n) {
 
 function main({ n, tokenCount }) {
   const data = Buffer.from('FE0000E0000000000000000000'.repeat(tokenCount), 'hex');
-  const parser = new Parser(repeat(data, n), { token: function() { } }, {}, {});
+  const parser = new Parser(repeat(data, n), { token: function() { } }, { onDoneProc: () => {} }, {});
 
   bench.start();
 

diff --git a/src/token/done-token-parser.ts b/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 < parser.position + 2) {
+    throw new NotEnoughDataError();
+  }
+  const status = buffer.readUInt16LE(offset);
+  offset += 2;
+
+  if (buffer.length < parser.position + 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 < parser.position + 4) {
+      throw new NotEnoughDataError();
+    }
+
+    rowCount = buffer.readUInt32LE(offset);
+    offset += 4;
+  } else {
+    if (buffer.length < parser.position + 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));
 }