implement simple retransmission scheme

This commit is contained in:
David Ventura
2024-05-03 12:35:29 +02:00
parent 03893f8d81
commit 204677ceae
2 changed files with 47 additions and 6 deletions
+11
View File
@@ -266,6 +266,17 @@ const makeDrwAck = (dv: DataView): DataView => {
} }
return outbuf; return outbuf;
}; };
export const handle_DrwAck = (session: Session, dv: DataView) => {
const packetlen = dv.add(2).readU16();
const str_type = dv.add(4).readU8();
const str_id = dv.add(5).readU8();
const ack_count = dv.add(6).readU16();
for (let i = 0; i < ack_count; i++) {
const ack_id = dv.add(8 + i * 2).readU16();
session.ackDrw(ack_id);
}
};
export const handle_Drw = (session: Session, dv: DataView) => { export const handle_Drw = (session: Session, dv: DataView) => {
const ack = makeDrwAck(dv); const ack = makeDrwAck(dv);
session.send(ack); session.send(ack);
+36 -6
View File
@@ -2,13 +2,15 @@ import { createSocket, RemoteInfo } from "node:dgram";
import EventEmitter from "node:events"; import EventEmitter from "node:events";
import { Commands, CommandsByValue } from "./datatypes.js"; import { Commands, CommandsByValue } from "./datatypes.js";
import { handle_Drw, handle_P2PAlive, handle_P2PRdy, makeP2pRdy, noop, notImpl } from "./handlers.js"; import { handle_Drw, handle_DrwAck, handle_P2PAlive, handle_P2PRdy, makeP2pRdy, notImpl, noop } from "./handlers.js";
import { create_P2pAlive, DevSerial, SendStartVideo, SendVideoResolution, SendWifiDetails } from "./impl.js"; import { create_P2pAlive, DevSerial, SendStartVideo, SendVideoResolution, SendWifiDetails } from "./impl.js";
import { opt } from "./options.js"; import { opt } from "./options.js";
import { logger } from "./logger.js"; import { logger } from "./logger.js";
export type Session = { export type Session = {
send: (msg: DataView) => void; send: (msg: DataView) => void;
ackDrw: (id: number) => void;
unackedDrw: { [id: number]: { sent_ts: number; data: DataView } };
outgoingCommandId: number; outgoingCommandId: number;
ticket: number[]; ticket: number[];
eventEmitter: EventEmitter; eventEmitter: EventEmitter;
@@ -37,7 +39,8 @@ type msgCb = (
const handleIncoming: msgCb = (session, handlers, msg, rinfo) => { const handleIncoming: msgCb = (session, handlers, msg, rinfo) => {
const ab = new Uint8Array(msg).buffer; const ab = new Uint8Array(msg).buffer;
const dv = new DataView(ab); const dv = new DataView(ab);
const cmd = CommandsByValue[dv.readU16()]; const raw = dv.readU16();
const cmd = CommandsByValue[raw];
logger.log("trace", `<< ${cmd}`); logger.log("trace", `<< ${cmd}`);
handlers[cmd](session, dv, rinfo); handlers[cmd](session, dv, rinfo);
session.lastReceivedPacket = Date.now(); session.lastReceivedPacket = Date.now();
@@ -50,6 +53,7 @@ export const makeSession = (
onLogin: (s: Session) => void, onLogin: (s: Session) => void,
options: opt, options: opt,
): Session => { ): Session => {
let unackedDrw = {};
const sock = createSocket("udp4"); const sock = createSocket("udp4");
sock.on("error", (err) => { sock.on("error", (err) => {
@@ -73,33 +77,59 @@ export const makeSession = (
let buf = create_P2pAlive(); let buf = create_P2pAlive();
session.send(buf); session.send(buf);
} }
if (delta > 8000) { if (delta > 5000) {
session.eventEmitter.emit("disconnect"); session.eventEmitter.emit("disconnect");
} }
} }
}, 400); }, 400);
const resendTimer = setInterval(() => {
const now = Date.now();
for (const [key, value] of Object.entries(session.unackedDrw)) {
const { sent_ts, data } = value;
if (now - sent_ts > 100) {
const pkt_id = data.add(6).readU16();
logger.debug(`Resending packet ${pkt_id} as ${session.outgoingCommandId}`);
data.add(6).writeU16(session.outgoingCommandId);
session.outgoingCommandId++;
delete session.unackedDrw[key];
session.send(data);
}
}
}, 500);
const session: Session = { const session: Session = {
outgoingCommandId: 0, outgoingCommandId: 0,
ticket: [0, 0, 0, 0], ticket: [0, 0, 0, 0],
lastReceivedPacket: 0, lastReceivedPacket: 0,
eventEmitter: new EventEmitter(), eventEmitter: new EventEmitter(),
connected: true, connected: true,
timers: [sessTimer], timers: [sessTimer, resendTimer],
devName: dev.devId, devName: dev.devId,
started: false, started: false,
send: (msg: DataView) => { send: (msg: DataView) => {
const raw = msg.readU16(); const raw = msg.readU16();
const cmd = CommandsByValue[raw]; const cmd = CommandsByValue[raw];
// send command
if (raw == 0xf1d0 && msg.add(4).readU8() == 0xd1) {
const packet_id = msg.add(6).readU16();
logger.debug(`Sending Drw Packet with id ${packet_id}`);
unackedDrw[packet_id] = { sent_ts: Date.now(), data: msg };
}
logger.log("trace", `>> ${cmd}`); logger.log("trace", `>> ${cmd}`);
sock.send(new Uint8Array(msg.buffer), ra.port, session.dst_ip); sock.send(new Uint8Array(msg.buffer), ra.port, session.dst_ip);
}, },
ackDrw: (id: number) => {
logger.debug(`Removing ${id} from pending`);
delete unackedDrw[id];
},
dst_ip: ra.address, dst_ip: ra.address,
curImage: [], curImage: [],
rcvSeqId: 0, rcvSeqId: 0,
frame_is_bad: false, frame_is_bad: false,
frame_was_fixed: false, frame_was_fixed: false,
options: options, options: options,
unackedDrw,
}; };
session.eventEmitter.on("disconnect", () => { session.eventEmitter.on("disconnect", () => {
@@ -134,13 +164,13 @@ export const Handlers: Record<keyof typeof Commands, PacketHandler> = {
PunchPkt: notImpl, PunchPkt: notImpl,
P2PAlive: handle_P2PAlive, P2PAlive: handle_P2PAlive,
P2pRdy: handle_P2PRdy, P2pRdy: handle_P2PRdy,
DrwAck: noop, DrwAck: handle_DrwAck,
Drw: handle_Drw, Drw: handle_Drw,
P2PAliveAck: noop,
Close: notImpl, Close: notImpl,
LanSearchExt: notImpl, LanSearchExt: notImpl,
LanSearch: notImpl, LanSearch: notImpl,
P2PAliveAck: notImpl,
Hello: notImpl, Hello: notImpl,
P2pReq: notImpl, P2pReq: notImpl,
LstReq: notImpl, LstReq: notImpl,