Initial es6 async conversion work

bin/gamedig.js

@@ -5,7 +5,7 @@ const argv = require('minimist')(process.argv.slice(2)),
 
 const debug = argv.debug;
 delete argv.debug;
-const outputFormat = argv.output;
+const pretty = !!argv.pretty;
 delete argv.output;
 
 const options = {};
@@ -25,7 +25,7 @@ Gamedig.isCommandLine = true;
 
 Gamedig.query(options)
     .then((state) => {
-        if(outputFormat === 'pretty') {
+        if(pretty) {
             console.log(JSON.stringify(state,null,'  '));
         } else {
             console.log(JSON.stringify(state));
@@ -42,7 +42,7 @@ Gamedig.query(options)
             if (error instanceof Error) {
                 error = error.message;
             }
-            if (outputFormat === 'pretty') {
+            if (pretty) {
                 console.log(JSON.stringify({error: error}, null, '  '));
             } else {
                 console.log(JSON.stringify({error: error}));

lib/index.js

@@ -2,7 +2,7 @@ const dgram = require('dgram'),
     TypeResolver = require('./typeresolver'),
     HexUtil = require('./HexUtil');
 
-const activeQueries = [];
+const activeQueries = new Set();
 
 const udpSocket = dgram.createSocket('udp4');
 udpSocket.unref();
@@ -13,12 +13,9 @@ udpSocket.on('message', (buffer, rinfo) => {
         console.log(HexUtil.debugDump(buffer));
     }
     for(const query of activeQueries) {
-        if(
+        if(query.options.address !== rinfo.address) continue;
-            query.options.address !== rinfo.address
-            && query.options.altaddress !== rinfo.address
-        ) continue;
         if(query.options.port_query !== rinfo.port) continue;
-        query._udpResponse(buffer);
+        query._udpIncoming(buffer);
         break;
     }
 });
@@ -29,27 +26,14 @@ udpSocket.on('error', (e) => {
 class Gamedig {
 
     static query(options,callback) {
-        const promise = new Promise((resolve,reject) => {
+        const promise = (async () => {
             for (const key of Object.keys(options)) {
                 if (['port_query', 'port'].includes(key)) {
                     options[key] = parseInt(options[key]);
                 }
             }
 
-            options.callback = (state) => {
+            let query = TypeResolver.lookup(options.type);
-                if (state.error) reject(state.error);
-                else resolve(state);
-            };
-
-            let query;
-            try {
-                query = TypeResolver.lookup(options.type);
-            } catch(e) {
-                process.nextTick(() => {
-                    options.callback({error:e});
-                });
-                return;
-            }
             query.debug = Gamedig.debug;
             query.udpSocket = udpSocket;
             query.type = options.type;
@@ -73,17 +57,13 @@ class Gamedig {
                 query.options[key] = options[key];
             }
 
-            activeQueries.push(query);
+            activeQueries.add(query);
-
+            try {
-            query.on('finished',() => {
+                return await query.runAll();
-                const i = activeQueries.indexOf(query);
+            } finally {
-                if(i >= 0) activeQueries.splice(i, 1);
+                activeQueries.delete(query);
-            });
+            }
-
+        })();
-            process.nextTick(() => {
-                query.start();
-            });
-        });
 
         if (callback && callback instanceof Function) {
             if(callback.length === 2) {

lib/typeresolver.js

@@ -1,5 +1,6 @@
 const Path = require('path'),
-    fs = require('fs');
+    fs = require('fs'),
+    Core = require('../protocols/core');
 
 const protocolDir = Path.normalize(__dirname+'/../protocols');
 const gamesFile = Path.normalize(__dirname+'/../games.txt');
@@ -55,6 +56,10 @@ function createProtocolInstance(type) {
 }
 
 class TypeResolver {
+    /**
+     * @param {string} type
+     * @returns Core
+     */
     static lookup(type) {
         if(!type) throw Error('No game specified');
 

protocols/americasarmy.js

@@ -1,8 +1,8 @@
 const Gamespy2 = require('./gamespy2');
 
 class AmericasArmy extends Gamespy2 {
-    finalizeState(state) {
+    async run(state) {
-        super.finalizeState(state);
+        await super.run(state);
         state.name = this.stripColor(state.name);
         state.map = this.stripColor(state.map);
         for(const key of Object.keys(state.raw)) {

protocols/armagetron.js

@@ -7,10 +7,10 @@ class Armagetron extends Core {
         this.byteorder = 'be';
     }
 
-    run(state) {
+    async run(state) {
         const b = Buffer.from([0,0x35,0,0,0,0,0,0x11]);
 
-        this.udpSend(b,(buffer) => {
+        await this.udpSend(b,(buffer) => {
             const reader = this.reader(buffer);
 
             reader.skip(6);
@@ -37,7 +37,7 @@ class Armagetron extends Core {
             state.raw.uri = this.readString(reader);
             state.raw.globalids = this.readString(reader);
             this.finish(state);
-            return true;
+            return null;
         });
     }
 

protocols/ase.js

@@ -1,13 +1,14 @@
 const Core = require('./core');
 
 class Ase extends Core {
-    run(state) {
+    async run(state) {
-        this.udpSend('s',(buffer) => {
+        const buffer = await this.udpSend('s',(buffer) => {
             const reader = this.reader(buffer);
+            const header = reader.string({length: 4});
+            if (header === 'EYE1') return buffer;
+        });
 
-            const header = reader.string({length:4});
+        const reader = this.reader(buffer);
-            if(header !== 'EYE1') return;
-
         state.raw.gamename = this.readString(reader);
         state.raw.port = parseInt(this.readString(reader));
         state.name = this.readString(reader);
@@ -36,9 +37,6 @@ class Ase extends Core {
             if(flags & 32) player.time = parseInt(this.readString(reader));
             state.players.push(player);
         }
-
-            this.finish(state);
-        });
     }
 
     readString(reader) {

protocols/battlefield.js

@@ -1,5 +1,4 @@
-const async = require('async'),
+const Core = require('./core');
-    Core = require('./core');
 
 class Battlefield extends Core {
     constructor() {
@@ -8,13 +7,10 @@ class Battlefield extends Core {
         this.isBadCompany2 = false;
     }
 
-    run(state) {
+    async run(state) {
-        async.series([
+        await this.withTcp(async socket => {
-            (c) => {
+            {
-                this.query(['serverInfo'], (data) => {
+                const data = await this.query(socket, ['serverInfo']);
-                    if(this.debug) console.log(data);
-                    if(data.shift() !== 'OK') return this.fatal('Missing OK');
-
                 state.raw.name = data.shift();
                 state.raw.numplayers = parseInt(data.shift());
                 state.maxplayers = parseInt(data.shift());
@@ -25,10 +21,10 @@ class Battlefield extends Core {
 
                 const teamCount = data.shift();
                 state.raw.teams = [];
-                    for(let i = 0; i < teamCount; i++) {
+                for (let i = 0; i < teamCount; i++) {
                     const tickets = parseFloat(data.shift());
                     state.raw.teams.push({
-                            tickets:tickets
+                        tickets: tickets
                     });
                 }
 
@@ -39,7 +35,7 @@ class Battlefield extends Core {
                 state.password = (data.shift() === 'true');
                 state.raw.uptime = parseInt(data.shift());
                 state.raw.roundtime = parseInt(data.shift());
-                    if(this.isBadCompany2) {
+                if (this.isBadCompany2) {
                     data.shift();
                     data.shift();
                 }
@@ -47,45 +43,36 @@ class Battlefield extends Core {
                 state.raw.punkbusterversion = data.shift();
                 state.raw.joinqueue = (data.shift() === 'true');
                 state.raw.region = data.shift();
-                    if(!this.isBadCompany2) {
+                if (!this.isBadCompany2) {
                     state.raw.pingsite = data.shift();
                     state.raw.country = data.shift();
                     state.raw.quickmatch = (data.shift() === 'true');
                 }
+            }
 
-                    c();
+            {
-                });
+                const data = await this.query(socket, ['version']);
-            },
+                data.shift();
-            (c) => {
+                state.raw.version = data.shift();
-                this.query(['version'], (data) => {
+            }
-                    if(this.debug) console.log(data);
-                    if(data[0] !== 'OK') return this.fatal('Missing OK');
-
-                    state.raw.version = data[2];
-
-                    c();
-                });
-            },
-            (c) => {
-                this.query(['listPlayers','all'], (data) => {
-                    if(this.debug) console.log(data);
-                    if(data.shift() !== 'OK') return this.fatal('Missing OK');
 
+            {
+                const data = await this.query(socket, ['listPlayers', 'all']);
                 const fieldCount = parseInt(data.shift());
                 const fields = [];
-                    for(let i = 0; i < fieldCount; i++) {
+                for (let i = 0; i < fieldCount; i++) {
                     fields.push(data.shift());
                 }
                 const numplayers = data.shift();
-                    for(let i = 0; i < numplayers; i++) {
+                for (let i = 0; i < numplayers; i++) {
                     const player = {};
                     for (let key of fields) {
                         let value = data.shift();
 
-                            if(key === 'teamId') key = 'team';
+                        if (key === 'teamId') key = 'team';
-                            else if(key === 'squadId') key = 'squad';
+                        else if (key === 'squadId') key = 'squad';
 
-                            if(
+                        if (
                             key === 'kills'
                             || key === 'deaths'
                             || key === 'score'
@@ -102,39 +89,23 @@ class Battlefield extends Core {
                     }
                     state.players.push(player);
                 }
-
+            }
-                    this.finish(state);
         });
     }
-        ]);
+
-    }
+    async query(socket, params) {
-    query(params,c) {
+        const outPacket = this.buildPacket(params);
-        this.tcpSend(buildPacket(params), (data) => {
+        return await this.tcpSend(socket, outPacket, (data) => {
             const decoded = this.decodePacket(data);
-            if(!decoded) return false;
+            if(decoded) {
-            c(decoded);
+                if(this.debug) console.log(decoded);
-            return true;
+                if(decoded.shift() !== 'OK') throw new Error('Missing OK');
+                return decoded;
+            }
         });
     }
-    decodePacket(buffer) {
-        if(buffer.length < 8) return false;
-        const reader = this.reader(buffer);
-        const header = reader.uint(4);
-        const totalLength = reader.uint(4);
-        if(buffer.length < totalLength) return false;
 
-        const paramCount = reader.uint(4);
+    buildPacket(params) {
-        const params = [];
-        for(let i = 0; i < paramCount; i++) {
-            const len = reader.uint(4);
-            params.push(reader.string({length:len}));
-            const strNull = reader.uint(1);
-        }
-        return params;
-    }
-}
-
-function buildPacket(params) {
         const paramBuffers = [];
         for (const param of params) {
             paramBuffers.push(Buffer.from(param,'utf8'));
@@ -157,6 +128,23 @@ function buildPacket(params) {
         }
 
         return b;
+    }
+    decodePacket(buffer) {
+        if(buffer.length < 8) return false;
+        const reader = this.reader(buffer);
+        const header = reader.uint(4);
+        const totalLength = reader.uint(4);
+        if(buffer.length < totalLength) return false;
+
+        const paramCount = reader.uint(4);
+        const params = [];
+        for(let i = 0; i < paramCount; i++) {
+            const len = reader.uint(4);
+            params.push(reader.string({length:len}));
+            const strNull = reader.uint(1);
+        }
+        return params;
+    }
 }
 
 module.exports = Battlefield;

protocols/core.js

@@ -1,9 +1,11 @@
 const EventEmitter = require('events').EventEmitter,
     dns = require('dns'),
     net = require('net'),
-    async = require('async'),
     Reader = require('../lib/reader'),
-    HexUtil = require('../lib/HexUtil');
+    HexUtil = require('../lib/HexUtil'),
+    util = require('util'),
+    dnsLookupAsync = util.promisify(dns.lookup),
+    dnsResolveAsync = util.promisify(dns.resolve);
 
 class Core extends EventEmitter {
     constructor() {
@@ -13,23 +15,15 @@ class Core extends EventEmitter {
             attemptTimeout: 10000,
             maxAttempts: 1
         };
-        this.attempt = 1;
-        this.finished = false;
         this.encoding = 'utf8';
         this.byteorder = 'le';
         this.delimiter = '\0';
         this.srvRecord = null;
-        this.attemptTimeoutTimer = null;
-    }
 
-    fatal(err,noretry) {
+        this.attemptAbortables = new Set();
-        if(!noretry && this.attempt < this.options.maxAttempts) {
+        this.udpCallback = null;
-            this.attempt++;
+        this.udpLocked = false;
-            this.start();
+        this.lastAbortableId = 0;
-            return;
-        }
-
-        this.done({error: err.toString()});
     }
 
     initState() {
@@ -46,128 +40,138 @@ class Core extends EventEmitter {
         };
     }
 
-    finalizeState(state) {}
+    // Run all attempts
-
+    async runAll() {
-    finish(state) {
+        let result = null;
-        this.finalizeState(state);
+        let lastError = null;
-        this.done(state);
+        for (let attempt = 1; attempt <= this.options.maxAttempts; attempt++) {
-    }
+            try {
-
+                result = await this.runOnceSafe();
-    done(state) {
+                result.query.attempts = attempt;
-        if(this.finished) return;
+                break;
-
+            } catch (e) {
-        if(this.options.notes)
+                lastError = e;
-            state.notes = this.options.notes;
-
-        state.query = {};
-        if('host' in this.options) state.query.host = this.options.host;
-        if('address' in this.options) state.query.address = this.options.address;
-        if('port' in this.options) state.query.port = this.options.port;
-        if('port_query' in this.options) state.query.port_query = this.options.port_query;
-        state.query.type = this.type;
-        if('pretty' in this) state.query.pretty = this.pretty;
-        state.query.duration = Date.now() - this.startMillis;
-        state.query.attempts = this.attempt;
-
-        this.reset();
-        this.finished = true;
-        this.emit('finished',state);
-        if(this.options.callback) this.options.callback(state);
-    }
-
-    reset() {
-        clearTimeout(this.attemptTimeoutTimer);
-        if(this.timers) {
-            for (const timer of this.timers) {
-                clearTimeout(timer);
             }
         }
-        this.timers = [];
 
-        if(this.tcpSocket) {
+        if (result === null) {
-            this.tcpSocket.destroy();
+            throw lastError;
-            delete this.tcpSocket;
+        }
+        return result;
     }
 
-        this.udpTimeoutTimer = false;
+    // Runs a single attempt with a timeout and cleans up afterward
-        this.udpCallback = false;
+    async runOnceSafe() {
+        try {
+            const result = await this.timedPromise(this.runOnce(), this.options.attemptTimeout, "Attempt");
+            if (this.attemptAbortables.size) {
+                let out = [];
+                for (const abortable of this.attemptAbortables) {
+                    out.push(abortable.id + " " + abortable.stack);
+                }
+                throw new Error('Query succeeded, but abortables were not empty (async leak?):\n' + out.join('\n---\n'));
+            }
+            return result;
+        } finally {
+            // Clean up any lingering long-running functions
+            for (const abortable of this.attemptAbortables) {
+                try {
+                    abortable.abort();
+                } catch(e) {}
+            }
+            this.attemptAbortables.clear();
+        }
     }
 
-    start() {
+    timedPromise(promise, timeoutMs, timeoutMsg) {
+        return new Promise((resolve, reject) => {
+            const cancelTimeout = this.setTimeout(
+                () => reject(new Error(timeoutMsg + " - Timed out after " + timeoutMs + "ms")),
+                timeoutMs
+            );
+            promise.finally(cancelTimeout).then(resolve,reject);
+        });
+    }
+
+    async runOnce() {
+        const startMillis = Date.now();
         const options = this.options;
-        this.reset();
+        if (('host' in options) && !('address' in options)) {
-
+            options.address = await this.parseDns(options.host);
-        this.startMillis = Date.now();
-
-        this.attemptTimeoutTimer = setTimeout(() => {
-            this.fatal('timeout');
-        },this.options.attemptTimeout);
-
-        async.series([
-            (c) => {
-                // resolve host names
-                if(!('host' in options)) return c();
-                if(options.host.match(/\d+\.\d+\.\d+\.\d+/)) {
-                    options.address = options.host;
-                    c();
-                } else {
-                    this.parseDns(options.host,c);
         }
-            },
-            (c) => {
-                // calculate query port if needed
         if(!('port_query' in options) && 'port' in options) {
             const offset = options.port_query_offset || 0;
             options.port_query = options.port + offset;
         }
-                c();
+
-            },
+        const state = this.initState();
-            (c) => {
+        await this.run(state);
-                // run
+
-                this.run(this.initState());
+        if (this.options.notes)
+            state.notes = this.options.notes;
+
+        state.query = {};
+        if ('host' in this.options) state.query.host = this.options.host;
+        if ('address' in this.options) state.query.address = this.options.address;
+        if ('port' in this.options) state.query.port = this.options.port;
+        if ('port_query' in this.options) state.query.port_query = this.options.port_query;
+        state.query.type = this.type;
+        if ('pretty' in this) state.query.pretty = this.pretty;
+        state.query.duration = Date.now() - startMillis;
+
+        return state;
     }
 
-        ]);
+    async run(state) {}
-    }
 
-    run() {}
+    /**
-
+     * @param {string} host
-    parseDns(host,c) {
+     * @returns {Promise<string>}
-        const resolveStandard = (host,c) => {
+     */
+    async parseDns(host) {
+        const isIp = (host) => {
+            return !!host.match(/\d+\.\d+\.\d+\.\d+/);
+        };
+        const resolveStandard = async (host) => {
+            if(isIp(host)) return host;
             if(this.debug) console.log("Standard DNS Lookup: " + host);
-            dns.lookup(host, (err,address,family) => {
+            const {address,family} = await dnsLookupAsync(host);
-                if(err) return this.fatal(err);
             if(this.debug) console.log(address);
-                this.options.address = address;
+            return address;
-                c();
-            });
         };
-
+        const resolveSrv = async (srv,host) => {
-        const resolveSrv = (srv,host,c) => {
+            if(isIp(host)) return host;
             if(this.debug) console.log("SRV DNS Lookup: " + srv+'.'+host);
-            dns.resolve(srv+'.'+host, 'SRV', (err,addresses) => {
+            let records;
-                if(this.debug) console.log(err, addresses);
+            try {
-                if(err) return resolveStandard(host,c);
+                records = await dnsResolveAsync(srv + '.' + host, 'SRV');
-                if(addresses.length >= 1) {
+                if(this.debug) console.log(records);
-                    const line = addresses[0];
+                if(records.length >= 1) {
-                    this.options.port = line.port;
+                    const record = records[0];
-                    const srvhost = line.name;
+                    this.options.port = record.port;
-
+                    const srvhost = record.name;
-                    if(srvhost.match(/\d+\.\d+\.\d+\.\d+/)) {
+                    return await resolveStandard(srvhost);
-                        this.options.address = srvhost;
-                        c();
-                    } else {
-                        // resolve yet again
-                        resolveStandard(srvhost,c);
                 }
-                    return;
+            } catch(e) {
+                if (this.debug) console.log(e.toString());
             }
-                return resolveStandard(host,c);
+            return await resolveStandard(host);
-            });
         };
 
-        if(this.srvRecord) resolveSrv(this.srvRecord,host,c);
+        if(this.srvRecord) return await resolveSrv(this.srvRecord, host);
-        else resolveStandard(host,c);
+        else return await resolveStandard(host);
+    }
+
+    addAbortable(fn) {
+        const id = ++this.lastAbortableId;
+        const stack = new Error().stack;
+        const entry = { id: id, abort: fn, stack: stack };
+        if (this.debug) console.log("Adding abortable: " + id);
+        this.attemptAbortables.add(entry);
+        return () => {
+            if (this.debug) console.log("Removing abortable: " + id);
+            this.attemptAbortables.delete(entry);
+        }
     }
 
     // utils
@@ -184,125 +188,169 @@ class Core extends EventEmitter {
             }
         }
     }
-    setTimeout(c,t) {
-        if(this.finished) return 0;
-        const id = setTimeout(c,t);
-        this.timers.push(id);
-        return id;
-    }
 
     trueTest(str) {
         if(typeof str === 'boolean') return str;
         if(typeof str === 'number') return str !== 0;
         if(typeof str === 'string') {
             if(str.toLowerCase() === 'true') return true;
-            if(str === 'yes') return true;
+            if(str.toLowerCase() === 'yes') return true;
             if(str === '1') return true;
         }
         return false;
     }
 
-    _tcpConnect(c) {
+    /**
-        if(this.tcpSocket) return c(this.tcpSocket);
+     * @param {function(Socket):Promise} fn
-
+     * @returns {Promise<Socket>}
-        let connected = false;
+     */
-        let received = Buffer.from([]);
+    async withTcp(fn) {
         const address = this.options.address;
         const port = this.options.port_query;
 
-        const socket = this.tcpSocket = net.connect(port,address,() => {
+        const socket = net.connect(port,address);
-            if(this.debug) console.log(address+':'+port+" TCPCONNECTED");
-            connected = true;
-            c(socket);
-        });
         socket.setNoDelay(true);
-        if(this.debug) console.log(address+':'+port+" TCPCONNECT");
+        const cancelAbortable = this.addAbortable(() => socket.destroy());
 
+        if(this.debug) {
+            console.log(address+':'+port+" TCP Connecting");
             const writeHook = socket.write;
             socket.write = (...args) => {
-            if(this.debug) {
                 console.log(address+':'+port+" TCP-->");
                 console.log(HexUtil.debugDump(args[0]));
-            }
                 writeHook.apply(socket,args);
             };
-
+            socket.on('error', e => console.log('TCP Error: ' + e));
-        socket.on('error', () => {});
+            socket.on('close', () => console.log('TCP Closed'));
-        socket.on('close', () => {
-            if(!this.tcpCallback) return;
-            if(connected) return this.fatal('Socket closed while waiting on TCP');
-            else return this.fatal('TCP Connection Refused');
-        });
             socket.on('data', (data) => {
-            if(!this.tcpCallback) return;
                 if(this.debug) {
                     console.log(address+':'+port+" <--TCP");
                     console.log(HexUtil.debugDump(data));
                 }
-            received = Buffer.concat([received,data]);
-            if(this.tcpCallback(received)) {
-                clearTimeout(this.tcpTimeoutTimer);
-                this.tcpCallback = false;
-                received = Buffer.from([]);
-            }
             });
+            socket.on('ready', () => console.log(address+':'+port+" TCP Connected"));
         }
-    tcpSend(buffer,ondata) {
+
-        process.nextTick(() => {
+        try {
-            if(this.tcpCallback) return this.fatal('Attempted to send TCP packet while still waiting on a managed response');
+            await this.timedPromise(
-            this._tcpConnect((socket) => {
+                new Promise((resolve,reject) => {
+                    socket.on('ready', resolve);
+                    socket.on('close', () => reject(new Error('TCP Connection Refused')));
+                }),
+                this.options.socketTimeout,
+                'TCP Opening'
+            );
+            return await fn(socket);
+        } finally {
+            cancelAbortable();
+            socket.destroy();
+        }
+    }
+
+    setTimeout(callback, time) {
+        let cancelAbortable;
+        const onTimeout = () => {
+            cancelAbortable();
+            callback();
+        };
+        const timeout = setTimeout(onTimeout, time);
+        cancelAbortable = this.addAbortable(() => clearTimeout(timeout));
+        return () => {
+            cancelAbortable();
+            clearTimeout(timeout);
+        }
+    }
+
+    /**
+     * @param {Socket} socket
+     * @param {Buffer} buffer
+     * @param {function(Buffer):boolean} ondata
+     * @returns {Promise}
+     */
+    async tcpSend(socket,buffer,ondata) {
+        return await this.timedPromise(
+            new Promise(async (resolve,reject) => {
+                let received = Buffer.from([]);
+                const onData = (data) => {
+                    received = Buffer.concat([received, data]);
+                    const result = ondata(received);
+                    if (result !== undefined) {
+                        socket.off('data', onData);
+                        resolve(result);
+                    }
+                };
+                socket.on('data', onData);
                 socket.write(buffer);
-            });
+            }),
-            if(!ondata) return;
+            this.options.socketTimeout,
-
+            'TCP'
-            this.tcpTimeoutTimer = this.setTimeout(() => {
+        );
-                this.tcpCallback = false;
-                this.fatal('TCP Watchdog Timeout');
-            },this.options.socketTimeout);
-            this.tcpCallback = ondata;
-        });
     }
 
-    udpSend(buffer,onpacket,ontimeout) {
+    async withUdpLock(fn) {
-        process.nextTick(() => {
+        if (this.udpLocked) {
-            if(this.udpCallback) return this.fatal('Attempted to send UDP packet while still waiting on a managed response');
+            throw new Error('Attempted to lock UDP when already locked');
-            this._udpSendNow(buffer);
+        }
-            if(!onpacket) return;
+        this.udpLocked = true;
-
+        try {
-            this.udpTimeoutTimer = this.setTimeout(() => {
+            return await fn();
-                this.udpCallback = false;
+        } finally {
-                let timeout = false;
+            this.udpLocked = false;
-                if(!ontimeout || ontimeout() !== true) timeout = true;
+            this.udpCallback = null;
-                if(timeout) this.fatal('UDP Watchdog Timeout');
+        }
-            },this.options.socketTimeout);
-            this.udpCallback = onpacket;
-        });
     }
-    _udpSendNow(buffer) {
-        if(!('port_query' in this.options)) return this.fatal('Attempted to send without setting a port');
-        if(!('address' in this.options)) return this.fatal('Attempted to send without setting an address');
 
+    /**
+     * @param {Buffer|string} buffer
+     * @param {function(Buffer):T} onPacket
+     * @param {(function():T)=} onTimeout
+     * @returns Promise<T>
+     * @template T
+     */
+    async udpSend(buffer,onPacket,onTimeout) {
+        if(!('port_query' in this.options)) throw new Error('Attempted to send without setting a port');
+        if(!('address' in this.options)) throw new Error('Attempted to send without setting an address');
         if(typeof buffer === 'string') buffer = Buffer.from(buffer,'binary');
-
         if(this.debug) {
             console.log(this.options.address+':'+this.options.port_query+" UDP-->");
             console.log(HexUtil.debugDump(buffer));
         }
+
+        return await this.withUdpLock(async() => {
             this.udpSocket.send(buffer,0,buffer.length,this.options.port_query,this.options.address);
-    }
+
-    _udpResponse(buffer) {
+            return await new Promise((resolve,reject) => {
-        if(this.udpCallback) {
+                const cancelTimeout = this.setTimeout(() => {
-            const result = this.udpCallback(buffer);
+                    if (this.debug) console.log("UDP timeout detected");
-            if(result === true) {
+                    let success = false;
-                // we're done with this udp session
+                    if (onTimeout) {
-                clearTimeout(this.udpTimeoutTimer);
+                        const result = onTimeout();
-                this.udpCallback = false;
+                        if (result !== undefined) {
-            }
+                            if (this.debug) console.log("UDP timeout resolved by callback");
-        } else {
+                            resolve(result);
-            this.udpResponse(buffer);
+                            success = true;
                         }
                     }
-    udpResponse() {}
+                    if (!success) {
+                        reject(new Error('UDP Watchdog Timeout'));
+                    }
+                },this.options.socketTimeout);
+
+                this.udpCallback = (buffer) => {
+                    const result = onPacket(buffer);
+                    if(result !== undefined) {
+                        if (this.debug) console.log("UDP send finished by callback");
+                        cancelTimeout();
+                        resolve(result);
+                    }
+                };
+            });
+        });
+    }
+
+    _udpIncoming(buffer) {
+        this.udpCallback && this.udpCallback(buffer);
+    }
 }
 
 module.exports = Core;

protocols/samp.js

@@ -1,5 +1,4 @@
-const async = require('async'),
+const Core = require('./core');
-    Core = require('./core');
 
 class Samp extends Core {
     constructor() {
@@ -7,21 +6,21 @@ class Samp extends Core {
         this.encoding = 'win1252';
     }
 
-    run(state) {
+    async run(state) {
-        async.series([
+        // read info
-            (c) => {
+        {
-                this.sendPacket('i',(reader) => {
+            const reader = await this.sendPacket('i');
             state.password = !!reader.uint(1);
             state.raw.numplayers = reader.uint(2);
             state.maxplayers = reader.uint(2);
             state.name = this.readString(reader,4);
             state.raw.gamemode = this.readString(reader,4);
             this.map = this.readString(reader,4);
-                    c();
+        }
-                });
+
-            },
+        // read rules
-            (c) => {
+        {
-                this.sendPacket('r',(reader) => {
+            const reader = await this.sendPacket('r');
             const ruleCount = reader.uint(2);
             state.raw.rules = {};
             for(let i = 0; i < ruleCount; i++) {
@@ -31,11 +30,12 @@ class Samp extends Core {
             }
             if('mapname' in state.raw.rules)
                 state.map = state.raw.rules.mapname;
-                    c();
+        }
-                });
+
-            },
+        // read players
-            (c) => {
+        {
-                this.sendPacket('d',(reader) => {
+            const reader = await this.sendPacket('d', true);
+            if (reader !== null) {
                 const playerCount = reader.uint(2);
                 for(let i = 0; i < playerCount; i++) {
                     const player = {};
@@ -45,49 +45,44 @@ class Samp extends Core {
                     player.ping = reader.uint(4);
                     state.players.push(player);
                 }
-                    c();
+            } else {
-                },() => {
                 for(let i = 0; i < state.raw.numplayers; i++) {
                     state.players.push({});
                 }
-                    c();
-                });
-            },
-            (c) => {
-                this.finish(state);
             }
-        ]);
+        }
     }
     readString(reader,lenBytes) {
         const length = reader.uint(lenBytes);
         if(!length) return '';
-        const string = reader.string({length:length});
+        return reader.string({length:length});
-        return string;
     }
-    sendPacket(type,onresponse,ontimeout) {
+    async sendPacket(type,allowTimeout) {
-        const outbuffer = Buffer.alloc(11);
+        const outBuffer = Buffer.alloc(11);
-        outbuffer.writeUInt32BE(0x53414D50,0);
+        outBuffer.writeUInt32BE(0x53414D50,0);
         const ipSplit = this.options.address.split('.');
-        outbuffer.writeUInt8(parseInt(ipSplit[0]),4);
+        outBuffer.writeUInt8(parseInt(ipSplit[0]),4);
-        outbuffer.writeUInt8(parseInt(ipSplit[1]),5);
+        outBuffer.writeUInt8(parseInt(ipSplit[1]),5);
-        outbuffer.writeUInt8(parseInt(ipSplit[2]),6);
+        outBuffer.writeUInt8(parseInt(ipSplit[2]),6);
-        outbuffer.writeUInt8(parseInt(ipSplit[3]),7);
+        outBuffer.writeUInt8(parseInt(ipSplit[3]),7);
-        outbuffer.writeUInt16LE(this.options.port,8);
+        outBuffer.writeUInt16LE(this.options.port,8);
-        outbuffer.writeUInt8(type.charCodeAt(0),10);
+        outBuffer.writeUInt8(type.charCodeAt(0),10);
 
-        this.udpSend(outbuffer,(buffer) => {
+        return await this.udpSend(
+            outBuffer,
+            (buffer) => {
                 const reader = this.reader(buffer);
-            for(let i = 0; i < outbuffer.length; i++) {
+                for(let i = 0; i < outBuffer.length; i++) {
-                if(outbuffer.readUInt8(i) !== reader.uint(1)) return;
+                    if(outBuffer.readUInt8(i) !== reader.uint(1)) return;
                 }
-            onresponse(reader);
+                return reader;
-            return true;
+            },
-        },() => {
+            () => {
-            if(ontimeout) {
+                if(allowTimeout) {
-                ontimeout();
+                    return null;
-                return true;
                 }
-        });
+            }
+        );
     }
 }
 

protocols/valve.js

@@ -1,5 +1,4 @@
-const async = require('async'),
+const Bzip2 = require('compressjs').Bzip2,
-    Bzip2 = require('compressjs').Bzip2,
     Core = require('./core');
 
 class Valve extends Core {
@@ -28,22 +27,23 @@ class Valve extends Core {
         this._challenge = '';
     }
 
-    run(state) {
+    async run(state) {
-        async.series([
+        await this.queryInfo(state);
-            (c) => { this.queryInfo(state,c); },
+        await this.queryChallenge();
-            (c) => { this.queryChallenge(state,c); },
+        await this.queryPlayers(state);
-            (c) => { this.queryPlayers(state,c); },
+        await this.queryRules(state);
-            (c) => { this.queryRules(state,c); },
+        await this.cleanup(state);
-            (c) => { this.cleanup(state,c); },
-            (c) => { this.finish(state); }
-        ]);
     }
 
-    queryInfo(state,c) {
+    async queryInfo(state) {
-        this.sendPacket(
+        const b = await this.sendPacket(
-            0x54,false,'Source Engine Query\0',
+            0x54,
+            false,
+            'Source Engine Query\0',
             this.goldsrcInfo ? 0x6D : 0x49,
-            (b) => {
+            false
+        );
+
         const reader = this.reader(b);
 
         if(this.goldsrcInfo) state.raw.address = reader.string();
@@ -121,27 +121,38 @@ class Valve extends Core {
             if(this.debug) console.log("GOLDSRC DETECTED - USING MODIFIED SPLIT FORMAT");
             this.goldsrcSplits = true;
         }
-
-                c();
-            }
-        );
     }
 
-    queryChallenge(state,c) {
+    async queryChallenge() {
         if(this.legacyChallenge) {
-            this.sendPacket(0x57,false,null,0x41,(b) => {
             // sendPacket will catch the response packet and
             // save the challenge for us
-                c();
+            await this.sendPacket(
-            });
+                0x57,
-        } else {
+                false,
-            c();
+                null,
+                0x41,
+                false
+            );
         }
     }
 
-    queryPlayers(state,c) {
+    async queryPlayers(state) {
         state.raw.players = [];
-        this.sendPacket(0x55,true,null,0x44,(b) => {
+
+        // CSGO doesn't even respond sometimes if host_players_show is not 2
+        // Ignore timeouts in only this case
+        const allowTimeout = state.raw.steamappid === 730;
+
+        const b = await this.sendPacket(
+            0x55,
+            true,
+            null,
+            0x44,
+            allowTimeout
+        );
+        if (b === null) return; // timed out
+
         const reader = this.reader(b);
         const num = reader.uint(1);
         for(let i = 0; i < num; i++) {
@@ -162,21 +173,13 @@ class Valve extends Core {
                 name:name, score:score, time:time
             });
         }
-
-            c();
-        }, () => {
-            // CSGO doesn't even respond sometimes if host_players_show is not 2
-            // Ignore timeouts in only this case
-            if (state.raw.steamappid === 730) {
-                c();
-                return true;
-            }
-        });
     }
 
-    queryRules(state,c) {
+    async queryRules(state) {
         state.raw.rules = {};
-        this.sendPacket(0x56,true,null,0x45,(b) => {
+        const b = await this.sendPacket(0x56,true,null,0x45,true);
+        if (b === null) return; // timed out - the server probably just has rules disabled
+
         const reader = this.reader(b);
         const num = reader.uint(2);
         for(let i = 0; i < num; i++) {
@@ -184,17 +187,9 @@ class Valve extends Core {
             const value = reader.string();
             state.raw.rules[key] = value;
         }
-            c();
-        }, () => {
-            // no rules were returned after timeout --
-            // the server probably has them disabled
-            // ignore the timeout
-            c();
-            return true;
-        });
     }
 
-    cleanup(state,c) {
+    async cleanup(state) {
         // Battalion 1944 puts its info into rules fields for some reason
         if ('bat_name_s' in state.raw.rules) {
             state.name = state.raw.rules.bat_name_s;
@@ -234,62 +229,98 @@ class Valve extends Core {
             if (sortedPlayers.length) state.players.push(sortedPlayers.pop());
             else state.players.push({});
         }
-
-        c();
     }
 
     /**
+     * Sends a request packet and returns only the response type expected
      * @param {number} type
      * @param {boolean} sendChallenge
      * @param {?string|Buffer} payload
      * @param {number} expect
-     * @param {function(Buffer)} callback
+     * @param {boolean=} allowTimeout
-     * @param {(function():boolean)=} ontimeout
+     * @returns Buffer|null
      **/
-    sendPacket(
+    async sendPacket(
         type,
         sendChallenge,
         payload,
         expect,
-        callback,
+        allowTimeout
-        ontimeout
     ) {
-        const packetStorage = {};
+        for (let keyRetry = 0; keyRetry < 3; keyRetry++) {
-
+            let retryQuery = false;
-        const receivedFull = (reader) => {
+            const response = await this.sendPacketRaw(
+                type, sendChallenge, payload,
+                (payload) => {
+                    const reader = this.reader(payload);
                     const type = reader.uint(1);
-
+                    if (type === 0x41) {
-            if(type === 0x41) {
                         const key = reader.uint(4);
-
+                        if (this._challenge !== key) {
-                if(this.debug) console.log('Received challenge key: ' + key);
+                            if (this.debug) console.log('Received new challenge key: ' + key);
-
-                if(this._challenge !== key) {
                             this._challenge = key;
-                    if(sendChallenge) {
+                            retryQuery = true;
+                            if (keyRetry === 0 && sendChallenge) {
                                 if (this.debug) console.log('Restarting query');
-                        send();
+                                return null;
-                        return true;
                             }
                         }
-
+                    }
-                return;
+                    if (this.debug) console.log("Received " + type.toString(16) + " expected " + expect.toString(16));
+                    if (type === expect) {
+                        return reader.rest();
+                    }
+                },
+                () => {
+                    if (allowTimeout) return null;
+                }
+            );
+            if (!retryQuery) return response;
+        }
+        throw new Error('Received too many challenge key responses');
     }
 
-            if(this.debug) console.log("Received "+type.toString(16)+" expected "+expect.toString(16));
+    /**
-            if(type !== expect) return;
+     * Sends a request packet and assembles partial responses
-            callback(reader.rest());
+     * @param {number} type
-            return true;
+     * @param {boolean} sendChallenge
-        };
+     * @param {?string|Buffer} payload
+     * @param {function(Buffer)} onResponse
+     * @param {function()} onTimeout
+     **/
+    async sendPacketRaw(
+        type,
+        sendChallenge,
+        payload,
+        onResponse,
+        onTimeout
+    ) {
+        if (typeof payload === 'string') payload = Buffer.from(payload, 'binary');
+        const challengeLength = sendChallenge ? 4 : 0;
+        const payloadLength = payload ? payload.length : 0;
 
-        const receivedOne = (buffer) => {
+        const b = Buffer.alloc(5 + challengeLength + payloadLength);
+        b.writeInt32LE(-1, 0);
+        b.writeUInt8(type, 4);
+
+        if (sendChallenge) {
+            let challenge = this._challenge;
+            if (!challenge) challenge = 0xffffffff;
+            if (this.byteorder === 'le') b.writeUInt32LE(challenge, 5);
+            else b.writeUInt32BE(challenge, 5);
+        }
+        if (payloadLength) payload.copy(b, 5 + challengeLength);
+
+        const packetStorage = {};
+        return await this.udpSend(
+            b,
+            (buffer) => {
                 const reader = this.reader(buffer);
-
                 const header = reader.int(4);
                 if(header === -1) {
                     // full package
                     if(this.debug) console.log("Received full packet");
-                return receivedFull(reader);
+                    return onResponse(reader.rest());
                 }
                 if(header === -2) {
                     // partial package
@@ -345,31 +376,11 @@ class Valve extends Core {
                     }
                     const assembledReader = this.reader(assembled);
                     assembledReader.skip(4); // header
-                return receivedFull(assembledReader);
+                    return onResponse(assembledReader.rest());
                 }
-        };
+            },
-
+            onTimeout
-        const send = (c) => {
+        );
-            if(typeof payload === 'string') payload = Buffer.from(payload,'binary');
-            const challengeLength = sendChallenge ? 4 : 0;
-            const payloadLength = payload ? payload.length : 0;
-
-            const b = Buffer.alloc(5 + challengeLength + payloadLength);
-            b.writeInt32LE(-1, 0);
-            b.writeUInt8(type, 4);
-
-            if(sendChallenge) {
-                let challenge = this._challenge;
-                if(!challenge) challenge = 0xffffffff;
-                if(this.byteorder === 'le') b.writeUInt32LE(challenge, 5);
-                else b.writeUInt32BE(challenge, 5);
-            }
-            if(payloadLength) payload.copy(b, 5+challengeLength);
-
-            this.udpSend(b,receivedOne,ontimeout);
-        };
-
-        send();
     }
 }