index.js

var WriteStream = require("write-stream")
    , uuid = require("node-uuid")
    , assert = require("assert")

    , Introducer = require("./introducer")
    , PeerConnectionPool = require("./peerConnectionPool")

// This URI doesn't exist yet
var introducer = Introducer("discoverynetwork.co/echo", "unique group name")
var id = uuid()
var pool = PeerConnectionPool(introducer, function (stream) {
    // I am an echo stream
    stream.pipe(stream)
})

introducer.on("peer", function onpeer(remoteId) {
    var stream = pool.connect(remoteId)

    stream.pipe(WriteStream(function onwrite(message) {
        assert.equal(message, "hello world")
    }))

    stream.write("hello world")
})

pool.listen(id)

introducer.emit("peer", id)

introducer.js

var shoe = require("shoe")
    , MuxDemux = require("mux-demux")
    , RemoteEventEmitter = require("remote-events")

module.exports = Introducer

function Introducer(uri, groupName)  {
    var stream = shoe(uri)
        , mdm = MuxDemux()
        , channel = mdm.createStream(groupName)
        , emitter = RemoteEventEmitter()

    stream.pipe(mdm).pipe(stream)
    channel.pipe(emitter.getStream()).pipe(channel)

    return emitter
}

peerConnectionPool.js

var EventEmitter = require("events").EventEmitter()
    , uuid = require("node-uuid")
    // This module exists now.
    , PeerConnection = require("peer-connection")

module.exports = PeerConnectionPool

function PeerConnectionPool(emitter, onConnection) {
    var peerHash = {}
        , pool = new EventEmitter()
        , id

    pool.listen = listen
    pool.connect = connect

    if (onConnection) {
        pool.on("connection", onConnection)
    }

    return pool

    function listen(localId) {
        id = localId

        emitter.on(id + ":offer", function (remoteId, offer) {
            var pc = peerHash[remoteId] = PeerConnection()

            pc.createAnswer(offer, function (err, answer) {
                if (err) {
                    return pool.emit("error", err)
                }

                emitter.emit(remoteId + ":answer", id, answer)

                pc.on("connection", function (stream) {
                    pool.emit("connection", stream)
                })
            })

            pc.on("candidate", function (candidate) {
                emitter.emit(remoteId + ":candidate", id, candidate)
            })
        })

        emitter.on(id + ":answer", function (remoteId, answer) {
            var pc = peerHash[remoteId]

            pc.setRemote(answer)
        })

        emitter.on(id + ":candidate", function (remoteId, candidate) {
            var pc = peerHash[remoteId]

            pc.addCandidate(candidate)
        })
    }

    function connect(remoteId, name) {
        if (!id) {
            throw new Error("must listen before connect")
        }

        var pc = peerHash[remoteId] = PeerConnection()

        pc.createOffer(function (err, offer) {
            if (err) {
                return pool.emit("error", err)
            }

            emitter.emit(remoteId + ":offer", id, offer)
        })

        pc.on("candidate", function (candidate) {
            emitter.emit(remoteId + ":candidate", id, candidate)
        })

        return pc.connect(name)
    }
}