StreamID previously could be 0

Before streaming was implemented the field used for streamID today used to be 0 always
So streamID == 0 was made a special case for backward compatibility
But raftjs in some situations still sent packets with streamID == 0 and this is fixed now

Author: robdobsn

src/RaftConnector.ts

@@ -14,7 +14,7 @@ import RaftChannelWebSocket from "./RaftChannelWebSocket";
 import RaftChannelWebSerial from "./RaftChannelWebSerial";
 import RaftChannelSimulated from "./RaftChannelSimulated";
 import RaftCommsStats from "./RaftCommsStats";
-import { RaftEventFn, RaftOKFail, RaftFileSendType, RaftFileDownloadResult, RaftProgressCBType, RaftStreamDataProgressCBType, RaftBridgeSetupResp, RaftFileDownloadFn, RaftReportMsg } from "./RaftTypes";
+import { RaftEventFn, RaftOKFail, RaftFileSendType, RaftFileDownloadResult, RaftProgressCBType, RaftStreamDataProgressCBType, RaftBridgeSetupResp, RaftFileDownloadFn, RaftReportMsg, RaftRtStreamDataCBType, RaftRtStreamHandle, RaftRtStreamOptions, RaftRtStreamStartResp } from "./RaftTypes";
 import RaftSystemUtils from "./RaftSystemUtils";
 import RaftFileHandler from "./RaftFileHandler";
 import RaftStreamHandler from "./RaftStreamHandler";
@@ -87,6 +87,10 @@ export default class RaftConnector {
   // Update manager
   private _raftUpdateManager: RaftUpdateManager | null = null;
 
+  // Open-ended RT stream callbacks keyed by streamID
+  private _rtStreamCallbacks = new Map<number, RaftRtStreamDataCBType>();
+  private _fallbackRtStreamCallback: { streamID: number, callback: RaftRtStreamDataCBType } | null = null;
+
   /**
    * RaftConnector constructor
    * @param getSystemTypeCB - callback to get system type
@@ -193,6 +197,22 @@ export default class RaftConnector {
     return this._commsStats;
   }
 
+  /**
+   * getOperationQueueDepth
+   * @returns number of high-level device operations queued or running
+   */
+  getOperationQueueDepth(): number {
+    return 0;
+  }
+
+  /**
+   * isOperationBusy
+   * @returns true when a high-level device operation is queued or running
+   */
+  isOperationBusy(): boolean {
+    return false;
+  }
+
   /**
    * Get Raft message handler (to allow message sending and receiving)
    * @returns RaftMsgHandler - Raft message handler
@@ -390,18 +410,24 @@ export default class RaftConnector {
    *
    */
   async sendRICRESTMsg(commandName: string, params: object,
+    bridgeID: number | undefined = undefined): Promise<RaftOKFail> {
+    return this._sendRICRESTMsg(commandName, params, bridgeID);
+  }
+
+  private async _sendRICRESTMsg(commandName: string, params: object,
     bridgeID: number | undefined = undefined): Promise<RaftOKFail> {
     try {
       // Format the paramList as query string
       const paramEntries = Object.entries(params);
       let paramQueryStr = '';
       for (const param of paramEntries) {
         if (paramQueryStr.length > 0) paramQueryStr += '&';
-        paramQueryStr += param[0] + '=' + param[1];
+        paramQueryStr += `${encodeURIComponent(param[0])}=${encodeURIComponent(String(param[1]))}`;
       }
       // Format the url to send
       if (paramQueryStr.length > 0) commandName += '?' + paramQueryStr;
-      return await this._raftMsgHandler.sendRICRESTURL<RaftOKFail>(commandName, bridgeID);
+      const response = await this._raftMsgHandler.sendRICRESTURL<RaftOKFail | null>(commandName, bridgeID);
+      return response ?? { rslt: 'fail' };
     } catch (error) {
       RaftLog.warn(`sendRICRESTMsg failed ${error}`);
       return { rslt: 'fail' };
@@ -455,6 +481,13 @@ export default class RaftConnector {
     fileBlockData: Uint8Array
   ): void {
     // RaftLog.info(`onRxFileBlock filePos ${filePos} fileBlockData ${RaftUtils.bufferToHex(fileBlockData)}`);
+    const streamID = (filePos >>> 24) & 0xff;
+    const streamFilePos = filePos & 0x00ffffff;
+    const streamCallback = this._rtStreamCallbacks.get(streamID);
+    if (streamID !== 0 && streamCallback) {
+      streamCallback(fileBlockData, streamFilePos, streamID);
+      return;
+    }
     this._raftFileHandler.onFileBlock(filePos, fileBlockData);
   }
 
@@ -516,6 +549,82 @@ export default class RaftConnector {
     return false;
   }
 
+  /**
+   * openRtStream - open an indefinite bidirectional RT stream.
+   * The returned handle can send byte blocks and closes with ufEnd.
+   */
+  async openRtStream(options: RaftRtStreamOptions): Promise<RaftRtStreamHandle> {
+    const cmdMsg = JSON.stringify({
+      cmdName: "ufStart",
+      reqStr: "ufStart",
+      fileType: "rtstream",
+      fileName: options.fileName,
+      endpoint: options.endpoint,
+      fileLen: 0,
+    });
+
+    const startResp = await this._raftMsgHandler.sendRICRESTCmdFrame<RaftRtStreamStartResp>(cmdMsg);
+    if (!startResp || startResp.rslt !== "ok" || startResp.streamID === undefined) {
+      throw new Error(`openRtStream failed ${startResp?.rslt ?? "no response"}`);
+    }
+
+    const streamID = startResp.streamID;
+    const maxBlockSize = startResp.maxBlockSize || this._raftStreamHandler.maxBlockSize;
+    let txFilePos = 0;
+    let sendQueue = Promise.resolve();
+    this._rtStreamCallbacks.set(streamID, options.onData);
+    this._fallbackRtStreamCallback = { streamID, callback: options.onData };
+
+    const sendBytes = async (bytes: Uint8Array): Promise<boolean> => {
+      let sentOk = false;
+      sendQueue = sendQueue
+        .catch(() => {
+          // Keep later terminal input flowing even if an earlier block failed.
+        })
+        .then(async () => {
+          sentOk = await this._raftMsgHandler.sendStreamBlock(bytes, txFilePos, streamID);
+          if (sentOk) {
+            txFilePos = (txFilePos + bytes.length) & 0x00ffffff;
+          }
+        });
+      await sendQueue;
+      return sentOk;
+    };
+
+    const close = async (): Promise<boolean> => {
+      this._rtStreamCallbacks.delete(streamID);
+      if (this._fallbackRtStreamCallback?.streamID === streamID) {
+        this._fallbackRtStreamCallback = null;
+      }
+      const endMsg = JSON.stringify({
+        cmdName: "ufEnd",
+        reqStr: "ufEnd",
+        streamID,
+      });
+      try {
+        const endResp = await this._raftMsgHandler.sendRICRESTCmdFrame<RaftOKFail>(endMsg);
+        return endResp?.rslt === "ok";
+      } catch (error) {
+        RaftLog.warn(`closeRtStream failed ${streamID}: ${error}`);
+        return false;
+      }
+    };
+
+    // Some endpoints use an initial empty block as an attach signal.
+    // Keep this opt-in because older endpoints reject zero-length ufBlock frames.
+    if (options.sendInitialEmptyBlock) {
+      await sendBytes(new Uint8Array());
+    }
+
+    return {
+      streamID,
+      maxBlockSize,
+      sendBytes,
+      sendText: (text: string) => sendBytes(new TextEncoder().encode(text)),
+      close,
+    };
+  }
+
   /**
    * streamAudio - stream audio
    * @param streamContents audio data

src/RaftFileHandler.ts

@@ -65,6 +65,7 @@ export default class RaftFileHandler {
   private _ackedFilePos = 0;
   private _batchAckReceived = false;
   private _isTxCancelled = false;
+  private _fileTxStreamID = 0;
 
   // File receive info
   private _isRxCancelled = false;
@@ -78,7 +79,6 @@ export default class RaftFileHandler {
   private _fileRxLastAckTime = 0;
   private _fileRxLastBlockTime = 0;
   private _fileRxLastAckPos = 0;
-  private OVERALL_FILE_TRANSFER_TIMEOUT_MS = 100000;
   private FILE_RX_ACK_RESEND_TIMEOUT_MS = 1000;
 
   // RaftCommsStats
@@ -111,6 +111,7 @@ export default class RaftFileHandler {
     progressCallback: ((sent: number, total: number, progress: number) => void) | undefined,
   ): Promise<boolean> {
     this._isTxCancelled = false;
+    this._fileTxStreamID = 0;
 
     // Send file start message
     if (!await this._sendFileStartMsg(fileName, fileType, fileDest, fileContents))
@@ -166,8 +167,8 @@ export default class RaftFileHandler {
       RaftLog.warn(`sendFileStartMsg error ${err}`);
       return false;
     }
-    if (fileStartResp.rslt !== 'ok') {
-      RaftLog.warn(`sendFileStartMsg error ${fileStartResp.rslt}`);
+    if (!fileStartResp || fileStartResp.rslt !== 'ok') {
+      RaftLog.warn(`sendFileStartMsg error ${fileStartResp?.rslt ?? 'no response'}`);
       return false;
     }
 
@@ -182,12 +183,17 @@ export default class RaftFileHandler {
     } else {
       this._batchAckSize = this._requestedBatchAckSize;
     }
+    const streamID = fileStartResp.streamID;
+    this._fileTxStreamID = streamID !== undefined && streamID > 0 && streamID <= 0xff ? streamID : 0;
     RaftLog.debug(
       `_fileSendStartMsg fileBlockSize req ${this._requestedFileBlockSize} resp ${fileStartResp.batchMsgSize} actual ${this._fileBlockSize}`,
     );
     RaftLog.debug(
       `_fileSendStartMsg batchAckSize req ${this._requestedBatchAckSize} resp ${fileStartResp.batchAckSize} actual ${this._batchAckSize}`,
     );
+    RaftLog.debug(
+      `_fileSendStartMsg streamID ${this._fileTxStreamID}`,
+    );
     return true;
   }
 
@@ -203,7 +209,8 @@ export default class RaftFileHandler {
         ? 'espfwupdate'
         : 'fileupload';
     const fileLen = fileContents.length;
-    const cmdMsg = `{"cmdName":"ufEnd","reqStr":"${reqStr}","fileType":"${fileDest}","fileName":"${fileName}","fileLen":${fileLen}}`;
+    const streamIDJson = this._fileTxStreamID !== 0 ? `,"streamID":${this._fileTxStreamID}` : '';
+    const cmdMsg = `{"cmdName":"ufEnd","reqStr":"${reqStr}","fileType":"${fileDest}","fileName":"${fileName}","fileLen":${fileLen}${streamIDJson}}`;
 
     // Await outstanding promises
     try {
@@ -224,12 +231,13 @@ export default class RaftFileHandler {
       RaftLog.warn(`sendFileEndMsg error ${err}`);
       return false;
     }
-    return fileEndResp.rslt === 'ok';
+    return fileEndResp?.rslt === 'ok';
   }
 
   async _sendFileCancelMsg(): Promise<void> {
     // File cancel command message
-    const cmdMsg = `{"cmdName":"ufCancel"}`;
+    const streamIDJson = this._fileTxStreamID !== 0 ? `,"streamID":${this._fileTxStreamID}` : '';
+    const cmdMsg = `{"cmdName":"ufCancel"${streamIDJson}}`;
 
     // Await outstanding promises
     await this.awaitOutstandingMsgPromises(true);
@@ -368,7 +376,7 @@ export default class RaftFileHandler {
     await this.awaitOutstandingMsgPromises(false);
 
     // Send
-    const promRslt = this._msgHandler.sendFileBlock(fileContents.subarray(blockStart, blockEnd), blockStart);
+    const promRslt = this._msgHandler.sendFileBlock(fileContents.subarray(blockStart, blockEnd), blockStart, this._fileTxStreamID);
     if (!promRslt) {
       return false;
     }
@@ -438,8 +446,8 @@ export default class RaftFileHandler {
       // Establish a bridge
       const bridgedDeviceSerialPort = "Serial" + fileSource.slice(bridgeSerialPrefix.length);
       const cmdResp = await this._msgHandler.createCommsBridge(bridgedDeviceSerialPort, "fileSource");
-      if (cmdResp.rslt != "ok") {
-        RaftLog.warn(`fileReceive - failed to setup bridge ${cmdResp.rslt}`);
+      if (!cmdResp || cmdResp.rslt != "ok") {
+        RaftLog.warn(`fileReceive - failed to setup bridge ${cmdResp?.rslt ?? 'no response'}`);
         return new RaftFileDownloadResult();
       }
       bridgeID = cmdResp.bridgeID;
@@ -498,7 +506,7 @@ export default class RaftFileHandler {
       return false;
     }
     RaftLog.info(`_receiveFileStartMsg rslt ${JSON.stringify(cmdResp)}`);
-    if (cmdResp.rslt === 'ok') {
+    if (cmdResp?.rslt === 'ok') {
       this._fileRxBatchMsgSize = cmdResp.batchMsgSize;
       this._fileRxBatchAckSize = cmdResp.batchAckSize;
       this._fileRxStreamID = cmdResp.streamID;
@@ -511,6 +519,10 @@ export default class RaftFileHandler {
       this._fileRxLastBlockTime = Date.now();
       this._fileRxActive = true;
     }
+    if (!cmdResp) {
+      RaftLog.warn(`_receiveFileStartMsg failed no response`);
+      return false;
+    }
     return cmdResp.rslt === 'ok';
   }
 
@@ -581,6 +593,7 @@ export default class RaftFileHandler {
               `elapsed ${now - startTime}ms overallTimeout ${overallTimeoutMs}ms ` +
               `blockGap ${now - this._fileRxLastBlockTime}ms blockTimeout ${blockTimeoutMs}ms`);
           this._fileRxActive = false;
+          this._sendFileRxCancelMsg(bridgeID);
           reject(new Error('fileReceive failed'));
           return;
         }
@@ -652,6 +665,12 @@ export default class RaftFileHandler {
     }
 
     // Check deferred CRC if start response didn't include one
+    if (!cmdResp) {
+      RaftLog.warn(`_receiveFileEnd failed no response`);
+      this._fileRxActive = false;
+      return false;
+    }
+
     if (this._fileRxCrc16 < 0 && cmdResp.crc16) {
       const expectedCrc = parseInt(cmdResp.crc16, 16);
       const actualCrc = RaftMiniHDLC.crc16(this._fileRxBuffer);
@@ -684,8 +703,20 @@ export default class RaftFileHandler {
   ): void {
     // RaftLog.info(`onFileBlock filePos ${filePos} fileBlockData ${RaftUtils.bufferToHex(fileBlockData)}`);
 
+    if (!this._fileRxActive) {
+      RaftLog.verbose(`onFileBlock ignored inactive transfer filePos ${filePos} len ${fileBlockData.length}`);
+      return;
+    }
+
+    const streamID = (filePos >>> 24) & 0xff;
+    const streamFilePos = filePos & 0x00ffffff;
+    if (streamID !== 0 && streamID !== this._fileRxStreamID) {
+      RaftLog.verbose(`onFileBlock ignored stale streamID ${streamID} active ${this._fileRxStreamID}`);
+      return;
+    }
+
     // Check if this is the next block we are expecting
-    if (filePos === this._fileRxBuffer.length) {
+    if (streamFilePos === this._fileRxBuffer.length) {
 
       // Add to buffer
       const tmpArray = new Uint8Array(this._fileRxBuffer.length + fileBlockData.length);
@@ -700,7 +731,7 @@ export default class RaftFileHandler {
       // RaftLog.info(`onFileBlock filePos ${filePos} fileBlockData ${RaftUtils.bufferToHex(fileBlockData)} added to buffer`);
 
     } else {
-      RaftLog.warn(`onFileBlock expected streamID ${this._fileRxStreamID} filePos ${filePos} fileBlockData ${RaftUtils.bufferToHex(fileBlockData)} out of sequence`);
+      RaftLog.warn(`onFileBlock expected streamID ${this._fileRxStreamID} filePos ${streamFilePos} fileBlockData ${RaftUtils.bufferToHex(fileBlockData)} out of sequence`);
     }
   }