pool: refactor packet handling.

lib/net/peer.js

@@ -31,7 +31,6 @@ var NetAddress = require('../primitives/netaddress');
 var services = common.services;
 var invTypes = InvItem.types;
 var packetTypes = packets.types;
-var VerifyResult = errors.VerifyResult;
 
 /**
  * Represents a remote peer.
@@ -1501,6 +1500,13 @@ Peer.prototype.handlePacket = co(function* handlePacket(packet) {
 
   if (entry)
     entry.resolve(packet);
+
+  if (packet.type === packetTypes.UNKNOWN) {
+    this.emit('unknown', packet);
+    return;
+  }
+
+  this.emit(packet.cmd, packet);
 });
 
 /**
@@ -1513,9 +1519,7 @@ Peer.prototype.handlePacket = co(function* handlePacket(packet) {
 Peer.prototype.flushMerkle = co(function* flushMerkle() {
   assert(this.lastMerkle);
 
-  this.emit('merkleblock', this.lastMerkle);
-
-  yield this.pool.handleBlock(this, this.lastMerkle);
+  yield this.pool.addBlock(this, this.lastMerkle);
 
   this.lastMerkle = null;
   this.waitingTX = 0;
@@ -1535,6 +1539,8 @@ Peer.prototype.handleFilterLoad = co(function* handleFilterLoad(packet) {
 
   this.spvFilter = packet.filter;
   this.noRelay = false;
+
+  yield this.pool.handleFilterLoad(this, packet);
 });
 
 /**
@@ -1555,12 +1561,14 @@ Peer.prototype.handleFilterAdd = co(function* handleFilterAdd(packet) {
     this.spvFilter.add(data);
 
   this.noRelay = false;
+
+  yield this.pool.handleFilterAdd(this, packet);
 });
 
 /**
  * Handle `filterclear` packet.
  * @private
- * @param {FilterClearPacket}
+ * @param {FilterClearPacket} packet
  */
 
 Peer.prototype.handleFilterClear = co(function* handleFilterClear(packet) {
@@ -1568,54 +1576,24 @@ Peer.prototype.handleFilterClear = co(function* handleFilterClear(packet) {
     this.spvFilter.reset();
 
   this.noRelay = false;
+
+  yield this.pool.handleFilterClear(this, packet);
 });
 
 /**
  * Handle `merkleblock` packet.
  * @private
- * @param {MerkleBlockPacket}
+ * @param {MerkleBlockPacket} packet
  */
 
 Peer.prototype.handleMerkleBlock = co(function* handleMerkleBlock(packet) {
-  var block = packet.block;
-  var ret = new VerifyResult();
-
-  // Potential DoS.
-  if (!this.options.spv) {
-    this.logger.warning(
-      'Peer sent unsolicited merkleblock (%s).',
-      this.hostname);
-    this.increaseBan(100);
-    return;
-  }
-
-  if (this.lastMerkle) {
-    this.logger.warning(
-      'Peer sent a merkleblock prematurely (%s).',
-      this.hostname);
-    this.increaseBan(100);
-    return;
-  }
-
-  if (!block.verify(ret)) {
-    this.logger.warning(
-      'Peer sent an invalid merkleblock (%s).',
-      this.hostname);
-    this.reject(block, 'invalid', ret.reason, ret.score);
-    return;
-  }
-
-  this.lastMerkle = block;
-  this.waitingTX = block.matches.length;
-
-  if (this.waitingTX === 0)
-    yield this.flushMerkle();
+  yield this.pool.handleMerkleBlock(this, packet);
 });
 
 /**
  * Handle `feefilter` packet.
  * @private
- * @param {FeeFilterPacket}
+ * @param {FeeFilterPacket} packet
  */
 
 Peer.prototype.handleFeeFilter = co(function* handleFeeFilter(packet) {
@@ -1628,28 +1606,26 @@ Peer.prototype.handleFeeFilter = co(function* handleFeeFilter(packet) {
 
   this.feeRate = rate;
 
-  this.emit('feefilter', rate);
+  yield this.pool.handleFeeFilter(this, packet);
 });
 
 /**
  * Handle `getheaders` packet.
  * @private
- * @param {GetHeadersPacket}
+ * @param {GetHeadersPacket} packet
  */
 
 Peer.prototype.handleGetHeaders = co(function* handleGetHeaders(packet) {
-  this.emit('getheaders', packet);
   yield this.pool.handleGetHeaders(this, packet);
 });
 
 /**
  * Handle `getblocks` packet.
  * @private
- * @param {GetBlocksPacket}
+ * @param {GetBlocksPacket} packet
  */
 
 Peer.prototype.handleGetBlocks = co(function* handleGetBlocks(packet) {
-  this.emit('getblocks', packet);
   yield this.pool.handleGetBlocks(this, packet);
 });
 
@@ -1700,8 +1676,6 @@ Peer.prototype.handleVersion = co(function* handleVersion(packet) {
     }
   }
 
-  this.emit('version', packet);
-
   this.send(new packets.VerackPacket());
 
   yield this.pool.handleVersion(this, packet);
@@ -1710,75 +1684,77 @@ Peer.prototype.handleVersion = co(function* handleVersion(packet) {
 /**
  * Handle `verack` packet.
  * @private
- * @param {VerackPacket}
+ * @param {VerackPacket} packet
  */
 
 Peer.prototype.handleVerack = co(function* handleVerack(packet) {
+  if (this.ack) {
+    this.logger.debug('Peer sent duplicate ack (%s).', this.hostname);
+    return;
+  }
+
   this.ack = true;
-  this.emit('verack', packet);
   this.logger.debug('Received verack (%s).', this.hostname);
+
+  yield this.pool.handleVerack(this, packet);
 });
 
 /**
  * Handle `mempool` packet.
  * @private
- * @param {MempoolPacket}
+ * @param {MempoolPacket} packet
  */
 
 Peer.prototype.handleMempool = co(function* handleMempool(packet) {
-  this.emit('mempool', packet);
-  yield this.pool.handleMempool(this);
+  yield this.pool.handleMempool(this, packet);
 });
 
 /**
  * Handle `getdata` packet.
  * @private
- * @param {GetDataPacket}
+ * @param {GetDataPacket} packet
  */
 
 Peer.prototype.handleGetData = co(function* handleGetData(packet) {
-  this.emit('getdata', packet.items);
-  yield this.pool.handleGetData(this, packet.items);
+  yield this.pool.handleGetData(this, packet);
 });
 
 /**
  * Handle `notfound` packet.
  * @private
- * @param {NotFoundPacket}
+ * @param {NotFoundPacket} packet
  */
 
 Peer.prototype.handleNotFound = co(function* handleNotFound(packet) {
-  this.emit('notfound', packet.items);
-  yield this.pool.handleNotFound(this, packet.items);
+  yield this.pool.handleNotFound(this, packet);
 });
 
 /**
  * Handle `addr` packet.
  * @private
- * @param {AddrPacket}
+ * @param {AddrPacket} packet
  */
 
 Peer.prototype.handleAddr = co(function* handleAddr(packet) {
-  this.emit('addr', packet.items);
-  yield this.pool.handleAddr(this, packet.items);
+  yield this.pool.handleAddr(this, packet);
 });
 
 /**
  * Handle `ping` packet.
  * @private
- * @param {PingPacket}
+ * @param {PingPacket} packet
  */
 
 Peer.prototype.handlePing = co(function* handlePing(packet) {
-  this.emit('ping', this.minPing);
   if (packet.nonce)
     this.send(new packets.PongPacket(packet.nonce));
+  yield this.pool.handlePing(this, packet);
 });
 
 /**
  * Handle `pong` packet.
  * @private
- * @param {PongPacket}
+ * @param {PongPacket} packet
  */
 
 Peer.prototype.handlePong = co(function* handlePong(packet) {
@@ -1811,111 +1787,84 @@ Peer.prototype.handlePong = co(function* handlePong(packet) {
 
   this.challenge = null;
 
-  this.emit('pong', this.minPing);
+  yield this.pool.handlePong(this, packet);
 });
 
 /**
  * Handle `getaddr` packet.
  * @private
- * @param {GetAddrPacket}
+ * @param {GetAddrPacket} packet
  */
 
 Peer.prototype.handleGetAddr = co(function* handleGetAddr(packet) {
-  this.emit('getaddr', packet);
   yield this.pool.handleGetAddr(this, packet);
 });
 
 /**
  * Handle `inv` packet.
  * @private
- * @param {InvPacket}
+ * @param {InvPacket} packet
  */
 
 Peer.prototype.handleInv = co(function* handleInv(packet) {
-  this.emit('inv', packet.items);
-  yield this.pool.handleInv(this, packet.items);
+  yield this.pool.handleInv(this, packet);
 });
 
 /**
  * Handle `headers` packet.
  * @private
- * @param {HeadersPacket}
+ * @param {HeadersPacket} packet
  */
 
 Peer.prototype.handleHeaders = co(function* handleHeaders(packet) {
-  this.emit('headers', packet.items);
-  yield this.pool.handleHeaders(this, packet.items);
+  yield this.pool.handleHeaders(this, packet);
 });
 
 /**
  * Handle `sendheaders` packet.
  * @private
- * @param {SendHeadersPacket}
+ * @param {SendHeadersPacket} packet
  */
 
 Peer.prototype.handleSendHeaders = co(function* handleSendHeaders(packet) {
   this.preferHeaders = true;
-  this.emit('sendheaders');
+  yield this.pool.handleSendHeaders(this, packet);
 });
 
 /**
  * Handle `block` packet.
  * @private
- * @param {BlockPacket}
+ * @param {BlockPacket} packet
  */
 
 Peer.prototype.handleBlock = co(function* handleBlock(packet) {
-  if (this.options.spv) {
-    this.logger.warning(
-      'Peer sent unsolicited block (%s).',
-      this.hostname);
-    return;
-  }
-
-  this.emit('block', packet.block);
-
-  yield this.pool.handleBlock(this, packet.block);
+  yield this.pool.handleBlock(this, packet);
 });
 
 /**
  * Handle `tx` packet.
  * @private
- * @param {TXPacket}
+ * @param {TXPacket} packet
  */
 
 Peer.prototype.handleTX = co(function* handleTX(packet) {
-  var tx = packet.tx;
-
-  if (this.lastMerkle) {
-    assert(this.waitingTX > 0);
-    if (this.lastMerkle.hasTX(tx)) {
-      this.lastMerkle.addTX(tx);
-      if (--this.waitingTX === 0)
-        yield this.flushMerkle();
-      return;
-    }
-  }
-
-  this.emit('tx', tx);
-
-  yield this.pool.handleTX(this, tx);
+  yield this.pool.handleTX(this, packet);
 });
 
 /**
  * Handle `reject` packet.
  * @private
- * @param {RejectPacket} reject
+ * @param {RejectPacket} packet
  */
 
-Peer.prototype.handleReject = co(function* handleReject(reject) {
-  this.emit('reject', reject);
-  yield this.pool.handleReject(this, reject);
+Peer.prototype.handleReject = co(function* handleReject(packet) {
+  yield this.pool.handleReject(this, packet);
 });
 
 /**
  * Handle `encinit` packet.
  * @private
- * @param {EncinitPacket}
+ * @param {EncinitPacket} packet
  */
 
 Peer.prototype.handleEncinit = co(function* handleEncinit(packet) {
@@ -1924,15 +1873,15 @@ Peer.prototype.handleEncinit = co(function* handleEncinit(packet) {
 
   this.bip151.encinit(packet.publicKey, packet.cipher);
 
-  this.emit('encinit', packet);
-
   this.send(this.bip151.toEncack());
+
+  yield this.pool.handleEncinit(this, packet);
 });
 
 /**
  * Handle `encack` packet.
  * @private
- * @param {EncackPacket}
+ * @param {EncackPacket} packet
  */
 
 Peer.prototype.handleEncack = co(function* handleEncack(packet) {
@@ -1941,13 +1890,13 @@ Peer.prototype.handleEncack = co(function* handleEncack(packet) {
 
   this.bip151.encack(packet.publicKey);
 
-  this.emit('encack', packet);
+  yield this.pool.handleEncack(this, packet);
 });
 
 /**
  * Handle `authchallenge` packet.
  * @private
- * @param {AuthChallengePacket}
+ * @param {AuthChallengePacket} packet
  */
 
 Peer.prototype.handleAuthChallenge = co(function* handleAuthChallenge(packet) {
@@ -1958,15 +1907,15 @@ Peer.prototype.handleAuthChallenge = co(function* handleAuthChallenge(packet) {
 
   sig = this.bip150.challenge(packet.hash);
 
-  this.emit('authchallenge', packet.hash);
-
   this.send(new packets.AuthReplyPacket(sig));
+
+  yield this.pool.handleAuthChallenge(this, packet);
 });
 
 /**
  * Handle `authreply` packet.
  * @private
- * @param {AuthReplyPacket}
+ * @param {AuthReplyPacket} packet
  */
 
 Peer.prototype.handleAuthReply = co(function* handleAuthReply(packet) {
@@ -1980,13 +1929,13 @@ Peer.prototype.handleAuthReply = co(function* handleAuthReply(packet) {
   if (hash)
     this.send(new packets.AuthProposePacket(hash));
 
-  this.emit('authreply', packet.signature);
+  yield this.pool.handleAuthReply(this, packet);
 });
 
 /**
  * Handle `authpropose` packet.
  * @private
- * @param {AuthProposePacket}
+ * @param {AuthProposePacket} packet
  */
 
 Peer.prototype.handleAuthPropose = co(function* handleAuthPropose(packet) {
@@ -1999,18 +1948,17 @@ Peer.prototype.handleAuthPropose = co(function* handleAuthPropose(packet) {
 
   this.send(new packets.AuthChallengePacket(hash));
 
-  this.emit('authpropose', packet.hash);
+  yield this.pool.handleAuthPropose(this, packet);
 });
 
 /**
  * Handle an unknown packet.
  * @private
- * @param {UnknownPacket}
+ * @param {UnknownPacket} packet
  */
 
 Peer.prototype.handleUnknown = co(function* handleUnknown(packet) {
-  this.logger.warning('Unknown packet: %s.', packet.cmd);
-  this.emit('unknown', packet);
+  yield this.pool.handleUnknown(this, packet);
 });
 
 /**
@@ -2046,7 +1994,8 @@ Peer.prototype.handleSendCmpct = co(function* handleSendCmpct(packet) {
 
   this.compactMode = packet.mode;
   this.compactWitness = packet.version === 2;
-  this.emit('sendcmpct', packet);
+
+  yield this.pool.handleSendCmpct(this, packet);
 });
 
 /**
@@ -2056,8 +2005,7 @@ Peer.prototype.handleSendCmpct = co(function* handleSendCmpct(packet) {
  */
 
 Peer.prototype.handleCmpctBlock = co(function* handleCmpctBlock(packet) {
-  this.emit('cmpctblock', packet.block);
-  yield this.pool.handleCmpctBlock(this, packet.block);
+  yield this.pool.handleCmpctBlock(this, packet);
 });
 
 /**
@@ -2067,8 +2015,7 @@ Peer.prototype.handleCmpctBlock = co(function* handleCmpctBlock(packet) {
  */
 
 Peer.prototype.handleGetBlockTxn = co(function* handleGetBlockTxn(packet) {
-  this.emit('getblocktxn', packet.request);
-  yield this.pool.handleGetBlockTxn(this, packet.request);
+  yield this.pool.handleGetBlockTxn(this, packet);
 });
 
 /**
@@ -2078,8 +2025,7 @@ Peer.prototype.handleGetBlockTxn = co(function* handleGetBlockTxn(packet) {
  */
 
 Peer.prototype.handleBlockTxn = co(function* handleBlockTxn(packet) {
-  this.emit('blocktxn', packet.response);
-  yield this.pool.handleGetBlockTxn(this, packet.response);
+  yield this.pool.handleGetBlockTxn(this, packet);
 });
 
 /**