1 files changed, 38 insertions, 1 deletions
diff --git a/lib/payments.js b/lib/payments.js
index 9172c794..6a2d5ec1 100644
--- a/lib/payments.js
+++ b/lib/payments.js
@@ -190,10 +190,39 @@ export let credit = async (hash, amount, memo, ref, type = types.internal) => {
     let connected;
     let f = fs.createWriteStream(out);
 
+    client.subscribe(username);
+
     mimic.stdout.on("end", async () => {
       await wait(() => connected);
       client.publish(username, "start");
 
+      let n = 0;
+
+      let sentChunks = {}; // Dictionary to store chunks and their respective sequence numbers.
+
+      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
+        }
+      }
+
+      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");
+        }
+      });
+
       ffmpeg(raw)
           .audioBitrate(128)
           .audioFrequency(44100)
@@ -203,13 +232,21 @@ export let credit = async (hash, amount, memo, ref, type = types.internal) => {
           })
           .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);
+            n++;
           }) 
           .on('end', () => {
             console.log("FFMPEG done");
             f.end();
-            client.publish(username, "end");
           });
     });