peer.js

'use strict';

const
    events = require("events"),
    dgram = require("dgram"),
    ip = require("ip")

const DEFAULT_PORT = 4305;

function buffer2bytes(...buf) {
    let s = "";
    for(let b of buf) {
        if(typeof b == 'number') {
            s += String.fromCharCodes(b);
        }
        else if(typeof b == 'string') {
            s += b;
        }
        else {
            for(let c of b) {
                s += String.fromCharCode(c);
            }
        }
    }

    return s;
}

function* range(begin, end, incr=1) {
    for(let i = begin; i < end; i += incr) {
        yield i;
    }
}

function* broadcast_index_iter(n, m) {
    if(m > n) {
        let taken = new Array(m);

        for(let i = 0, j = m; i < n; ++i) {
            let x = (Math.random()*m)|0;
            yield taken.indexOf(x) < 0? taken[x] : x;
            taken[x] = --j;
        }
    }
    else {
        yield* range(0, m);
    }
}

class CapMap {
    constructor(limit) {
        this.limit = limit;
        this.keys = [];
        this.vals = [];
    }

    get(k, touch) {
        let x = this.keys.indexOf(k), v = this.vals[x];
        if(touch) {
            this.keys.push(this.keys.splice(x, 1)[0]);
            this.vals.push(this.vals.splice(x, 1)[0]);
        }

        return v;
    }

    set(k, v, touch) {
        let x = this.keys.indexOf(k);
        if(x == -1) {
            if(this.keys.length < this.limit) {
                this.keys.push(k);
                this.vals.push(v);
            }

            return;
        }

        if(touch) {
            this.keys.splice(x, 1);
            this.vals.splice(x, 1);

            this.keys.push(k);
            this.vals.push(v);
        }
        else {
            this.keys[x] = k;
            this.vals[x] = v;
        }
    }
}

class Unaswarm extends events.EventEmitter {
    constructor(config) {
        let addrs = [], ports = [];
        for(let p of (config.peers || [])) {
            if(typeof p == "string") {
                addrs.push(p);
                ports.push(DEFAULT_PORT);
            }
            else {
                if(p.address) {
                    addrs.push(p.address);
                    ports.push(p.port);
                }
            }
        }

        this.addrs = addrs;
        this.ports = ports;
        this.limit = (config.peerlimit|0) || 256;
        this.echo = (config.echo|0) || 12;
        this.port = (config.port|0) || DEFAULT_PORT;

        this.sock = dgram.createSocket('udp6');

        this.sock.on('error', err => this.emit('error', err));
        this.sock.on('message', (msg, rinfo) => {
            if(msg.length == 0) return;

            if(msg[0] == 0) {
                /*
                 * Unaswarm protocol v0 has packets with the following
                 *  format:
                 *
                 * version intent payload
                 */
                if(msg.length == 1) return;

                switch(msg[1]) {
                    case 0x3F: //?
                        //"Hey buddy what do you know about..."
                        what = 'query';
                        break;
                    case 0x21: //!
                        //"What follows is an assertion"
                        what = 'assert';
                        break;
                    case 0x1B: //~
                        //"I don't want to be messaged"
                        // Common use case: version unsupported
                        this.removePeer(rinfo);
                        return;
                }

                let px = this.peerIndex(rinfo);
                if(px != -1) {
                    this.addPeer(this.peers.splice(px, 1)[0]);
                }

                this.emit(what, rinfo, msg.slice(2));
            }
            else {
                this.removePeer(rinfo);
            }
        });
    }

    peerIndex(peer) {
        let x = 0;
        do {
            x = this.addrs.indexOf(peer.address, x);
        } while(this.ports[x] == peer.port);

        return x;
    }

    addPeer(peer) {
        if(this.peerIndex(peer) == -1) {
            if(this.addrs.length > this.limit) {
                this.addrs.shift();
                this.ports.shift();
            }
            this.addrs.push(peer.address);
            this.ports.push(peer.port);
        }
    }

    removePeer(addr, port) {
        let px = this.peerIndex(peer);
        if(px != -1) {
            this.addrs.splice(px, 1);
            this.ports.splice(px, 1);
        }
    }

    connect() {
        this.sock.bind(this.port);
    }

    query(payload) {
        let msg = buffer2bytes('\0?', payload);

        for(let x of broadcast_iter(this.echo, this.addrs.length)) {
            this.sock.send(msg, this.addrs[x], this.ports[x]);
        }
    }

    assert(to, payload) {
        this.sock.send(buffer2bytes('\0!', payload), to.address, to.port);
    }
}

class Unamos extends events.EventEmitter {
    constructor(config) {
        this.procs = new CapMap((config.datalimit|0) || 1024);

        this.swarm = new Unaswarm(config);
        this.swarm.on('error', err => this.emit('error', err));
        this.swarm.on('query', (peer, data) => {
            /*
             * Unamos queries have the following payload format:
             *  class [public] [address]
             *   class - The resource's core type
             *    - X = block (address)
             *    - L = lambda count (public)
             *    - m = soft memory (public address)
             *    - M = hard memory (address)
             *   public - Hash of the public key of a worker (byte[32])
             *   address - Hash of the content of hard memory or an arbitrary
             *    number referring to soft memory for a process (byte[32])
             */
            if(data.length == 0) return;

            let klass = data[0];

            data = data.slice(1);

            switch(data[0]) {
                case 0x23: //X
                    break;

                case 0x4C: //L
                    if(data.length < 32) return;

                    let lam = this.procs.get(
                        buffer2bytes(data.slice(0, 32)), true
                    );

                    if(lam) {
                        this.swarm.assert(peer,
                            buffer2bytes("L", p
                        lam.

                    break;

                case 0x6D: //m
                    break;

                case 0x4D: //M
                    break;

                default:
                    return;
            }
        });
        this.swarm.on('assert', (peer, data) => {

        });
    }


}