web -> desktop

src/background/bt/client.js

@@ -0,0 +1,98 @@
+'use strict'
+
+const Emiter = require('events')
+var util = require('util');
+var net = require('net');
+
+var PeerQueue = require('./peer-queue');
+var Wire = require('./wire');
+const debug = require('debug')('downloader');
+const config = require('../config')
+
+class Client extends Emiter
+{
+    constructor(options) {
+    	super();
+        this.timeout = config.downloader.timeout;
+        this.maxConnections = config.downloader.maxConnections;
+        debug('timeout', this.timeout)
+        debug('maxConnections', this.maxConnections)
+        this.activeConnections = 0;
+        this.peers = new PeerQueue(this.maxConnections);
+        this.on('download', this._download);
+
+       // if (typeof options.ignore === 'function') {
+        //    this.ignore = options.ignore;
+        //}
+        //else {
+            this.ignore = function (infohash, rinfo, ignore) {
+                ignore(false);
+            };
+       // }
+    }
+
+    _next(infohash, successful) {
+        var req = this.peers.shift(infohash, successful);
+        if (req) {
+            this.ignore(req.infohash.toString('hex'), req.rinfo, (drop) => {
+                if (!drop) {
+                    this.emit('download', req.rinfo, req.infohash);
+                }
+            });
+        }
+    }
+
+    _download(rinfo, infohash)
+    {
+        debug('start download', infohash.toString('hex'), 'connections', this.activeConnections);
+        this.activeConnections++;
+
+        var successful = false;
+        var socket = new net.Socket();
+
+        socket.setTimeout(this.timeout || 5000);
+        socket.connect(rinfo.port, rinfo.address, () => {
+            var wire = new Wire(infohash);
+            socket.pipe(wire).pipe(socket);
+
+            wire.on('metadata', (metadata, infoHash) => {
+                successful = true;
+                debug('successfuly downloader', infoHash, rinfo);
+                this.emit('complete', metadata, infoHash, rinfo);
+                socket.destroy();
+            });
+
+            wire.on('fail', () => {
+                socket.destroy();
+            });
+
+            wire.sendHandshake();
+        });
+
+        socket.on('error', (err) => {
+            socket.destroy();
+        });
+
+        socket.on('timeout', (err) => {
+            socket.destroy();
+        });
+
+        socket.once('close', () => {
+            this.activeConnections--;
+            this._next(infohash, successful);
+        });
+    }
+
+    add(rinfo, infohash) {
+        this.peers.push({infohash: infohash, rinfo: rinfo});
+        if (this.activeConnections < this.maxConnections && this.peers.length() > 0) {
+            this._next();
+        }
+    }
+
+    isIdle() {
+        return this.peers.length() === 0;
+    }
+}
+
+module.exports = Client;

src/background/bt/cpu-usage.js

@@ -0,0 +1,33 @@
+let startTime  = process.hrtime()
+let startUsage = process.cpuUsage()
+
+let keepTime = process.hrtime()
+let keepUsage = process.cpuUsage()
+let sw = false
+
+setInterval(() => {
+	if(!sw) {
+		keepTime = process.hrtime();
+    	keepUsage = process.cpuUsage();
+    	sw = true;
+	} else {
+		startTime = keepTime;
+		startUsage = keepUsage;
+		sw = false;
+	} 
+}, 500)
+
+module.exports = () => {
+    function secNSec2ms (secNSec) {
+      return secNSec[0] * 1000 + secNSec[1] / 1000000
+    }
+
+    var elapTime = process.hrtime(startTime)
+    var elapUsage = process.cpuUsage(startUsage)
+
+    var elapTimeMS = secNSec2ms(elapTime)
+    var elapUserMS = elapUsage.user
+    var elapSystMS = elapUsage.system
+
+    return Math.round(100 * ((elapUserMS + elapSystMS) / 1000) / elapTimeMS)
+}

src/background/bt/peer-queue.js

@@ -0,0 +1,55 @@
+'use strict';
+
+var PeerQueue = function (maxSize, perLimit) {
+    this.maxSize = maxSize || 200;
+    this.perLimit = perLimit || 10;
+    this.peers = {};
+    this.reqs = [];
+};
+
+PeerQueue.prototype._shift = function () {
+    if (this.length() > 0) {
+        var req = this.reqs.shift();
+        this.peers[req.infohash.toString('hex')] = [];
+        return req;
+    }
+};
+
+PeerQueue.prototype.push = function (peer) {
+    var infohashHex = peer.infohash.toString('hex');
+    var peers = this.peers[infohashHex];
+
+    if (peers && peers.length < this.perLimit) {
+        peers.push(peer);
+    }
+    else if (this.length() < this.maxSize) {
+        this.reqs.push(peer);
+    }
+};
+
+PeerQueue.prototype.shift = function (infohash, successful) {
+    if (infohash) {
+        var infohashHex = infohash.toString('hex');
+        if (successful === true) {
+            delete this.peers[infohashHex];
+        }
+        else {
+            var peers = this.peers[infohashHex];
+            if (peers) {
+                if (peers.length > 0) {
+                    return peers.shift();
+                }
+                else {
+                    delete this.peers[infohashHex];
+                }
+            }
+        }
+    }
+    return this._shift();
+};
+
+PeerQueue.prototype.length = function () {
+    return this.reqs.length;
+};
+
+module.exports = PeerQueue;

src/background/bt/spider.js

@@ -0,0 +1,285 @@
+'use strict'
+
+const dgram = require('dgram')
+const Emiter = require('events')
+const bencode = require('bencode')
+const {Table, Node} = require('./table')
+const Token = require('./token')
+const cpuUsage = require('./cpu-usage')
+const config = require('../config')
+const fs = require('fs')
+
+const _debug = require('debug')
+const cpuDebug = _debug('spider:cpu')
+const trafficDebug = _debug('spider:traffic')
+
+const bootstraps = [{
+    address: 'router.bittorrent.com',
+    port: 6881
+}, {
+    address: 'router.utorrent.com',
+    port: 6881
+}, {
+    address: 'dht.transmissionbt.com',
+    port: 6881
+}, {
+    address: 'dht.aelitis.com',
+    port: 6881
+}]
+
+function isValidPort(port) {
+    return port > 0 && port < (1 << 16)
+}
+
+function generateTid() {
+    return parseInt(Math.random() * 99).toString()
+}
+
+class Spider extends Emiter {
+    constructor(client) {
+        super()
+        const options = arguments.length? arguments[0]: {}
+        this.udp = dgram.createSocket('udp4')
+        this.table = new Table(options.tableCaption || 1000)
+        this.bootstraps = options.bootstraps || bootstraps
+        this.token = new Token()
+        this.client = client
+        this.ignore = false; // ignore all requests
+        this.initialized = false;
+        this.trafficSpeed = 0
+
+        this.walkInterval = config.spider.walkInterval;
+        this.cpuLimit = config.spider.cpuLimit;
+        this.cpuInterval = config.spider.cpuInterval;
+    }
+
+    send(message, address) {
+        const data = bencode.encode(message)
+        this.udp.send(data, 0, data.length, address.port, address.address)
+    }
+
+    findNode(id, address) {
+        const message = {
+            t: generateTid(),
+            y: 'q',
+            q: 'find_node',
+            a: {
+                id: id,
+                target: Node.generateID()
+            }
+        }
+        this.send(message, address)
+    }
+
+    join() {
+        this.bootstraps.forEach((bootstrap) => {
+            this.findNode(this.table.id, bootstrap)
+        })
+    }
+
+    walk() {
+        if(this.closing)
+            return
+
+    	if(!this.client || this.client.isIdle()) {
+            if(
+                !this.ignore 
+                && (this.cpuLimit <= 0 || cpuUsage() < this.cpuLimit + this.cpuInterval)
+                && (config.trafficMax <= 0 || this.trafficSpeed == 0 || this.trafficSpeed < config.trafficMax)
+            ) 
+            {
+                const node = this.table.shift()
+                if (node) {
+                    this.findNode(Node.neighbor(node.id, this.table.id), {address: node.address, port: node.port})
+                }
+            }
+    	}
+        setTimeout(()=>this.walk(), this.walkInterval)
+    }
+
+    onFoundNodes(data) {
+        const nodes = Node.decodeNodes(data)
+        nodes.forEach((node) => {
+            if (node.id != this.table.id && isValidPort(node.port)) {
+                this.table.add(node)
+            }
+        })
+        this.emit('nodes', nodes)
+    }
+
+    onFindNodeRequest(message, address) {
+        if(this.cpuLimit > 0 && cpuUsage() > this.cpuLimit) {
+            return
+        }
+
+        if(config.trafficIgnoreDHT && config.trafficMax > 0 && this.trafficSpeed > 0 && this.trafficSpeed > config.trafficMax) {
+            return
+        }
+
+    	const {t: tid, a: {id: nid, target: infohash}} = message
+
+        if (tid === undefined || target.length != 20 || nid.length != 20) {
+            return
+        }
+
+        this.send({
+            t: tid,
+            y: 'r',
+            r: {
+                id: Node.neighbor(nid, this.table.id),
+                nodes: Node.encodeNodes(this.table.first())
+            }
+        }, address)
+    }
+
+    onGetPeersRequest(message, address) {
+        if(this.cpuLimit > 0 && cpuUsage() > this.cpuLimit) {
+            return
+        }
+
+        if(config.trafficIgnoreDHT && config.trafficMax > 0 && this.trafficSpeed > 0 && this.trafficSpeed > config.trafficMax) {
+            return
+        }
+
+        const {t: tid, a: {id: nid, info_hash: infohash}} = message
+
+        if (tid === undefined || infohash.length != 20 || nid.length != 20) {
+            return
+        }
+
+        this.send({
+            t: tid,
+            y: 'r',
+            r: {
+                id: Node.neighbor(nid, this.table.id),
+                nodes: Node.encodeNodes(this.table.first()),
+                token: this.token.token
+            }
+        }, address)
+
+        this.emit('unensureHash', infohash.toString('hex').toUpperCase())
+    }
+
+    onAnnouncePeerRequest(message, address) {
+        let {t: tid, a: {info_hash: infohash, token: token, id: id, implied_port: implied, port: port}} = message
+        if (!tid) return
+
+        if (!this.token.isValid(token)) return
+       
+        port = (implied != undefined && implied != 0) ? address.port : (port || 0)
+        if (!isValidPort(port)) return
+
+        this.send({ t: tid, y: 'r', r: { id: Node.neighbor(id, this.table.id) } }, address)
+
+        let addressPair = {
+            address: address.address,
+            port: port
+        };
+    	this.emit('ensureHash', infohash.toString('hex').toUpperCase(), addressPair)
+        if(this.client && !this.ignore) {
+            cpuDebug('cpu usage:' + cpuUsage())
+            if(this.cpuLimit <= 0 || cpuUsage() <= this.cpuLimit + this.cpuInterval) {
+                this.client.add(addressPair, infohash);
+            }
+        }
+    }
+
+    onPingRequest(message, address) {
+        if(this.cpuLimit > 0 && cpuUsage() > this.cpuLimit) {
+            return
+        }
+
+        if(config.trafficIgnoreDHT && config.trafficMax > 0 && this.trafficSpeed > 0 && this.trafficSpeed > config.trafficMax) {
+            return
+        }
+
+    	this.send({ t: message.t, y: 'r', r: { id: Node.neighbor(message.a.id, this.table.id) } }, address)
+    }
+
+    parse(data, address) {
+        try {
+            const message = bencode.decode(data)
+            if (message.y.toString() == 'r' && message.r.nodes) {
+                this.onFoundNodes(message.r.nodes)
+            } else if (message.y.toString() == 'q') {
+            	switch(message.q.toString()) {
+            		case 'get_peers':
+            		this.onGetPeersRequest(message, address)
+            		break
+            		case 'announce_peer':
+            		this.onAnnouncePeerRequest(message, address)
+            		break
+            		case 'find_node':
+            		this.onFindNodeRequest(message, address)
+            		break
+            		case 'ping':
+            		this.onPingRequest(message, address)
+            		break
+            	}
+            }
+        } catch (err) {}
+    }
+
+    listen(port) {
+        if(this.initialized)
+            return
+        this.initialized = true
+
+        this.udp.bind(port)
+        this.udp.on('listening', () => {
+            console.log(`Listen DHT protocol on ${this.udp.address().address}:${this.udp.address().port}`)
+        })
+        this.udp.on('message', (data, addr) => {
+            this.parse(data, addr)
+        })
+        this.udp.on('error', (err) => {})
+        this.joinInterval = setInterval(() => { 
+            if(!this.client || this.client.isIdle()) {
+                this.join()
+            }
+        }, 3000)
+        this.join()
+        this.walk()
+
+        if(config.trafficMax > 0)
+        {
+            trafficDebug('inore dht traffic', config.trafficIgnoreDHT)
+            const path = `/sys/class/net/${config.trafficInterface}/statistics/rx_bytes`
+            if(fs.existsSync(path))
+            {
+                trafficDebug('limitation', config.trafficMax / 1024, 'kbps/s')
+                let traffic = 0
+                this.trafficInterval = setInterval(() => { 
+                    fs.readFile(path, (err, newTraffic) => {
+                        if(err)
+                            return
+
+                        if(traffic === 0)
+                            traffic = newTraffic
+
+                        this.trafficSpeed = (newTraffic - traffic) / config.trafficUpdateTime
+
+                        trafficDebug('traffic rx', this.trafficSpeed / 1024, 'kbps/s')
+
+                        traffic = newTraffic
+                    })
+                }, 1000 * config.trafficUpdateTime)
+            }
+        }
+    }
+
+    close(callback)
+    {
+        clearInterval(this.joinInterval)
+        if(this.trafficInterval)
+            clearInterval(this.trafficInterval)
+        this.closing = true
+        this.udp.close(() => {
+            this.initialized = false
+            if(callback)
+                callback()
+        })
+    }
+}
+
+module.exports = Spider

src/background/bt/table.js

@@ -0,0 +1,72 @@
+'use strict'
+
+const crypto = require('crypto')
+
+class Node {
+	static generateID() {
+		return crypto.createHash('sha1').update(crypto.randomBytes(20)).digest()
+	}
+
+	constructor(id) {
+		this.id = id || Node.generateNodeID()
+	}
+
+	static neighbor(target, id) {
+		return Buffer.concat([target.slice(0, 10), id.slice(10)])
+	}
+
+	static encodeNodes(nodes) {
+		return Buffer.concat(nodes.map((node)=> Buffer.concat([node.id, Node.encodeIP(node.address), Node.encodePort(node.port)])))
+	}
+
+	static decodeNodes(data) {
+		const nodes = []
+		for (let i = 0; i + 26 <= data.length; i += 26) {
+			nodes.push({
+				id: data.slice(i, i + 20),
+				address: `${data[i + 20]}.${data[i + 21]}.${data[i + 22]}.${data[i + 23]}`,
+				port: data.readUInt16BE(i + 24)
+			})
+		}
+		return nodes
+	}
+
+	static encodeIP(ip) {
+		return Buffer.from(ip.split('.').map((i)=>parseInt(i)))
+	}
+
+	static encodePort(port) {
+		const data = Buffer.alloc(2)
+		data.writeUInt16BE(port, 0)
+		return data
+	}
+}
+
+class Table{
+	constructor(cap) {
+		this.id = Node.generateID()
+		this.nodes = []
+		this.caption = cap
+	}
+	add(node) {
+		if (this.nodes.length < this.caption) {
+			this.nodes.push(node)
+		}
+	}
+	shift() {
+		return this.nodes.shift()
+	}
+	size() {
+		return this.nodes.length;
+	}
+	first() {
+		if(this.nodes.length >= 8) {
+			return this.nodes.slice(0, 8)
+		}else if(this.nodes.length > 0) {
+			return new Array(8).join().split(',').map(()=> this.nodes[0])
+		}
+		return []
+	}
+}
+
+module.exports = {Table, Node}

src/background/bt/token.js

@@ -0,0 +1,16 @@
+'use strict'
+
+module.exports = class {
+	constructor() {
+		this.generate()
+		setInterval(()=> this.generate(), 60000*15)
+	}
+
+	isValid(t) {
+		return t.toString() === this.token.toString()
+	}
+
+	generate() {
+		this.token = new Buffer([parseInt(Math.random()*200), parseInt(Math.random()*200)])
+	}
+}

src/background/bt/udp-tracker-request.js

@@ -0,0 +1,129 @@
+const dgram = require('dgram');
+const server = dgram.createSocket("udp4")
+const config = require('../config');
+const debug = require('debug')('peers-scrape');
+
+const ACTION_CONNECT = 0
+const ACTION_ANNOUNCE = 1
+const ACTION_SCRAPE = 2
+const ACTION_ERROR = 3
+
+const connectionIdHigh = 0x417
+const connectionIdLow = 0x27101980
+const requests = {};
+
+let message = function (buf, host, port) {
+    server.send(buf, 0, buf.length, port, host, function(err, bytes) {
+        if (err) {
+            console.log(err.message);
+        }
+    });
+};
+
+let connectTracker = function(connection) {
+    debug('start screape connection');
+    let buffer = new Buffer(16);
+
+    const transactionId = Math.floor((Math.random()*100000)+1);
+
+    buffer.fill(0);
+
+    buffer.writeUInt32BE(connectionIdHigh, 0);
+    buffer.writeUInt32BE(connectionIdLow, 4);
+    buffer.writeUInt32BE(ACTION_CONNECT, 8);
+    buffer.writeUInt32BE(transactionId, 12);
+
+    // очистка старых соединений
+    for(transaction in requests) {
+        if((new Date).getTime() - requests[transaction].date.getTime() > config.udpTrackersTimeout) {
+            delete requests[transaction];
+        }
+    }
+
+    requests[transactionId] = connection;
+    message(buffer, connection.host, connection.port);
+};
+
+let scrapeTorrent = function (connectionIdHigh, connectionIdLow, transactionId) {
+    let connection = requests[transactionId];
+    if(!connection)
+        return;
+
+    debug('start scrape');
+    let buffer = new Buffer(56)
+
+    buffer.fill(0);
+
+    buffer.writeUInt32BE(connectionIdHigh, 0);
+    buffer.writeUInt32BE(connectionIdLow, 4);
+    buffer.writeUInt32BE(ACTION_SCRAPE, 8);
+    buffer.writeUInt32BE(transactionId, 12);
+    buffer.write(connection.hash, 16, buffer.length, 'hex');
+
+    // do scrape
+    message(buffer, connection.host, connection.port);
+};
+
+server.on("message", function (msg, rinfo) {
+    let buffer = new Buffer(msg)
+
+    const action = buffer.readUInt32BE(0, 4);
+    const transactionId = buffer.readUInt32BE(4, 4);
+    
+    if(!(transactionId in requests))
+        return;
+
+    debug("returned action: " + action);
+    debug("returned transactionId: " + transactionId);
+
+    if (action === ACTION_CONNECT) {
+        debug("connect response");
+
+        let connectionIdHigh = buffer.readUInt32BE(8, 4);
+        let connectionIdLow = buffer.readUInt32BE(12, 4);
+
+        scrapeTorrent(connectionIdHigh, connectionIdLow, transactionId);
+
+    } else if (action === ACTION_SCRAPE) {
+        debug("scrape response");
+
+        let seeders = buffer.readUInt32BE(8, 4);
+        let completed = buffer.readUInt32BE(12, 4);
+        let leechers = buffer.readUInt32BE(16, 4);
+
+        let connection = requests[transactionId];
+        connection.callback({
+            host: connection.host,
+            port: connection.port,
+            hash: connection.hash,
+            seeders,
+            completed,
+            leechers
+        })
+        delete requests[transactionId];
+    } else if (action === ACTION_ERROR) {
+        delete requests[transactionId];
+        console.log("error in scrape response");
+    }
+});
+
+let getPeersStatistic = (host, port, hash, callback) => {
+    let connection = {
+        host, port, hash, callback, date: new Date()
+    }
+    connectTracker(connection);
+}
+
+server.on("listening", function () {
+    var address = server.address();
+    console.log("listening udp tracker respose on " + address.address + ":" + address.port);
+});
+
+server.bind(config.udpTrackersPort);
+
+module.exports = getPeersStatistic;
+
+//getPeersStatistic('tracker.glotorrents.com', 6969, "d096ff66557a5ea7030680967610e38b37434ea8", (data) => {
+//    console.log(data)
+//});
+

src/background/bt/wire.js

@@ -0,0 +1,247 @@
+'use strict';
+
+var stream = require('stream');
+var crypto = require('crypto');
+var util = require('util');
+
+var BitField = require('bitfield');
+var bencode = require('bencode');
+
+var {Node} = require('./table');
+
+var BT_RESERVED = new Buffer([0x00, 0x00, 0x00, 0x00, 0x00, 0x10, 0x00, 0x01]);
+var BT_PROTOCOL = new Buffer('BitTorrent protocol');
+var PIECE_LENGTH = Math.pow(2, 14);
+var MAX_METADATA_SIZE = 10000000;
+var BITFIELD_GROW = 1000;
+var EXT_HANDSHAKE_ID = 0;
+var BT_MSG_ID = 20;
+
+var Wire = function(infohash) {
+    stream.Duplex.call(this);
+
+    this._bitfield = new BitField(0, { grow: BITFIELD_GROW });
+    this._infohash = infohash;
+
+    this._buffer = [];
+    this._bufferSize = 0;
+
+    this._next = null;
+    this._nextSize = 0;
+
+    this._metadata = null;
+    this._metadataSize = null;
+    this._numPieces = 0;
+    this._ut_metadata = null;
+
+    this._onHandshake();
+}
+
+util.inherits(Wire, stream.Duplex);
+
+Wire.prototype._onMessageLength = function (buffer) {
+    if (buffer.length >= 4) {
+        var length = buffer.readUInt32BE(0);
+        if (length > 0) {
+            this._register(length, this._onMessage)
+        }
+    }
+};
+
+Wire.prototype._onMessage = function (buffer) {
+    this._register(4, this._onMessageLength)
+    if (buffer[0] == BT_MSG_ID) {
+        this._onExtended(buffer.readUInt8(1), buffer.slice(2));
+    }
+};
+
+Wire.prototype._onExtended = function(ext, buf) {
+    if (ext === 0) {
+        try {
+            this._onExtHandshake(bencode.decode(buf));
+        }
+        catch (err) {
+            this._fail();
+        }
+    }
+    else {
+        this._onPiece(buf);
+    }
+};
+
+Wire.prototype._register = function (size, next) {
+    this._nextSize = size;
+    this._next = next;
+};
+
+Wire.prototype.end = function() {
+    stream.Duplex.prototype.end.apply(this, arguments);
+};
+
+Wire.prototype._onHandshake = function() {
+    this._register(1, function(buffer) {
+        if (buffer.length == 0) {
+            this.end();
+            return this._fail();
+        }
+        var pstrlen = buffer.readUInt8(0);
+        this._register(pstrlen + 48, function(handshake) {
+            var protocol = handshake.slice(0, pstrlen);
+            if (protocol.toString() !== BT_PROTOCOL.toString()) {
+                this.end();
+                this._fail();
+                return;
+            }
+            handshake = handshake.slice(pstrlen);
+            if ( !!(handshake[5] & 0x10) ) {
+                this._register(4, this._onMessageLength);
+                this._sendExtHandshake();
+            }
+            else {
+                this._fail();
+            }
+        }.bind(this));
+    }.bind(this));
+};
+
+Wire.prototype._onExtHandshake = function(extHandshake) {
+    if (!extHandshake.metadata_size || !extHandshake.m.ut_metadata
+            || extHandshake.metadata_size > MAX_METADATA_SIZE) {
+        this._fail();
+        return;
+    }
+
+    this._metadataSize = extHandshake.metadata_size;
+    this._numPieces = Math.ceil(this._metadataSize / PIECE_LENGTH);
+    this._ut_metadata = extHandshake.m.ut_metadata;
+
+    this._requestPieces();
+}
+
+Wire.prototype._requestPieces = function() {
+    this._metadata = new Buffer(this._metadataSize);
+    for (var piece = 0; piece < this._numPieces; piece++) {
+        this._requestPiece(piece);
+    }
+};
+
+Wire.prototype._requestPiece = function(piece) {
+    var msg = Buffer.concat([
+        new Buffer([BT_MSG_ID]),
+        new Buffer([this._ut_metadata]),
+        bencode.encode({msg_type: 0, piece: piece})
+    ]);
+    this._sendMessage(msg);
+};
+
+Wire.prototype._sendPacket = function(packet) {
+    this.push(packet);
+};
+
+Wire.prototype._sendMessage = function(msg) {
+    var buf = new Buffer(4);
+    buf.writeUInt32BE(msg.length, 0);
+    this._sendPacket(Buffer.concat([buf, msg]));
+};
+
+Wire.prototype.sendHandshake = function() {
+    var peerID = Node.generateID();
+    var packet = Buffer.concat([
+        new Buffer([BT_PROTOCOL.length]),
+        BT_PROTOCOL, BT_RESERVED, this._infohash,  peerID
+    ]);
+    this._sendPacket(packet);
+};
+
+Wire.prototype._sendExtHandshake = function() {
+    var msg = Buffer.concat([
+        new Buffer([BT_MSG_ID]),
+        new Buffer([EXT_HANDSHAKE_ID]),
+        bencode.encode({m: {ut_metadata: 1}})
+    ]);
+    this._sendMessage(msg);
+};
+
+Wire.prototype._onPiece = function(piece) {
+    var dict, trailer;
+    try {
+        var str = piece.toString();
+        var trailerIndex = str.indexOf('ee') + 2;
+        dict = bencode.decode(str.substring(0, trailerIndex));
+        trailer = piece.slice(trailerIndex);
+    }
+    catch (err) {
+        this._fail();
+        return;
+    }
+    if (dict.msg_type != 1) {
+        this._fail();
+        return;
+    }
+    if (trailer.length > PIECE_LENGTH) {
+        this._fail();
+        return;
+    }
+    trailer.copy(this._metadata, dict.piece * PIECE_LENGTH);
+    this._bitfield.set(dict.piece);
+    this._checkDone();
+};
+
+Wire.prototype._checkDone = function () {
+    var done = true;
+    for (var piece = 0; piece < this._numPieces; piece++) {
+        if (!this._bitfield.get(piece)) {
+            done = false;
+            break;
+        }
+    }
+    if (!done) {
+        return
+    }
+    this._onDone(this._metadata);
+};
+
+Wire.prototype._onDone = function(metadata) {
+    try {
+        var info = bencode.decode(metadata).info;
+        if (info) {
+            metadata = bencode.encode(info);
+        }
+    }
+    catch (err) {
+        this._fail();
+        return;
+    }
+    var infohash = crypto.createHash('sha1').update(metadata).digest('hex');
+    if (this._infohash.toString('hex') != infohash ) {
+        this._fail();
+        return false;
+    }
+    this.emit('metadata', {info: bencode.decode(metadata, 'utf8')}, this._infohash);
+};
+
+Wire.prototype._fail = function() {
+    this.emit('fail');
+};
+
+Wire.prototype._write = function (buf, encoding, next) {
+    this._bufferSize += buf.length;
+    this._buffer.push(buf);
+
+    while (this._bufferSize >= this._nextSize) {
+        var buffer = Buffer.concat(this._buffer);
+        this._bufferSize -= this._nextSize;
+        this._buffer = this._bufferSize
+            ? [buffer.slice(this._nextSize)]
+            : [];
+        this._next(buffer.slice(0, this._nextSize));
+    }
+
+    next(null);
+}
+
+Wire.prototype._read = function() {
+    // do nothing
+};
+
+module.exports = Wire;