Merge pull request #226 from carlaKC/138-consumeswapsubscription

multi: consume and log sever state updates

Author: carlaKC

client.go

@@ -162,7 +162,7 @@ func NewClient(dbDir string, cfg *ClientConfig) (*Client, func(), error) {
 	}
 
 	cleanup := func() {
-		swapServerClient.Close()
+		swapServerClient.stop()
 	}
 
 	return client, cleanup, nil

loopin.go

@@ -5,6 +5,7 @@ import (
 	"crypto/rand"
 	"crypto/sha256"
 	"fmt"
+	"sync"
 	"time"
 
 	"github.com/btcsuite/btcutil"
@@ -61,6 +62,8 @@ type loopInSwap struct {
 	htlcTxHash *chainhash.Hash
 
 	timeoutAddr btcutil.Address
+
+	wg sync.WaitGroup
 }
 
 // loopInInitResult contains information about a just-initiated loop in swap.
@@ -293,9 +296,25 @@ func (s *loopInSwap) sendUpdate(ctx context.Context) error {
 func (s *loopInSwap) execute(mainCtx context.Context,
 	cfg *executeConfig, height int32) error {
 
+	defer s.wg.Wait()
+
 	s.executeConfig = *cfg
 	s.height = height
 
+	// Create context for our state subscription which we will cancel once
+	// swap execution has completed, ensuring that we kill the subscribe
+	// goroutine.
+	subCtx, cancel := context.WithCancel(mainCtx)
+	defer cancel()
+
+	s.wg.Add(1)
+	go func() {
+		defer s.wg.Done()
+		subscribeAndLogUpdates(
+			subCtx, s.hash, s.log, s.server.SubscribeLoopInUpdates,
+		)
+	}()
+
 	// Announce swap by sending out an initial update.
 	err := s.sendUpdate(mainCtx)
 	if err != nil {

loopout.go

@@ -6,6 +6,7 @@ import (
 	"crypto/sha256"
 	"errors"
 	"fmt"
+	"sync"
 	"time"
 
 	"github.com/btcsuite/btcd/chaincfg/chainhash"
@@ -62,6 +63,8 @@ type loopOutSwap struct {
 
 	swapPaymentChan chan lndclient.PaymentResult
 	prePaymentChan  chan lndclient.PaymentResult
+
+	wg sync.WaitGroup
 }
 
 // executeConfig contains extra configuration to execute the swap.
@@ -254,9 +257,25 @@ func (s *loopOutSwap) sendUpdate(ctx context.Context) error {
 func (s *loopOutSwap) execute(mainCtx context.Context,
 	cfg *executeConfig, height int32) error {
 
+	defer s.wg.Wait()
+
 	s.executeConfig = *cfg
 	s.height = height
 
+	// Create context for our state subscription which we will cancel once
+	// swap execution has completed, ensuring that we kill the subscribe
+	// goroutine.
+	subCtx, cancel := context.WithCancel(mainCtx)
+	defer cancel()
+
+	s.wg.Add(1)
+	go func() {
+		defer s.wg.Done()
+		subscribeAndLogUpdates(
+			subCtx, s.hash, s.log, s.server.SubscribeLoopOutUpdates,
+		)
+	}()
+
 	// Execute swap.
 	err := s.executeAndFinalize(mainCtx)