peer: more interfaces, refactor unit tests

htlcswitch/link.go

@@ -134,6 +134,9 @@ type ChannelLinkConfig struct {
 	// TODO(conner): remove after refactoring htlcswitch testing framework.
 	Switch *Switch
 
+	// BestHeight returns the best known height.
+	BestHeight func() uint32
+
 	// ForwardPackets attempts to forward the batch of htlcs through the
 	// switch. The function returns and error in case it fails to send one or
 	// more packets. The link's quit signal should be provided to allow
@@ -2674,7 +2677,7 @@ func (l *channelLink) processRemoteAdds(fwdPkg *channeldb.FwdPkg,
 			continue
 		}
 
-		heightNow := l.cfg.Switch.BestHeight()
+		heightNow := l.cfg.BestHeight()
 
 		pld, err := chanIterator.HopPayload()
 		if err != nil {

htlcswitch/link_test.go

@@ -1951,6 +1951,7 @@ func newSingleLinkTestHarness(chanAmt, chanReserve btcutil.Amount) (
 		FwrdingPolicy:      globalPolicy,
 		Peer:               alicePeer,
 		Switch:             aliceSwitch,
+		BestHeight:         aliceSwitch.BestHeight,
 		Circuits:           aliceSwitch.CircuitModifier(),
 		ForwardPackets:     aliceSwitch.ForwardPackets,
 		DecodeHopIterators: decoder.DecodeHopIterators,
@@ -4454,6 +4455,7 @@ func (h *persistentLinkHarness) restartLink(
 		FwrdingPolicy:      globalPolicy,
 		Peer:               alicePeer,
 		Switch:             aliceSwitch,
+		BestHeight:         aliceSwitch.BestHeight,
 		Circuits:           aliceSwitch.CircuitModifier(),
 		ForwardPackets:     aliceSwitch.ForwardPackets,
 		DecodeHopIterators: decoder.DecodeHopIterators,

htlcswitch/test_utils.go

@@ -1119,6 +1119,7 @@ func (h *hopNetwork) createChannelLink(server, peer *mockServer,
 	link := NewChannelLink(
 		ChannelLinkConfig{
 			Switch:             server.htlcSwitch,
+			BestHeight:         server.htlcSwitch.BestHeight,
 			FwrdingPolicy:      h.globalPolicy,
 			Peer:               peer,
 			Circuits:           server.htlcSwitch.CircuitModifier(),

peer/brontide.go

@@ -121,6 +121,56 @@ type TimestampedError struct {
 	Timestamp time.Time
 }
 
+// ChannelSwitch is an implementation of the MessageSwitch interface that wraps
+// htlcswitch.Switch.
+type ChannelSwitch struct {
+	*htlcswitch.Switch
+}
+
+// NewChannelSwitch initializes a ChannelSwitch given a raw *htlcswitch.Switch
+// pointer.
+func NewChannelSwitch(innerSwitch *htlcswitch.Switch) *ChannelSwitch {
+	return &ChannelSwitch{innerSwitch}
+}
+
+// BestHeight returns innerSwitch's BestHeight.
+func (s *ChannelSwitch) BestHeight() uint32 {
+	return s.Switch.BestHeight()
+}
+
+// CircuitModifier returns the innerSwitch's CircuitModifier.
+func (s *ChannelSwitch) CircuitModifier() htlcswitch.CircuitModifier {
+	return s.Switch.CircuitModifier()
+}
+
+// GetLink retrieves a MessageLink given a ChannelID from the inner Switch.
+func (s *ChannelSwitch) GetLink(cid lnwire.ChannelID) (MessageLink, error) {
+	return s.Switch.GetLink(cid)
+}
+
+// InitLink initializes a ChannelLink in the Switch.
+func (s *ChannelSwitch) InitLink(linkCfg htlcswitch.ChannelLinkConfig,
+	lnChan *lnwallet.LightningChannel) error {
+
+	link := htlcswitch.NewChannelLink(linkCfg, lnChan)
+
+	// Before adding our new link, purge the Switch of any pending or live
+	// links going by the same channel id. If one is found, we'll shut it
+	// down to ensure that the mailboxes are only ever under the control of
+	// one link.
+	s.RemoveLink(link.ChanID())
+
+	// With the ChannelLink created, we'll now notify the Switch so this
+	// channel can be used to dispatch local payments and also passively
+	// forward payments.
+	return s.AddLink(link)
+}
+
+// RemoveLink removes a ChannelLink from the Switch.
+func (s *ChannelSwitch) RemoveLink(cid lnwire.ChannelID) {
+	s.Switch.RemoveLink(cid)
+}
+
 // Config defines configuration fields that are necessary for a peer object
 // to function.
 type Config struct {
@@ -175,9 +225,9 @@ type Config struct {
 	// ReadPool is the task pool that manages reuse of read buffers.
 	ReadPool *pool.Read
 
-	// Switch is a pointer to the htlcswitch. It is used to setup, get, and
-	// tear-down ChannelLinks.
-	Switch *htlcswitch.Switch
+	// Switch is an implementation of MessageSwitch. It is used to setup,
+	// get, and tear-down MessageLinks.
+	Switch MessageSwitch
 
 	// InterceptSwitch is a pointer to the InterceptableSwitch, a wrapper around
 	// the regular Switch. We only export it here to pass ForwardPackets to the
@@ -813,7 +863,7 @@ func (p *Brontide) addLink(chanPoint *wire.OutPoint,
 		FetchLastChannelUpdate:  p.cfg.FetchLastChanUpdate,
 		HodlMask:                p.cfg.Hodl.Mask(),
 		Registry:                p.cfg.Invoices,
-		Switch:                  p.cfg.Switch,
+		BestHeight:              p.cfg.Switch.BestHeight,
 		Circuits:                p.cfg.Switch.CircuitModifier(),
 		ForwardPackets:          p.cfg.InterceptSwitch.ForwardPackets,
 		FwrdingPolicy:           *forwardingPolicy,
@@ -841,18 +891,7 @@ func (p *Brontide) addLink(chanPoint *wire.OutPoint,
 		HtlcNotifier:            p.cfg.HtlcNotifier,
 	}
 
-	link := htlcswitch.NewChannelLink(linkCfg, lnChan)
-
-	// Before adding our new link, purge the switch of any pending or live
-	// links going by the same channel id. If one is found, we'll shut it
-	// down to ensure that the mailboxes are only ever under the control of
-	// one link.
-	p.cfg.Switch.RemoveLink(link.ChanID())
-
-	// With the channel link created, we'll now notify the htlc switch so
-	// this channel can be used to dispatch local payments and also
-	// passively forward payments.
-	return p.cfg.Switch.AddLink(link)
+	return p.cfg.Switch.InitLink(linkCfg, lnChan)
 }
 
 // maybeSendNodeAnn sends our node announcement to the remote peer if at least
@@ -1139,10 +1178,9 @@ func (ms *msgStream) AddMsg(msg lnwire.Message) {
 }
 
 // waitUntilLinkActive waits until the target link is active and returns a
-// ChannelLink to pass messages to. It accomplishes this by subscribing to
+// MessageLink to pass messages to. It accomplishes this by subscribing to
 // an ActiveLinkEvent which is emitted by the link when it first starts up.
-func waitUntilLinkActive(p *Brontide,
-	cid lnwire.ChannelID) htlcswitch.ChannelLink {
+func waitUntilLinkActive(p *Brontide, cid lnwire.ChannelID) MessageLink {
 
 	// Subscribe to receive channel events.
 	//
@@ -1211,7 +1249,7 @@ func waitUntilLinkActive(p *Brontide,
 // lookups.
 func newChanMsgStream(p *Brontide, cid lnwire.ChannelID) *msgStream {
 
-	var chanLink htlcswitch.ChannelLink
+	var chanLink MessageLink
 
 	apply := func(msg lnwire.Message) {
 		// This check is fine because if the link no longer exists, it will

peer/interfaces.go

@@ -4,6 +4,8 @@ import (
 	"net"
 	"time"
 
+	"github.com/lightningnetwork/lnd/htlcswitch"
+	"github.com/lightningnetwork/lnd/lnwallet"
 	"github.com/lightningnetwork/lnd/lnwire"
 )
 
@@ -54,3 +56,35 @@ type MessageConn interface {
 	// ReadNextBody reads the next body.
 	ReadNextBody([]byte) ([]byte, error)
 }
+
+// MessageLink is an interface that contains some functionality from a
+// htlcswitch.ChannelLink.
+type MessageLink interface {
+	// ChanID returns the ChannelID of the MessageLink.
+	ChanID() lnwire.ChannelID
+
+	// HandleChannelUpdate passes lnwire.Message to the MessageLink.
+	HandleChannelUpdate(lnwire.Message)
+}
+
+// MessageSwitch is an interface that manages setup, retrieval, and shutdown of
+// MessageLink implementations.
+type MessageSwitch interface {
+	// BestHeight returns the best height known to the MessageSwitch.
+	BestHeight() uint32
+
+	// CircuitModifier returns a reference to a CircuitModifier.
+	CircuitModifier() htlcswitch.CircuitModifier
+
+	// GetLink retrieves a MessageLink given a ChannelID.
+	GetLink(lnwire.ChannelID) (MessageLink, error)
+
+	// InitLink creates a link given a ChannelLinkConfig and
+	// LightningChannel.
+	InitLink(htlcswitch.ChannelLinkConfig,
+		*lnwallet.LightningChannel) error
+
+	// RemoveLink removes a MessageLink from the MessageSwitch given a
+	// ChannelID.
+	RemoveLink(lnwire.ChannelID)
+}

peer/test_utils.go

@@ -371,7 +371,7 @@ func createTestPeer(notifier chainntnfs.ChainNotifier,
 		PubKeyBytes: pubKey,
 		ErrorBuffer: errBuffer,
 		ChainIO:     chainIO,
-		Switch:      htlcSwitch,
+		Switch:      NewChannelSwitch(htlcSwitch),
 
 		ChanActiveTimeout: chanActiveTimeout,
 		InterceptSwitch:   htlcswitch.NewInterceptableSwitch(htlcSwitch),

server.go

@@ -3237,7 +3237,7 @@ func (s *server) peerConnected(conn net.Conn, connReq *connmgr.ConnReq,
 		ErrorBuffer:             errBuffer,
 		WritePool:               s.writePool,
 		ReadPool:                s.readPool,
-		Switch:                  s.htlcSwitch,
+		Switch:                  peer.NewChannelSwitch(s.htlcSwitch),
 		InterceptSwitch:         s.interceptableSwitch,
 		ChannelDB:               s.remoteChanDB,
 		ChannelGraph:            s.localChanDB.ChannelGraph(),