chainntnfs+chainreg: add mempool watcher to BitcoindNotifier

This commit adds the mempool watcher to bitcoind notifier to allow the
notifier managing the starting and stopping of the watcher.

chainntnfs/bitcoindnotify/bitcoind.go

@@ -69,6 +69,9 @@ type BitcoindNotifier struct {
 	// which the transaction could have confirmed within the chain.
 	confirmHintCache chainntnfs.ConfirmHintCache
 
+	// memNotifier notifies clients of events related to the mempool.
+	memNotifier *chainntnfs.MempoolNotifier
+
 	wg   sync.WaitGroup
 	quit chan struct{}
 }
@@ -96,7 +99,8 @@ func New(chainConn *chain.BitcoindConn, chainParams *chaincfg.Params,
 		spendHintCache:   spendHintCache,
 		confirmHintCache: confirmHintCache,
 
-		blockCache: blockCache,
+		blockCache:  blockCache,
+		memNotifier: chainntnfs.NewMempoolNotifier(),
 
 		quit: make(chan struct{}),
 	}
@@ -113,6 +117,7 @@ func (b *BitcoindNotifier) Start() error {
 	b.start.Do(func() {
 		startErr = b.startNotifier()
 	})
+
 	return startErr
 }
 
@@ -142,6 +147,9 @@ func (b *BitcoindNotifier) Stop() error {
 	}
 	b.txNotifier.TearDown()
 
+	// Stop the mempool notifier.
+	b.memNotifier.TearDown()
+
 	return nil
 }
 
@@ -441,15 +449,17 @@ out:
 				b.bestBlock = newBestBlock
 
 			case chain.RelevantTx:
-				// We only care about notifying on confirmed
-				// spends, so if this is a mempool spend, we can
-				// ignore it and wait for the spend to appear in
-				// on-chain.
+				tx := btcutil.NewTx(&item.TxRecord.MsgTx)
+
+				// If this is a mempool spend, we'll ask the
+				// mempool notifier to hanlde it.
 				if item.Block == nil {
+					b.memNotifier.ProcessRelevantSpendTx(tx)
 					continue
 				}
 
-				tx := btcutil.NewTx(&item.TxRecord.MsgTx)
+				// Otherwise this is a confirmed spend, and
+				// we'll ask the tx notifier to handle it.
 				err := b.txNotifier.ProcessRelevantSpendTx(
 					tx, uint32(item.Block.Height),
 				)
@@ -993,3 +1003,28 @@ func (b *BitcoindNotifier) GetBlock(hash *chainhash.Hash) (*wire.MsgBlock,
 
 	return b.blockCache.GetBlock(hash, b.chainConn.GetBlock)
 }
+
+// SubscribeMempoolSpent allows the caller to register a subscription to watch
+// for a spend of an outpoint in the mempool.The event will be dispatched once
+// the outpoint is spent in the mempool.
+//
+// NOTE: part of the MempoolWatcher interface.
+func (b *BitcoindNotifier) SubscribeMempoolSpent(
+	outpoint wire.OutPoint) (*chainntnfs.MempoolSpendEvent, error) {
+
+	event := b.memNotifier.SubscribeInput(outpoint)
+
+	ops := []*wire.OutPoint{&outpoint}
+
+	return event, b.chainConn.NotifySpent(ops)
+}
+
+// CancelMempoolSpendEvent allows the caller to cancel a subscription to watch
+// for a spend of an outpoint in the mempool.
+//
+// NOTE: part of the MempoolWatcher interface.
+func (b *BitcoindNotifier) CancelMempoolSpendEvent(
+	sub *chainntnfs.MempoolSpendEvent) {
+
+	b.memNotifier.UnsubscribeEvent(sub)
+}