2016-07-13 03:14:07 +03:00
|
|
|
package main
|
|
|
|
|
|
|
|
import (
|
|
|
|
"bytes"
|
2017-03-16 04:56:25 +03:00
|
|
|
"crypto/sha256"
|
2018-04-25 07:07:30 +03:00
|
|
|
"fmt"
|
2016-07-13 03:14:07 +03:00
|
|
|
"sync"
|
2018-04-25 07:07:30 +03:00
|
|
|
"sync/atomic"
|
2016-09-19 21:55:02 +03:00
|
|
|
"time"
|
2016-07-13 03:14:07 +03:00
|
|
|
|
2016-09-21 03:00:11 +03:00
|
|
|
"github.com/davecgh/go-spew/spew"
|
2018-04-25 07:07:30 +03:00
|
|
|
"github.com/lightningnetwork/lnd/chainntnfs"
|
2016-09-19 21:55:02 +03:00
|
|
|
"github.com/lightningnetwork/lnd/channeldb"
|
2017-08-22 09:25:41 +03:00
|
|
|
"github.com/lightningnetwork/lnd/lnwire"
|
2018-06-30 02:03:46 +03:00
|
|
|
"github.com/lightningnetwork/lnd/zpay32"
|
2017-01-06 00:56:27 +03:00
|
|
|
"github.com/roasbeef/btcd/chaincfg/chainhash"
|
2016-07-13 03:14:07 +03:00
|
|
|
"github.com/roasbeef/btcutil"
|
|
|
|
)
|
|
|
|
|
2016-09-19 21:55:02 +03:00
|
|
|
var (
|
|
|
|
// debugPre is the default debug preimage which is inserted into the
|
|
|
|
// invoice registry if the --debughtlc flag is activated on start up.
|
2017-01-13 08:01:50 +03:00
|
|
|
// All nodes initialized with the flag active will immediately settle
|
|
|
|
// any incoming HTLC whose rHash corresponds with the debug
|
2016-09-19 21:55:02 +03:00
|
|
|
// preimage.
|
2017-01-06 00:56:27 +03:00
|
|
|
debugPre, _ = chainhash.NewHash(bytes.Repeat([]byte{1}, 32))
|
2016-07-13 03:14:07 +03:00
|
|
|
|
2017-03-16 04:56:25 +03:00
|
|
|
debugHash = chainhash.Hash(sha256.Sum256(debugPre[:]))
|
2016-09-19 21:55:02 +03:00
|
|
|
)
|
2016-07-13 03:14:07 +03:00
|
|
|
|
|
|
|
// invoiceRegistry is a central registry of all the outstanding invoices
|
|
|
|
// created by the daemon. The registry is a thin wrapper around a map in order
|
|
|
|
// to ensure that all updates/reads are thread safe.
|
|
|
|
type invoiceRegistry struct {
|
|
|
|
sync.RWMutex
|
2016-09-19 21:55:02 +03:00
|
|
|
|
|
|
|
cdb *channeldb.DB
|
|
|
|
|
2016-10-15 05:47:10 +03:00
|
|
|
clientMtx sync.Mutex
|
|
|
|
nextClientID uint32
|
|
|
|
notificationClients map[uint32]*invoiceSubscription
|
|
|
|
|
2018-04-25 07:07:30 +03:00
|
|
|
newSubscriptions chan *invoiceSubscription
|
|
|
|
subscriptionCancels chan uint32
|
|
|
|
invoiceEvents chan *invoiceEvent
|
|
|
|
|
2017-01-13 08:01:50 +03:00
|
|
|
// debugInvoices is a map which stores special "debug" invoices which
|
2016-09-19 21:55:02 +03:00
|
|
|
// should be only created/used when manual tests require an invoice
|
|
|
|
// that *all* nodes are able to fully settle.
|
2017-01-06 00:56:27 +03:00
|
|
|
debugInvoices map[chainhash.Hash]*channeldb.Invoice
|
2018-04-25 07:07:30 +03:00
|
|
|
|
|
|
|
wg sync.WaitGroup
|
|
|
|
quit chan struct{}
|
2016-07-13 03:14:07 +03:00
|
|
|
}
|
|
|
|
|
2016-09-19 21:55:02 +03:00
|
|
|
// newInvoiceRegistry creates a new invoice registry. The invoice registry
|
|
|
|
// wraps the persistent on-disk invoice storage with an additional in-memory
|
2017-01-13 08:01:50 +03:00
|
|
|
// layer. The in-memory layer is in place such that debug invoices can be added
|
2016-09-19 21:55:02 +03:00
|
|
|
// which are volatile yet available system wide within the daemon.
|
|
|
|
func newInvoiceRegistry(cdb *channeldb.DB) *invoiceRegistry {
|
2016-07-13 03:14:07 +03:00
|
|
|
return &invoiceRegistry{
|
2016-10-15 05:47:10 +03:00
|
|
|
cdb: cdb,
|
2017-01-06 00:56:27 +03:00
|
|
|
debugInvoices: make(map[chainhash.Hash]*channeldb.Invoice),
|
2016-10-15 05:47:10 +03:00
|
|
|
notificationClients: make(map[uint32]*invoiceSubscription),
|
2018-04-25 07:07:30 +03:00
|
|
|
newSubscriptions: make(chan *invoiceSubscription),
|
|
|
|
subscriptionCancels: make(chan uint32),
|
|
|
|
invoiceEvents: make(chan *invoiceEvent),
|
|
|
|
quit: make(chan struct{}),
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// Start starts the registry and all goroutines it needs to carry out its task.
|
|
|
|
func (i *invoiceRegistry) Start() error {
|
|
|
|
i.wg.Add(1)
|
|
|
|
|
|
|
|
go i.invoiceEventNotifier()
|
|
|
|
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
// Stop signals the registry for a graceful shutdown.
|
|
|
|
func (i *invoiceRegistry) Stop() {
|
|
|
|
close(i.quit)
|
|
|
|
|
|
|
|
i.wg.Wait()
|
|
|
|
}
|
|
|
|
|
|
|
|
// invoiceEvent represents a new event that has modified on invoice on disk.
|
|
|
|
// Only two event types are currently supported: newly created invoices, and
|
|
|
|
// instance where invoices are settled.
|
|
|
|
type invoiceEvent struct {
|
|
|
|
isSettle bool
|
|
|
|
|
|
|
|
invoice *channeldb.Invoice
|
|
|
|
}
|
|
|
|
|
|
|
|
// invoiceEventNotifier is the dedicated goroutine responsible for accepting
|
|
|
|
// new notification subscriptions, cancelling old subscriptions, and
|
|
|
|
// dispatching new invoice events.
|
|
|
|
func (i *invoiceRegistry) invoiceEventNotifier() {
|
|
|
|
defer i.wg.Done()
|
|
|
|
|
|
|
|
for {
|
|
|
|
select {
|
|
|
|
// A new invoice subscription has just arrived! We'll query for
|
|
|
|
// any backlog notifications, then add it to the set of
|
|
|
|
// clients.
|
|
|
|
case newClient := <-i.newSubscriptions:
|
|
|
|
// Before we add the client to our set of active
|
|
|
|
// clients, we'll first attempt to deliver any backlog
|
|
|
|
// invoice events.
|
|
|
|
err := i.deliverBacklogEvents(newClient)
|
|
|
|
if err != nil {
|
|
|
|
ltndLog.Errorf("unable to deliver backlog invoice "+
|
|
|
|
"notifications: %v", err)
|
|
|
|
}
|
|
|
|
|
|
|
|
ltndLog.Infof("New invoice subscription "+
|
|
|
|
"client: id=%v", newClient.id)
|
|
|
|
|
|
|
|
// With the backlog notifications delivered (if any),
|
|
|
|
// we'll add this to our active subscriptions and
|
|
|
|
// continue.
|
|
|
|
i.notificationClients[newClient.id] = newClient
|
|
|
|
|
|
|
|
// A client no longer wishes to receive invoice notifications.
|
|
|
|
// So we'll remove them from the set of active clients.
|
|
|
|
case clientID := <-i.subscriptionCancels:
|
|
|
|
ltndLog.Infof("Cancelling invoice subscription for "+
|
|
|
|
"client=%v", clientID)
|
|
|
|
|
|
|
|
delete(i.notificationClients, clientID)
|
|
|
|
|
|
|
|
// A sub-systems has just modified the invoice state, so we'll
|
|
|
|
// dispatch notifications to all registered clients.
|
|
|
|
case event := <-i.invoiceEvents:
|
|
|
|
for _, client := range i.notificationClients {
|
|
|
|
select {
|
|
|
|
case client.ntfnQueue.ChanIn() <- &invoiceEvent{
|
|
|
|
isSettle: event.isSettle,
|
|
|
|
invoice: event.invoice,
|
|
|
|
}:
|
|
|
|
case <-i.quit:
|
|
|
|
return
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
case <-i.quit:
|
|
|
|
return
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// deliverBacklogEvents will attempts to query the invoice database for any
|
|
|
|
// notifications that the client has missed since it reconnected last.
|
|
|
|
func (i *invoiceRegistry) deliverBacklogEvents(client *invoiceSubscription) error {
|
|
|
|
// First, we'll query the database to see if based on the provided
|
|
|
|
// addIndex and settledIndex we need to deliver any backlog
|
|
|
|
// notifications.
|
|
|
|
addEvents, err := i.cdb.InvoicesAddedSince(client.addIndex)
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
settleEvents, err := i.cdb.InvoicesSettledSince(client.settleIndex)
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
// If we have any to deliver, then we'll append them to the end of the
|
|
|
|
// notification queue in order to catch up the client before delivering
|
|
|
|
// any new notifications.
|
|
|
|
for _, addEvent := range addEvents {
|
|
|
|
select {
|
|
|
|
case client.ntfnQueue.ChanIn() <- &invoiceEvent{
|
|
|
|
isSettle: false,
|
|
|
|
invoice: &addEvent,
|
|
|
|
}:
|
|
|
|
case <-i.quit:
|
|
|
|
return fmt.Errorf("registry shutting down")
|
|
|
|
}
|
|
|
|
}
|
|
|
|
for _, settleEvent := range settleEvents {
|
|
|
|
select {
|
|
|
|
case client.ntfnQueue.ChanIn() <- &invoiceEvent{
|
|
|
|
isSettle: true,
|
|
|
|
invoice: &settleEvent,
|
|
|
|
}:
|
|
|
|
case <-i.quit:
|
|
|
|
return fmt.Errorf("registry shutting down")
|
|
|
|
}
|
2016-07-13 03:14:07 +03:00
|
|
|
}
|
2018-04-25 07:07:30 +03:00
|
|
|
|
|
|
|
return nil
|
2016-07-13 03:14:07 +03:00
|
|
|
}
|
|
|
|
|
2018-07-03 06:20:17 +03:00
|
|
|
// AddDebugInvoice adds a debug invoice for the specified amount, identified
|
2017-01-13 08:01:50 +03:00
|
|
|
// by the passed preimage. Once this invoice is added, subsystems within the
|
2018-07-03 06:20:17 +03:00
|
|
|
// daemon add/forward HTLCs that are able to obtain the proper preimage
|
|
|
|
// required for redemption in the case that we're the final destination.
|
2017-01-06 00:56:27 +03:00
|
|
|
func (i *invoiceRegistry) AddDebugInvoice(amt btcutil.Amount, preimage chainhash.Hash) {
|
2017-03-16 04:56:25 +03:00
|
|
|
paymentHash := chainhash.Hash(sha256.Sum256(preimage[:]))
|
2016-07-13 03:14:07 +03:00
|
|
|
|
2016-09-21 03:00:11 +03:00
|
|
|
invoice := &channeldb.Invoice{
|
2016-09-19 21:55:02 +03:00
|
|
|
CreationDate: time.Now(),
|
|
|
|
Terms: channeldb.ContractTerm{
|
2017-08-22 09:25:41 +03:00
|
|
|
Value: lnwire.NewMSatFromSatoshis(amt),
|
2016-09-19 21:55:02 +03:00
|
|
|
PaymentPreimage: preimage,
|
|
|
|
},
|
2016-07-13 03:14:07 +03:00
|
|
|
}
|
2016-09-21 03:00:11 +03:00
|
|
|
|
|
|
|
i.Lock()
|
|
|
|
i.debugInvoices[paymentHash] = invoice
|
2016-07-13 03:14:07 +03:00
|
|
|
i.Unlock()
|
2016-09-21 03:00:11 +03:00
|
|
|
|
|
|
|
ltndLog.Debugf("Adding debug invoice %v", newLogClosure(func() string {
|
|
|
|
return spew.Sdump(invoice)
|
|
|
|
}))
|
2016-07-13 03:14:07 +03:00
|
|
|
}
|
|
|
|
|
2016-09-19 21:55:02 +03:00
|
|
|
// AddInvoice adds a regular invoice for the specified amount, identified by
|
2017-01-13 08:01:50 +03:00
|
|
|
// the passed preimage. Additionally, any memo or receipt data provided will
|
|
|
|
// also be stored on-disk. Once this invoice is added, subsystems within the
|
2018-04-25 07:07:30 +03:00
|
|
|
// daemon add/forward HTLCs are able to obtain the proper preimage required for
|
|
|
|
// redemption in the case that we're the final destination.
|
2016-09-19 21:55:02 +03:00
|
|
|
func (i *invoiceRegistry) AddInvoice(invoice *channeldb.Invoice) error {
|
2016-09-21 03:00:11 +03:00
|
|
|
ltndLog.Debugf("Adding invoice %v", newLogClosure(func() string {
|
|
|
|
return spew.Sdump(invoice)
|
|
|
|
}))
|
|
|
|
|
2018-04-25 07:07:30 +03:00
|
|
|
if err := i.cdb.AddInvoice(invoice); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
// We'll launch a new goroutine to notify all of our active listeners
|
|
|
|
// that a new invoice was just added.
|
|
|
|
i.notifyClients(invoice, false)
|
2016-10-15 05:47:10 +03:00
|
|
|
|
2018-04-25 07:07:30 +03:00
|
|
|
return nil
|
2016-09-19 21:55:02 +03:00
|
|
|
}
|
|
|
|
|
2018-06-30 02:03:46 +03:00
|
|
|
// LookupInvoice looks up an invoice by its payment hash (R-Hash), if found
|
|
|
|
// then we're able to pull the funds pending within an HTLC. We'll also return
|
|
|
|
// what the expected min final CLTV delta is, pre-parsed from the payment
|
|
|
|
// request. This may be used by callers to determine if an HTLC is well formed
|
|
|
|
// according to the cltv delta.
|
|
|
|
//
|
2016-09-19 21:55:02 +03:00
|
|
|
// TODO(roasbeef): ignore if settled?
|
2018-06-30 02:03:46 +03:00
|
|
|
func (i *invoiceRegistry) LookupInvoice(rHash chainhash.Hash) (channeldb.Invoice, uint32, error) {
|
2016-09-19 21:55:02 +03:00
|
|
|
// First check the in-memory debug invoice index to see if this is an
|
|
|
|
// existing invoice added for debugging.
|
2016-07-13 03:14:07 +03:00
|
|
|
i.RLock()
|
2018-04-25 07:07:30 +03:00
|
|
|
debugInv, ok := i.debugInvoices[rHash]
|
2016-07-13 03:14:07 +03:00
|
|
|
i.RUnlock()
|
|
|
|
|
2016-09-19 21:55:02 +03:00
|
|
|
// If found, then simply return the invoice directly.
|
|
|
|
if ok {
|
2018-04-25 07:07:30 +03:00
|
|
|
return *debugInv, 0, nil
|
2016-09-19 21:55:02 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
// Otherwise, we'll check the database to see if there's an existing
|
|
|
|
// matching invoice.
|
2017-11-12 03:09:14 +03:00
|
|
|
invoice, err := i.cdb.LookupInvoice(rHash)
|
|
|
|
if err != nil {
|
2018-06-30 02:03:46 +03:00
|
|
|
return channeldb.Invoice{}, 0, err
|
2017-11-12 03:09:14 +03:00
|
|
|
}
|
|
|
|
|
2018-06-30 02:03:46 +03:00
|
|
|
payReq, err := zpay32.Decode(
|
|
|
|
string(invoice.PaymentRequest), activeNetParams.Params,
|
|
|
|
)
|
|
|
|
if err != nil {
|
|
|
|
return channeldb.Invoice{}, 0, err
|
|
|
|
}
|
|
|
|
|
|
|
|
return *invoice, uint32(payReq.MinFinalCLTVExpiry()), nil
|
2016-07-13 03:14:07 +03:00
|
|
|
}
|
|
|
|
|
2016-09-19 21:55:02 +03:00
|
|
|
// SettleInvoice attempts to mark an invoice as settled. If the invoice is a
|
2017-01-13 08:01:50 +03:00
|
|
|
// debug invoice, then this method is a noop as debug invoices are never fully
|
2016-09-19 21:55:02 +03:00
|
|
|
// settled.
|
2018-04-25 07:07:30 +03:00
|
|
|
func (i *invoiceRegistry) SettleInvoice(rHash chainhash.Hash,
|
|
|
|
amtPaid lnwire.MilliSatoshi) error {
|
|
|
|
|
2016-09-22 05:41:26 +03:00
|
|
|
ltndLog.Debugf("Settling invoice %x", rHash[:])
|
2016-09-21 03:00:11 +03:00
|
|
|
|
2016-09-19 21:55:02 +03:00
|
|
|
// First check the in-memory debug invoice index to see if this is an
|
|
|
|
// existing invoice added for debugging.
|
|
|
|
i.RLock()
|
|
|
|
if _, ok := i.debugInvoices[rHash]; ok {
|
|
|
|
// Debug invoices are never fully settled, so we simply return
|
|
|
|
// immediately in this case.
|
|
|
|
i.RUnlock()
|
2016-07-13 03:14:07 +03:00
|
|
|
|
2016-09-19 21:55:02 +03:00
|
|
|
return nil
|
2016-07-13 03:14:07 +03:00
|
|
|
}
|
2016-09-19 21:55:02 +03:00
|
|
|
i.RUnlock()
|
|
|
|
|
|
|
|
// If this isn't a debug invoice, then we'll attempt to settle an
|
|
|
|
// invoice matching this rHash on disk (if one exists).
|
2018-04-25 07:07:30 +03:00
|
|
|
if err := i.cdb.SettleInvoice(rHash, amtPaid); err != nil {
|
2016-10-15 05:47:10 +03:00
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
// Launch a new goroutine to notify any/all registered invoice
|
|
|
|
// notification clients.
|
|
|
|
go func() {
|
|
|
|
invoice, err := i.cdb.LookupInvoice(rHash)
|
|
|
|
if err != nil {
|
|
|
|
ltndLog.Errorf("unable to find invoice: %v", err)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2017-01-15 05:18:08 +03:00
|
|
|
ltndLog.Infof("Payment received: %v", spew.Sdump(invoice))
|
|
|
|
|
2018-04-25 07:07:30 +03:00
|
|
|
i.notifyClients(&invoice, true)
|
2016-10-15 05:47:10 +03:00
|
|
|
}()
|
|
|
|
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2016-12-25 03:51:25 +03:00
|
|
|
// notifyClients notifies all currently registered invoice notification clients
|
2016-10-15 05:47:10 +03:00
|
|
|
// of a newly added/settled invoice.
|
|
|
|
func (i *invoiceRegistry) notifyClients(invoice *channeldb.Invoice, settle bool) {
|
2018-04-25 07:07:30 +03:00
|
|
|
event := &invoiceEvent{
|
|
|
|
isSettle: settle,
|
|
|
|
invoice: invoice,
|
|
|
|
}
|
2016-10-15 05:47:10 +03:00
|
|
|
|
2018-04-25 07:07:30 +03:00
|
|
|
select {
|
|
|
|
case i.invoiceEvents <- event:
|
|
|
|
case <-i.quit:
|
2016-10-15 05:47:10 +03:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// invoiceSubscription represents an intent to receive updates for newly added
|
|
|
|
// or settled invoices. For each newly added invoice, a copy of the invoice
|
|
|
|
// will be sent over the NewInvoices channel. Similarly, for each newly settled
|
|
|
|
// invoice, a copy of the invoice will be sent over the SettledInvoices
|
|
|
|
// channel.
|
|
|
|
type invoiceSubscription struct {
|
2018-04-25 07:07:30 +03:00
|
|
|
cancelled uint32 // To be used atomically.
|
|
|
|
|
|
|
|
// NewInvoices is a channel that we'll use to send all newly created
|
|
|
|
// invoices with an invoice index greater than the specified
|
|
|
|
// StartingInvoiceIndex field.
|
|
|
|
NewInvoices chan *channeldb.Invoice
|
|
|
|
|
|
|
|
// SettledInvoices is a channel that we'll use to send all setted
|
|
|
|
// invoices with an invoices index greater than the specified
|
|
|
|
// StartingInvoiceIndex field.
|
2016-10-15 05:47:10 +03:00
|
|
|
SettledInvoices chan *channeldb.Invoice
|
|
|
|
|
2018-04-25 07:07:30 +03:00
|
|
|
// addIndex is the highest add index the caller knows of. We'll use
|
|
|
|
// this information to send out an event backlog to the notifications
|
|
|
|
// subscriber. Any new add events with an index greater than this will
|
|
|
|
// be dispatched before any new notifications are sent out.
|
|
|
|
addIndex uint64
|
|
|
|
|
|
|
|
// settleIndex is the highest settle index the caller knows of. We'll
|
|
|
|
// use this information to send out an event backlog to the
|
|
|
|
// notifications subscriber. Any new settle events with an index
|
|
|
|
// greater than this will be dispatched before any new notifications
|
|
|
|
// are sent out.
|
|
|
|
settleIndex uint64
|
|
|
|
|
|
|
|
ntfnQueue *chainntnfs.ConcurrentQueue
|
|
|
|
|
|
|
|
id uint32
|
|
|
|
|
2016-10-15 05:47:10 +03:00
|
|
|
inv *invoiceRegistry
|
2018-04-25 07:07:30 +03:00
|
|
|
|
|
|
|
cancelChan chan struct{}
|
|
|
|
|
|
|
|
wg sync.WaitGroup
|
2016-10-15 05:47:10 +03:00
|
|
|
}
|
|
|
|
|
2017-01-13 08:01:50 +03:00
|
|
|
// Cancel unregisters the invoiceSubscription, freeing any previously allocated
|
2016-10-15 05:47:10 +03:00
|
|
|
// resources.
|
|
|
|
func (i *invoiceSubscription) Cancel() {
|
2018-04-25 07:07:30 +03:00
|
|
|
if !atomic.CompareAndSwapUint32(&i.cancelled, 0, 1) {
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
select {
|
|
|
|
case i.inv.subscriptionCancels <- i.id:
|
|
|
|
case <-i.inv.quit:
|
|
|
|
}
|
|
|
|
|
|
|
|
i.ntfnQueue.Stop()
|
|
|
|
close(i.cancelChan)
|
|
|
|
|
|
|
|
i.wg.Wait()
|
2016-10-15 05:47:10 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
// SubscribeNotifications returns an invoiceSubscription which allows the
|
|
|
|
// caller to receive async notifications when any invoices are settled or
|
2018-04-25 07:07:30 +03:00
|
|
|
// added. The invoiceIndex parameter is a streaming "checkpoint". We'll start
|
|
|
|
// by first sending out all new events with an invoice index _greater_ than
|
|
|
|
// this value. Afterwards, we'll send out real-time notifications.
|
|
|
|
func (i *invoiceRegistry) SubscribeNotifications(addIndex, settleIndex uint64) *invoiceSubscription {
|
2016-10-15 05:47:10 +03:00
|
|
|
client := &invoiceSubscription{
|
|
|
|
NewInvoices: make(chan *channeldb.Invoice),
|
|
|
|
SettledInvoices: make(chan *channeldb.Invoice),
|
2018-04-25 07:07:30 +03:00
|
|
|
addIndex: addIndex,
|
|
|
|
settleIndex: settleIndex,
|
2016-11-22 06:33:29 +03:00
|
|
|
inv: i,
|
2018-04-25 07:07:30 +03:00
|
|
|
ntfnQueue: chainntnfs.NewConcurrentQueue(20),
|
|
|
|
cancelChan: make(chan struct{}),
|
2016-10-15 05:47:10 +03:00
|
|
|
}
|
2018-04-25 07:07:30 +03:00
|
|
|
client.ntfnQueue.Start()
|
2016-10-15 05:47:10 +03:00
|
|
|
|
|
|
|
i.clientMtx.Lock()
|
|
|
|
client.id = i.nextClientID
|
|
|
|
i.nextClientID++
|
|
|
|
i.clientMtx.Unlock()
|
|
|
|
|
2018-04-25 07:07:30 +03:00
|
|
|
// Before we register this new invoice subscription, we'll launch a new
|
|
|
|
// goroutine that will proxy all notifications appended to the end of
|
|
|
|
// the concurrent queue to the two client-side channels the caller will
|
|
|
|
// feed off of.
|
|
|
|
i.wg.Add(1)
|
|
|
|
go func() {
|
|
|
|
defer i.wg.Done()
|
|
|
|
|
|
|
|
for {
|
|
|
|
select {
|
|
|
|
// A new invoice event has been sent by the
|
|
|
|
// invoiceRegistry! We'll figure out if this is an add
|
|
|
|
// event or a settle event, then dispatch the event to
|
|
|
|
// the client.
|
|
|
|
case ntfn := <-client.ntfnQueue.ChanOut():
|
|
|
|
invoiceEvent := ntfn.(*invoiceEvent)
|
|
|
|
|
|
|
|
targetChan := client.NewInvoices
|
|
|
|
if invoiceEvent.isSettle {
|
|
|
|
targetChan = client.SettledInvoices
|
|
|
|
}
|
|
|
|
|
|
|
|
select {
|
|
|
|
case targetChan <- invoiceEvent.invoice:
|
|
|
|
|
|
|
|
case <-client.cancelChan:
|
|
|
|
return
|
|
|
|
|
|
|
|
case <-i.quit:
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
case <-client.cancelChan:
|
|
|
|
return
|
|
|
|
|
|
|
|
case <-i.quit:
|
|
|
|
return
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}()
|
|
|
|
|
|
|
|
select {
|
|
|
|
case i.newSubscriptions <- client:
|
|
|
|
case <-i.quit:
|
|
|
|
}
|
|
|
|
|
2016-10-15 05:47:10 +03:00
|
|
|
return client
|
2016-07-13 03:14:07 +03:00
|
|
|
}
|