send missing chunks

1 files changed, 63 insertions, 50 deletions

diff --git a/lib/payments.js b/lib/payments.js
index 6a2d5ec1..ad877290 100644
--- a/lib/payments.js
+++ b/lib/payments.js
@@ -11,8 +11,10 @@ import dollarsToWords from "dollars-to-words";
 import got from "got";
 import mqtt from "mqtt";
 import { spawn } from "child_process";
-import ffmpeg from 'fluent-ffmpeg';
+import ffmpeg from "fluent-ffmpeg";
 import fs from "fs";
+import EventEmitter from "events";
+import crc32 from "buffer-crc32";
 
 export const types = {
   internal: "internal",
@@ -183,12 +185,18 @@ export let credit = async (hash, amount, memo, ref, type = types.internal) => {
       parseFloat(((amount * p.rate) / 100000000).toFixed(2))
     );
 
-    const client  = mqtt.connect('mqtt://ln.coinos.io', config.mqtt);
+    const client = mqtt.connect("mqtt://ln.coinos.io", config.mqtt);
     const raw = `/app/data/uploads/${p.id}-raw.mp3`;
     const out = `/app/data/uploads/${p.id}.mp3`;
-    const mimic = spawn("/mimic", ["-t", words, "-voice", "/cmu_us_axb.flitevox", "-o", raw]);
+    const mimic = spawn("/mimic", [
+      "-t",
+      words,
+      "-voice",
+      "/cmu_us_axb.flitevox",
+      "-o",
+      raw
+    ]);
     let connected;
-    let f = fs.createWriteStream(out);
 
     client.subscribe(username);
 
@@ -197,61 +205,66 @@ export let credit = async (hash, amount, memo, ref, type = types.internal) => {
       client.publish(username, "start");
 
       let n = 0;
+      let t = 0;
+      let sz = 480;
+      let chunks = [];
+      let buf = Buffer.alloc(0);
 
-      let sentChunks = {}; // Dictionary to store chunks and their respective sequence numbers.
+      let send = n => {
+        let chunk = chunks[n];
+        if (!chunk) return;
 
-      function retransmit(chunkNumber) {
-        if(sentChunks[chunkNumber]){
-          console.log("retransmitting", chunkNumber)
-          client.publish(username, chunkNumber.toString());
-          client.publish(username, sentChunks[chunkNumber]);
-          setTimeout(() => {
-            if (sentChunks[chunkNumber]) { // Retransmit if not acknowledged
-              retransmit(chunkNumber);
-            }
-          }, 5000); // Check for retransmission every 5 seconds
-        }
-      }
+        let num = Buffer.alloc(2);
+        num.writeUInt16BE(n, 0);
+
+        let flag = Buffer.from("dddddddd", "hex");
+        let checksum = crc32(chunk).reverse();
+
+        let payload = Buffer.concat([flag, num, checksum, chunk]);
+        client.publish(username, payload);
+      };
+
+      client.on("message", (topic, message) => {
+        if (topic !== username) return;
+        let m = message.toString();
+        if (m.startsWith("missing:")) {
+          let missing = m.replace("missing:", "").split(",");
 
-      client.on('message', (topic, message) => {
-        if (message.length < 20) console.log(message, message.toString());
-        if (topic === username && message.toString().startsWith('ack:')) {
-          let seqNumber = parseInt(message.split(':')[1]);
-          console.log("ACK", seqNumber);
-          delete sentChunks[seqNumber]; // Remove acknowledged chunk from dictionary
-          if (!Object.keys(sentChunks).length) client.publish(username, "end");
+          for (let n of missing) send(chunk);
+          client.publish(username, `end:${n}`);
         }
-      });
 
       ffmpeg(raw)
-          .audioBitrate(128)
-          .audioFrequency(44100)
-          .toFormat('mp3')
-          .on('error', (err) => {
-            console.log('Error:', err);
-          })
-          .stream()
-          .on('data', (chunk) => {
-            let seq = n;
-            client.publish(username, n.toString());
-            sentChunks[n] = chunk;
-            client.publish(username, chunk);
-            f.write(chunk);
-            setTimeout(() => {
-              if (sentChunks[seq]) { // Retransmit if not acknowledged
-                retransmit(seq);
-              }
-            }, 5000);
+        .audioBitrate(128)
+        .audioFrequency(44100)
+        .toFormat("mp3")
+        .on("error", err => {
+          console.log("Error:", err);
+        })
+        .stream()
+        .on("data", data => {
+          t += data.length;
+          let i = 0;
+          buf = Buffer.concat([buf, data]);
+
+          while (i < data.length) {
+            chunks[n] = data.slice(i, i + sz);
+            send(n);
+
             n++;
-          }) 
-          .on('end', () => {
-            console.log("FFMPEG done");
-            f.end();
-          });
+            i += sz;
+          }
+        })
+        .on("end", () => {
+          client.publish(username, `end:${chunks.length}`);
+        });
     });
 
-    client.on('connect', function () {
-      client.publish(username, `${p.amount}:${p.tip}:${p.rate}:${p.created}:${p.id}`);
+    client.on("connect", function() {
+      client.publish(
+        username,
+        `pay:${p.amount}:${p.tip}:${p.rate}:${p.created}:${p.id}`
+      );
       setTimeout(() => (connected = true), 50);
     });
   }