diff options
| -rw-r--r-- | lib/payments.js | 113 | ||||
| -rw-r--r-- | package.json | 1 | ||||
| -rw-r--r-- | pnpm-lock.yaml | 7 |
3 files changed, 71 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); }); } diff --git a/package.json b/package.json index f617d2a8..d7d97c3d 100644 --- a/package.json +++ b/package.json @@ -16,6 +16,7 @@ "@fastify/websocket": "^7.2.0", "bcrypt": "^5.0.0", "bech32": "^2.0.0", + "buffer-crc32": "^0.2.13", "clightning-client": "github:SimonVrouwe/lightning-client-js#2022-11-04_improve_doc", "date-fns": "^2.21.3", "dollars-to-words": "github:asoltys/dollars-to-words", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index a5b398a1..4a1ad2b3 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -32,6 +32,9 @@ dependencies: bech32: specifier: ^2.0.0 version: 2.0.0 + buffer-crc32: + specifier: ^0.2.13 + version: 0.2.13 clightning-client: specifier: github:SimonVrouwe/lightning-client-js#2022-11-04_improve_doc version: github.com/SimonVrouwe/lightning-client-js/946642bafca572fb5d9ce2dc1316d415fa950a05 @@ -1376,6 +1379,10 @@ packages: safe-buffer: 5.2.1 dev: false + resolution: {integrity: sha512-VO9Ht/+p3SN7SKWqcrgEzjGbRSJYTx+Q1pTQC0wrWqHx0vpJraQ6GtHx8tvcg1rlK1byhU5gccxgOgj7B0TDkQ==} + dev: false + resolution: {integrity: sha512-zRpUiDwd/xk6ADqPMATG8vc9VPrkck7T07OIx0gnjmJAnHnTVXNQG3vfvWNuiZIkwu9KrKdA1iJKfsfTVxE6NA==} dev: false |