Trial of moving the packet parsing straight into the input data handler

Author: gfwilliams

core/serial.js

@@ -51,6 +51,12 @@ To add a new serial device, you must add an object to
   var flowControlXOFF = false;
   /// Set up when flow control received - if no response is received we start sending anyway
   var flowControlTimeout;
+  /// used by rxDataHandler - last received character
+  let rxDataHandlerLastCh = 0;
+  /// used by rxDataHandler - used for parsing
+  let rxDataHandlerPacket = undefined;
+  /// timeout for unfinished packet
+  let rxDataHandlerTimeout = undefined;
 
 
   function init() {
@@ -159,6 +165,88 @@ To add a new serial device, you must add an object to
     return openSerialInternal(serialPort, connectCallback, disconnectCallback, 5);
   }
 
+  var rxDataHandler = function(data) {
+    if (!(data instanceof ArrayBuffer)) console.warn("Serial port implementation is not returning ArrayBuffers")
+
+    // Filter incoming data to handle and remove control characters
+    const filteredData = new Uint8Array(data).filter((ch) => {
+      let lastCh = rxDataHandlerLastCh;
+      rxDataHandlerLastCh = ch;
+
+      if (rxDataHandlerPacket!==undefined) {
+        rxDataHandlerPacket.push(ch);
+        let flags = (rxDataHandlerPacket[0]<<8) | rxDataHandlerPacket[1];
+        let len = flags & 0x1FFF;
+        let rxLen = rxDataHandlerPacket.length;
+        if (rxLen>=2 && rxLen>=(len+2)) {
+          console.log("Got packet end");
+          if (rxDataHandlerTimeout) {
+            clearTimeout(rxDataHandlerTimeout);
+            rxDataHandlerTimeout = undefined;
+          }
+          emit("packet", flags&0xE000, new Uint8Array(rxDataHandlerPacket.slice(2)));
+          rxDataHandlerPacket = undefined; // stop packet reception
+        }
+        return false; // don't pass data in packet through
+      } else switch (ch) {
+        case 16: // DLE - potential start of packet (ignore)
+          return false;
+
+        case 1: // SOH - start of packet if after DLE
+          if (lastCh == 16/*DLE*/) {
+            console.log("Packet start received");
+            rxDataHandlerPacket = [];
+            rxDataHandlerTimeout = setTimeout(()=>{
+              rxDataHandlerTimeout = undefined;
+              console.log(`Packet timeout (2s - contents ${JSON.stringify(rxDataHandlerPacket)})`);
+              // TODO: emit packet error?
+              rxDataHandlerPacket = undefined;
+            }, 2000);
+            return false;
+          } else
+            return true; // else pass through
+
+        case 17: // XON
+          if (Espruino.Config.SERIAL_FLOW_CONTROL && rxDataHandlerPacket===undefined) {
+            console.log("XON received => resume upload")
+            flowControlXOFF = false
+            if (flowControlTimeout) {
+              clearTimeout(flowControlTimeout)
+              flowControlTimeout = undefined
+            }
+          }
+          return false;
+
+        case 19: // XOFF
+          if (Espruino.Config.SERIAL_FLOW_CONTROL && rxDataHandlerPacket===undefined) {
+            console.log("XOFF received => pause upload")
+            flowControlXOFF = true
+            if (flowControlTimeout) clearTimeout(flowControlTimeout)
+            flowControlTimeout = setTimeout(function () {
+              console.log(
+                `XOFF timeout (${FLOW_CONTROL_RESUME_TIMEOUT}s) => resume upload anyway`
+              )
+              flowControlXOFF = false
+              flowControlTimeout = undefined
+            }, FLOW_CONTROL_RESUME_TIMEOUT)
+          }
+          return false;
+
+        case 6: // ACK
+          emit("ack")
+          return false;
+
+        case 21: // NACK
+          emit("nack")
+          return false;
+      }
+
+      return true;
+    })
+
+    if (readListener) readListener(filteredData.buffer)
+  };
+
   var openSerialInternal=function(serialPort, connectCallback, disconnectCallback, attempts) {
     /* If openSerial is called, we need to have called getPorts first
       in order to figure out which one of the serial_ implementations
@@ -214,52 +302,8 @@ To add a new serial device, you must add an object to
           connectCallback = undefined;
         });
       }
-    }, function(data) { // RECEIEVE DATA
-      if (!(data instanceof ArrayBuffer)) console.warn("Serial port implementation is not returning ArrayBuffers")
-
-      // Filter incoming data to handle and remove control characters
-      const filteredData = new Uint8Array(data).filter((v) => {
-        switch (v) {
-          case 17: // XON
-            if (Espruino.Config.SERIAL_FLOW_CONTROL) {
-              console.log("XON received => resume upload")
-              flowControlXOFF = false
-              if (flowControlTimeout) {
-                clearTimeout(flowControlTimeout)
-                flowControlTimeout = undefined
-              }
-            }
-            return false
-
-          case 19: // XOFF
-            if (Espruino.Config.SERIAL_FLOW_CONTROL) {
-              console.log("XOFF received => pause upload")
-              flowControlXOFF = true
-              if (flowControlTimeout) clearTimeout(flowControlTimeout)
-              flowControlTimeout = setTimeout(function () {
-                console.log(
-                  `XOFF timeout (${FLOW_CONTROL_RESUME_TIMEOUT}s) => resume upload anyway`
-                )
-                flowControlXOFF = false
-                flowControlTimeout = undefined
-              }, FLOW_CONTROL_RESUME_TIMEOUT)
-            }
-            return false
-
-          case 6: // ACK
-            emit("ack")
-            return false
-
-          case 21: // NACK
-            emit("nack")
-            return false
-        }
-
-        return true
-      })
-
-      if (readListener) readListener(filteredData.buffer)
-    }, function(error) { // DISCONNECT
+    }, rxDataHandler, // RECEIEVE DATA
+    function(error) { // DISCONNECT
       currentDevice = undefined;
       if (writeTimeout!==undefined)
         clearTimeout(writeTimeout);

core/utils.js

@@ -512,57 +512,6 @@
     ) + data;               // Data blob
   }
 
-  /**
-   * Take an input buffer and look for the initial control characters and then attempt to parse a 
-   * complete data packet from the buffer. Any complete packet is sent via `emit("packet")` and then
-   * stripped from `buffer` modifiying it.
-   * @param {Uint8Array} buffer 
-   * @returns {Uint8Array}
-   */
-  function parsePacketsFromBuffer(buffer) {
-
-    // Find DLE
-    const dle = buffer.findIndex(v => v === 0x10)
-    if (dle < 0) return buffer
-
-    // Check for SOH
-    if (buffer.at(dle + 1) !== 0x1) {
-      // console.warn("DLE not followed by SOH")
-      // TODO: Not stripping out this invalid control will cause a loop
-      buffer.set([undefined], dle) // Remove this DLE
-      return buffer
-    }
-
-    // Check there's still space for headers
-    if (buffer.at(dle + 2) === undefined || buffer.at(dle + 3) === undefined) {
-      console.warn("NO SPACE FOR HEADERS")
-      return buffer
-    }
-    const upper = buffer.at(dle + 2)
-    const lower = buffer.at(dle + 3)
-
-    // Parse heading from 2 bytes after control headers
-    const heading = new Number(upper << 8) | new Number(lower)
-    const pkLen = heading & 0x1FFF
-    const pkTyp = heading & 0xE000
-
-    // Ignoring heading bytes, check if there's enough bytes in the buffer to satisfy pkLen
-    if (buffer.length < dle + 4 + pkLen) {
-      return buffer
-    }
-
-    // Pick out a packet from the buffer and emit it via the event handler
-    const packet = buffer.subarray(dle, dle + 4 + pkLen)
-    console.log("Packet recieved... type:", pkTyp, "length:", pkLen)
-    Espruino.Core.Serial.emit('packet', pkTyp, packet.subarray(4, packet.length))
-
-    // Fill the buffer region of the packet that was sent with undefined
-    buffer.fill(undefined, 0, dle + packet.length)
-
-    // Return the input buffer but with the stripped packet filtered out
-    return buffer.filter(v => v !== undefined)
-  }
-
   /**
    * Send a packet
    * @param {number} pkType 
@@ -584,23 +533,23 @@
 
     let allData
     function onPacket(rxPkType, data) {
-      tidy()
-      const packetData = String.fromCharCode(...data)
-
+      const packetData = String.fromCharCode(...data);
       // TODO: Depending on the rx type and tx type match up packet types, wait for x number of data
       if (pkTypes[pkType] === pkTypes.EVAL && rxPkType === pkTypes.RESPONSE) {
+        tidy();
         callback(packetData)
-
-      // If the packet type is data, we need to wait for the 0 length `DATA` packet and then send all of the data joined together
+        // If the packet type is data, we need to wait for the 0 length `DATA` packet and then send all of the data joined together
       } else if (pkTypes[pkType] === pkTypes.FILE_RECV && rxPkType === pkTypes.DATA) {
         if (data.length === 0) {
-          callback(allData)
-          console.log("zero packet")
+          tidy();
+          console.log("zero packet");
+          callback(allData);
         } else {
           console.log("appending data", String.fromCharCode(...data))
           allData += String.fromCharCode(...data)
         }
       }else {
+        tidy();
         callback("nodata")
       }
     }
@@ -612,31 +561,13 @@
       Espruino.Core.Serial.removeListener("packet", onPacket)
     }
 
-    // Attach event handlers for this packet event
-    Espruino.Core.Serial.on("ack", onAck)
-    Espruino.Core.Serial.on("nack", onNack)
-    Espruino.Core.Serial.on("packet", onPacket)
-
     // Write packet to serial port
     Espruino.Core.Serial.write(createPacket(pkType, data), undefined, function () {
-      // TODO: Add 1 sec timeout
-
-      let dataBuffer = new Uint8Array()
-
-      // Each time data comes in, expand the buffer and add the new data to it
-      // TODO: This seems problematic if there are subsequent/concurrent calls
-      Espruino.Core.Serial.startListening((data) => {
-        const newBuffer = new Uint8Array(data)
-
-        const tempBuffer = new Uint8Array(dataBuffer.length + newBuffer.length)
-        tempBuffer.set(dataBuffer, 0)
-        tempBuffer.set(newBuffer, dataBuffer.length)
-
-        dataBuffer = tempBuffer
-
-        // Now we've added more data to the buffer, try to parse out any packets
-        dataBuffer = parsePacketsFromBuffer(dataBuffer)
-      })
+      // Attach event handlers for this packet event
+      Espruino.Core.Serial.on("ack", onAck);
+      Espruino.Core.Serial.on("nack", onNack);
+      Espruino.Core.Serial.on("packet", onPacket);
+      // TODO: Timeout handling?
     })
   }