chainntnfs+chainreg: add mempool watcher to BtcdNotifier

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

chainntnfs/btcdnotify/btcd.go

@@ -86,6 +86,9 @@ type BtcdNotifier 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{}
 }
@@ -115,7 +118,8 @@ func New(config *rpcclient.ConnConfig, chainParams *chaincfg.Params,
 		spendHintCache:   spendHintCache,
 		confirmHintCache: confirmHintCache,
 
-		blockCache: blockCache,
+		blockCache:  blockCache,
+		memNotifier: chainntnfs.NewMempoolNotifier(),
 
 		quit: make(chan struct{}),
 	}
@@ -146,6 +150,7 @@ func (b *BtcdNotifier) Start() error {
 	b.start.Do(func() {
 		startErr = b.startNotifier()
 	})
+
 	return startErr
 }
 
@@ -183,6 +188,9 @@ func (b *BtcdNotifier) Stop() error {
 	}
 	b.txNotifier.TearDown()
 
+	// Stop the mempool notifier.
+	b.memNotifier.TearDown()
+
 	return nil
 }
 
@@ -492,16 +500,19 @@ out:
 
 		case item := <-b.txUpdates.ChanOut():
 			newSpend := item.(*txUpdate)
+			tx := newSpend.tx
 
-			// 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.
+			// If this is a mempool spend, we'll ask the mempool
+			// notifier to hanlde it.
 			if newSpend.details == nil {
+				b.memNotifier.ProcessRelevantSpendTx(tx)
 				continue
 			}
 
+			// Otherwise this is a confirmed spend, and we'll ask
+			// the tx notifier to handle it.
 			err := b.txNotifier.ProcessRelevantSpendTx(
-				newSpend.tx, uint32(newSpend.details.Height),
+				tx, uint32(newSpend.details.Height),
 			)
 			if err != nil {
 				chainntnfs.Log.Errorf("Unable to process "+
@@ -1051,3 +1062,28 @@ func (b *BtcdNotifier) 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 *BtcdNotifier) 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 *BtcdNotifier) CancelMempoolSpendEvent(
+	sub *chainntnfs.MempoolSpendEvent) {
+
+	b.memNotifier.UnsubscribeEvent(sub)
+}