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.
This commit is contained in:
1 file changed
+41
-5
@@ -86,6 +86,9 @@ type BtcdNotifier struct {
|
|||||||
// which the transaction could have confirmed within the chain.
|
// which the transaction could have confirmed within the chain.
|
||||||
confirmHintCache chainntnfs.ConfirmHintCache
|
confirmHintCache chainntnfs.ConfirmHintCache
|
||||||
|
|
||||||
|
// memNotifier notifies clients of events related to the mempool.
|
||||||
|
memNotifier *chainntnfs.MempoolNotifier
|
||||||
|
|
||||||
wg sync.WaitGroup
|
wg sync.WaitGroup
|
||||||
quit chan struct{}
|
quit chan struct{}
|
||||||
}
|
}
|
||||||
@@ -115,7 +118,8 @@ func New(config *rpcclient.ConnConfig, chainParams *chaincfg.Params,
|
|||||||
spendHintCache: spendHintCache,
|
spendHintCache: spendHintCache,
|
||||||
confirmHintCache: confirmHintCache,
|
confirmHintCache: confirmHintCache,
|
||||||
|
|
||||||
blockCache: blockCache,
|
blockCache: blockCache,
|
||||||
|
memNotifier: chainntnfs.NewMempoolNotifier(),
|
||||||
|
|
||||||
quit: make(chan struct{}),
|
quit: make(chan struct{}),
|
||||||
}
|
}
|
||||||
@@ -146,6 +150,7 @@ func (b *BtcdNotifier) Start() error {
|
|||||||
b.start.Do(func() {
|
b.start.Do(func() {
|
||||||
startErr = b.startNotifier()
|
startErr = b.startNotifier()
|
||||||
})
|
})
|
||||||
|
|
||||||
return startErr
|
return startErr
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -183,6 +188,9 @@ func (b *BtcdNotifier) Stop() error {
|
|||||||
}
|
}
|
||||||
b.txNotifier.TearDown()
|
b.txNotifier.TearDown()
|
||||||
|
|
||||||
|
// Stop the mempool notifier.
|
||||||
|
b.memNotifier.TearDown()
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -492,16 +500,19 @@ out:
|
|||||||
|
|
||||||
case item := <-b.txUpdates.ChanOut():
|
case item := <-b.txUpdates.ChanOut():
|
||||||
newSpend := item.(*txUpdate)
|
newSpend := item.(*txUpdate)
|
||||||
|
tx := newSpend.tx
|
||||||
|
|
||||||
// We only care about notifying on confirmed spends, so
|
// If this is a mempool spend, we'll ask the mempool
|
||||||
// if this is a mempool spend, we can ignore it and wait
|
// notifier to hanlde it.
|
||||||
// for the spend to appear in on-chain.
|
|
||||||
if newSpend.details == nil {
|
if newSpend.details == nil {
|
||||||
|
b.memNotifier.ProcessRelevantSpendTx(tx)
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Otherwise this is a confirmed spend, and we'll ask
|
||||||
|
// the tx notifier to handle it.
|
||||||
err := b.txNotifier.ProcessRelevantSpendTx(
|
err := b.txNotifier.ProcessRelevantSpendTx(
|
||||||
newSpend.tx, uint32(newSpend.details.Height),
|
tx, uint32(newSpend.details.Height),
|
||||||
)
|
)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
chainntnfs.Log.Errorf("Unable to process "+
|
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)
|
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)
|
||||||
|
}
|
||||||
Reference in new issue
Block a user