(fix) remove the noisystream packing for default dtls

Author: collapsinghierarchy

src/commands/recv.js

@@ -196,7 +196,6 @@ export async function run(outDir, opts, ctx = {}) {
         { label: "Bytes", value: humanBytes(written) },
         { label: "Duration", value: formatDuration(durationMs) },
         { label: "Throughput", value: throughputStr },
-        { label: "Announced", value: announcedValue },
         { label: "Destination", value: destinationInfo },
       ],
     });

src/transfer/default.js

@@ -4,14 +4,7 @@ import { createAuthSender, createAuthReceiver } from "@noisytransfer/noisyauth";
 import { NoisyError } from "@noisytransfer/errors";
 import { unb64u as b64uToBytes } from "@noisytransfer/util";
 import { buildMetaHeader, stripMetaHeader } from "./meta-header.js";
-import {
-  packStreamInit,
-  packStreamData,
-  packStreamFin,
-  parseStreamInit,
-  parseStreamData,
-  parseStreamFin,
-} from "@noisytransfer/noisystream/frames";
+import { packStreamInit, packStreamFin, parseStreamInit, parseStreamFin } from "@noisytransfer/noisystream/frames";
 import { getLogger } from "../util/logger.js";
 import { confirmSAS } from "../core/sas-confirm.js";
 import { MAX_STREAM_CHUNK_BYTES } from "./constants.js";
@@ -196,28 +189,24 @@ export async function defaultSend(
 
   // 1a) Optional filename: embed as first data frame (encrypted by DTLS)
   // [ 4 bytes magic = 'N' 'T' 'M' '1' ] [ 1 byte nameLen ] [ name UTF-8 bytes ]
-  let seq = 0;
   if (name) {
     try {
       const header = buildMetaHeader(name);
       for (const slice of splitStreamChunk(header)) {
-        rtc.send(packStreamData({ sessionId, seq, chunk: slice }));
+         rtc.send(slice);
         await applyBackpressure(rtc);
-        seq += 1;
       }
     } catch {}
   }
 
   // 2) Stream data frames (ns_data)
-  let sent = 0;
   for await (const chunk of toAsyncIterable(source)) {
     const u8 = chunk instanceof Uint8Array ? chunk : Buffer.from(chunk);
     if (!u8.byteLength) continue;
     for (const slice of splitStreamChunk(u8)) {
-      rtc.send(packStreamData({ sessionId, seq, chunk: slice }));
+      rtc.send(slice);
       await applyBackpressure(rtc);
       sent += slice.byteLength;
-      seq += 1;
       onProgress?.(Math.min(sent, total), total);
     }
     onProgress?.(Math.min(sent, total), total);
@@ -280,46 +269,25 @@ export async function defaultRecv(rtc, { sessionId, sink, onProgress, assumeYes
   }
 
   const offMsg = rtc.onMessage?.((raw) => {
-    const m = toObjectMessage(raw);
-    if (!m) return;
-
     try {
-      // INIT
-      const init = safe(() => parseStreamInit(m));
-      if (init && init.sessionId === sessionId) {
-        const total = Number(init.totalBytes);
-        if (Number.isFinite(total) && total >= 0) {
-          announced = total;
-          try {
-            sink.info?.({ totalBytes: total });
-          } catch {}
-        } else {
-          announced = null;
-        }
-        getLogger().debug(
-          `DTLS: recv INIT totalBytes=${announced != null ? announced : "(none)"}`,
-        );
-        return;
-      }
-
-      // DATA
-      const data = safe(() => parseStreamData(m));
-      if (data && data.sessionId === sessionId) {
-        let u8 = data.chunk instanceof Uint8Array ? data.chunk : new Uint8Array(data.chunk);
+     // Data frames are expected to arrive as binary payloads (Uint8Array or views).
+      const maybeBinary = toUint8(raw);
+      if (maybeBinary) {
+        let u8 = maybeBinary;
         if (!metaSeen) {
           const info = stripMetaHeader(u8);
           if (info) {
             metaSeen = true;
-            // announce filename without touching totalBytes
             try {
               sink.info?.({ name: info.name });
             } catch {}
-            u8 = info.data; // write only payload portion
+            u8 = info.data;
             getLogger().debug("DTLS: recv META name=", info.name);
           } else {
-            metaSeen = true; // first data had no header; avoid re-checking later
+            metaSeen = true; 
           }
         }
+        if (!u8?.byteLength) return;
         run(async () => {
           await sink.write(u8);
           written += u8.byteLength;
@@ -334,6 +302,16 @@ export async function defaultRecv(rtc, { sessionId, sink, onProgress, assumeYes
         });
         return;
       }
+      const m = toObjectMessage(raw);
+      if (!m) return;
+
+      // INIT
+      const init = safe(() => parseStreamInit(m));
+      if (init && init.sessionId === sessionId) {
+        announced = Number(init.totalBytes) || 0;
+        getLogger().debug(`DTLS: recv INIT totalBytes=${announced}`);
+        return;
+      }
 
       // FIN
       const fin = safe(() => parseStreamFin(m));
@@ -415,3 +393,10 @@ function safe(fn) {
     return null;
   }
 }
+
+function toUint8(raw) {
+  if (raw instanceof Uint8Array) return raw;
+  if (ArrayBuffer.isView(raw)) return new Uint8Array(raw.buffer, raw.byteOffset, raw.byteLength);
+  if (raw instanceof ArrayBuffer) return new Uint8Array(raw);
+  return null;
+}