1-chat.js

var through = require("through")
var net = require("net")
var argv = require("optimist").argv
var serializer = require("stream-serializer")()

var myIp = require("my-local-ip")()
var port = argv.port || 8000
var name = argv.name || "Anonymous"

if (argv.server) {
    var broadcast = serializer(through(function (chunk) {
        this.queue(chunk)
    }))
    var server = net.createServer(function (socket) {
        socket.pipe(broadcast, {
            end: false
        }).pipe(socket)
    })

    server.listen(port, myIp)
} else {
    var socket = net.connect(port, myIp)
    var client = through(function (chunk) {
        console.log(">", chunk.ip + "@" + chunk.name, "::", chunk.message)
    })

    socket.pipe(serializer(client)).pipe(socket)

    process.stdin.on("data", function (chunk) {
        client.queue({
            message: String(chunk),
            ts: Date.now(),
            name: name,
            ip: myIp
        })
    })
}

2-reconnect.js

var through = require("through")
var net = require("net")
var argv = require("optimist").argv
var serializer = require("stream-serializer")()

var myIp = require("my-local-ip")()
var port = argv.port || 8000
var name = argv.name || "Anonymous"

if (argv.server) {
    var broadcast = serializer(through(function (chunk) {
        this.queue(chunk)
    }))
    var server = net.createServer(function (socket) {
        socket.pipe(broadcast, {
            end: false
        }).pipe(socket)
    })

    server.listen(port, myIp)
} else {
    var client = serializer(through(function (chunk) {
        console.log(">", chunk.ip + "@" + chunk.name, "::", chunk.message)
    }))
    client.pause()

    process.stdin.on("data", function write(chunk) {
        client.queue({
            message: String(chunk),
            ts: Date.now(),
            name: name,
            ip: myIp
        })
    })

    connect()
}

function connect() {
    var socket = net.connect(port, myIp)

    socket.on("close", reconnect)
    socket.on("error", reconnect)

    socket.pipe(client, {
        end: false
    }).pipe(socket)

    socket.once("connect", function () {
        client.resume()
    })

    function reconnect(error) {
        socket.removeListener("close", reconnect)
        client.pause()
        console.log("reconnecting")

        setTimeout(connect, 1000)
    }
}

3-p2p.js

var through = require("through")
var net = require("net")
var argv = require("optimist").argv
var serializer = require("stream-serializer")()
var dirty = require("dirty")

var myIp = require("my-local-ip")()
var port = argv.port || 8000
var name = argv.name || "Anonymous"
var db = dirty("distributed.ips")

var messages = {}

function messageId(chunk) {
    return chunk.ip + "~" + chunk.entropy + "~" + chunk.ts
}


var broadcast = serializer(through(function (chunk) {
    this.queue(chunk)
}))
broadcast.setMaxListeners(Infinity)
var server = net.createServer(function (socket) {
    socket.pipe(broadcast, {
        end: false
    }).pipe(socket)
})

server.listen(port, myIp)

var client = serializer(through(function (chunk) {
    if (!db.get(chunk.ip + ":" + chunk.port)) {
        db.set(chunk.ip + ":" + chunk.port, true)
    }

    var mid = messageId(chunk)
    if (messages[mid]) {
        return
    }

    messages[mid] = true

    console.log(">", chunk.ip + "@" + chunk.name, "::", chunk.message)

    this.queue(chunk)
}))
client.setMaxListeners(Infinity)

process.stdin.on("data", function write(chunk) {
    client.queue({
        message: String(chunk),
        ts: Date.now(),
        name: name,
        ip: myIp,
        port: port,
        entropy: Math.random()
    })
})

db.on("load", function () {
    db.forEach(function (key, port) {
        var parts = key.split(":")
        connect(parts[0], parts[1])
    })
    connect(myIp, 8000)
})

function connect(ip, port) {
    var socket = net.connect(port, ip)

    socket.on("close", reconnect)
    socket.on("error", reconnect)

    socket.pipe(client, {
        end: false
    }).pipe(socket)

    function reconnect() {
        socket.removeListener("close", reconnect)
        // console.log("reconnecting", ip + ":" + port)

        setTimeout(function () {
            connect(ip, port)
        }, 1000)
    }
}