diff --git a/instantout/reservation/actions_test.go b/instantout/reservation/actions_test.go index 2f989d92..d89e526d 100644 --- a/instantout/reservation/actions_test.go +++ b/instantout/reservation/actions_test.go @@ -47,16 +47,6 @@ type mockReservationClient struct { mock.Mock } -func (m *mockReservationClient) OpenReservation(ctx context.Context, - in *swapserverrpc.ServerOpenReservationRequest, - opts ...grpc.CallOption) (*swapserverrpc.ServerOpenReservationResponse, - error) { - - args := m.Called(ctx, in, opts) - return args.Get(0).(*swapserverrpc.ServerOpenReservationResponse), - args.Error(1) -} - func (m *mockReservationClient) ReservationNotificationStream( ctx context.Context, in *swapserverrpc.ReservationNotificationRequest, opts ...grpc.CallOption, @@ -68,6 +58,16 @@ func (m *mockReservationClient) ReservationNotificationStream( args.Error(1) } +func (m *mockReservationClient) OpenReservation(ctx context.Context, + in *swapserverrpc.ServerOpenReservationRequest, + opts ...grpc.CallOption) (*swapserverrpc.ServerOpenReservationResponse, + error) { + + args := m.Called(ctx, in, opts) + return args.Get(0).(*swapserverrpc.ServerOpenReservationResponse), + args.Error(1) +} + func (m *mockReservationClient) FetchL402(ctx context.Context, in *swapserverrpc.FetchL402Request, opts ...grpc.CallOption) (*swapserverrpc.FetchL402Response, error) { diff --git a/instantout/reservation/fsm.go b/instantout/reservation/fsm.go index 188fe1fa..86afbbaa 100644 --- a/instantout/reservation/fsm.go +++ b/instantout/reservation/fsm.go @@ -28,8 +28,9 @@ type Config struct { // swap server. ReservationClient swapserverrpc.ReservationServiceClient - // FetchL402 is the function used to fetch the l402 token. - FetchL402 func(context.Context) error + // NotificationManager is the manager that handles the notification + // subscriptions. + NotificationManager NotificationManager } // FSM is the state machine that manages the reservation lifecycle. diff --git a/instantout/reservation/interfaces.go b/instantout/reservation/interfaces.go index c999d1b9..04bf830d 100644 --- a/instantout/reservation/interfaces.go +++ b/instantout/reservation/interfaces.go @@ -3,6 +3,8 @@ package reservation import ( "context" "fmt" + + "github.com/lightninglabs/loop/swapserverrpc" ) var ( @@ -31,3 +33,10 @@ type Store interface { // made. ListReservations(ctx context.Context) ([]*Reservation, error) } + +// NotificationManager handles subscribing to incoming reservation +// subscriptions. +type NotificationManager interface { + SubscribeReservations(context.Context, + ) <-chan *swapserverrpc.ServerReservationNotification +} diff --git a/instantout/reservation/manager.go b/instantout/reservation/manager.go index 35f80ecb..faffbc59 100644 --- a/instantout/reservation/manager.go +++ b/instantout/reservation/manager.go @@ -22,9 +22,6 @@ type Manager struct { // activeReservations contains all the active reservationsFSMs. activeReservations map[ID]*FSM - // hasL402 is true if the client has a valid L402. - hasL402 bool - runCtx context.Context sync.Mutex @@ -59,14 +56,7 @@ func (m *Manager) Run(ctx context.Context, height int32) error { return err } - reservationResChan := make( - chan *reservationrpc.ServerReservationNotification, - ) - - err = m.RegisterReservationNotifications(reservationResChan) - if err != nil { - return err - } + ntfnChan := m.cfg.NotificationManager.SubscribeReservations(ctx) for { select { @@ -74,7 +64,7 @@ func (m *Manager) Run(ctx context.Context, height int32) error { log.Debugf("Received block %v", height) currentHeight = height - case reservationRes := <-reservationResChan: + case reservationRes := <-ntfnChan: log.Debugf("Received reservation %x", reservationRes.ReservationId) _, err := m.newReservation( @@ -157,101 +147,6 @@ func (m *Manager) newReservation(ctx context.Context, currentHeight uint32, return reservationFSM, nil } -// fetchL402 fetches the L402 from the server. This method will keep on -// retrying until it gets a valid response. -func (m *Manager) fetchL402(ctx context.Context) { - // Add a 0 timer so that we initially fetch the L402 immediately. - timer := time.NewTimer(0) - for { - select { - case <-ctx.Done(): - return - - case <-timer.C: - err := m.cfg.FetchL402(ctx) - if err != nil { - log.Warnf("Error fetching L402: %v", err) - timer.Reset(time.Second * 10) - continue - } - m.hasL402 = true - return - } - } -} - -// RegisterReservationNotifications registers a new reservation notification -// stream. -func (m *Manager) RegisterReservationNotifications( - reservationChan chan *reservationrpc.ServerReservationNotification) error { - - // In order to create a valid l402 we first are going to call - // the FetchL402 method. As a client might not have outbound capacity - // yet, we'll retry until we get a valid response. - if !m.hasL402 { - m.fetchL402(m.runCtx) - } - - ctx, cancel := context.WithCancel(m.runCtx) - - // We'll now subscribe to the reservation notifications. - reservationStream, err := m.cfg.ReservationClient. - ReservationNotificationStream( - ctx, &reservationrpc.ReservationNotificationRequest{}, - ) - if err != nil { - cancel() - return err - } - - log.Debugf("Successfully subscribed to reservation notifications") - - // We'll now start a goroutine that will forward all the reservation - // notifications to the reservationChan. - go func() { - for { - reservationRes, err := reservationStream.Recv() - if err == nil && reservationRes != nil { - log.Debugf("Received reservation %x", - reservationRes.ReservationId) - reservationChan <- reservationRes - continue - } - log.Errorf("Error receiving "+ - "reservation: %v", err) - - cancel() - - // If we encounter an error, we'll - // try to reconnect. - for { - select { - case <-m.runCtx.Done(): - return - - case <-time.After(time.Second * 10): - log.Debugf("Reconnecting to " + - "reservation notifications") - err = m.RegisterReservationNotifications( - reservationChan, - ) - if err != nil { - log.Errorf("Error "+ - "reconnecting: %v", err) - continue - } - - // If we were able to reconnect, we'll - // return. - return - } - } - } - }() - - return nil -} - // RecoverReservations tries to recover all reservations that are still active // from the database. func (m *Manager) RecoverReservations(ctx context.Context) error { diff --git a/instantout/reservation/manager_test.go b/instantout/reservation/manager_test.go index f04d5444..1dbb5a34 100644 --- a/instantout/reservation/manager_test.go +++ b/instantout/reservation/manager_test.go @@ -13,7 +13,6 @@ import ( "github.com/lightningnetwork/lnd/chainntnfs" "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" - "google.golang.org/grpc" ) var ( @@ -118,27 +117,22 @@ func newManagerTestContext(t *testing.T) *ManagerTestContext { sendChan := make(chan *swapserverrpc.ServerReservationNotification) - mockReservationClient.On( - "ReservationNotificationStream", mock.Anything, mock.Anything, - mock.Anything, - ).Return( - &dummyReservationNotificationServer{ - SendChan: sendChan, - }, nil, - ) - mockReservationClient.On( "OpenReservation", mock.Anything, mock.Anything, mock.Anything, ).Return( &swapserverrpc.ServerOpenReservationResponse{}, nil, ) + mockNtfnManager := &mockNtfnManager{ + sendChan: sendChan, + } + cfg := &Config{ - Store: store, - Wallet: mockLnd.WalletKit, - ChainNotifier: mockLnd.ChainNotifier, - FetchL402: func(context.Context) error { return nil }, - ReservationClient: mockReservationClient, + Store: store, + Wallet: mockLnd.WalletKit, + ChainNotifier: mockLnd.ChainNotifier, + ReservationClient: mockReservationClient, + NotificationManager: mockNtfnManager, } manager := NewManager(cfg) @@ -152,17 +146,15 @@ func newManagerTestContext(t *testing.T) *ManagerTestContext { } } -type dummyReservationNotificationServer struct { - grpc.ClientStream - - // SendChan is the channel that is used to send notifications. - SendChan chan *swapserverrpc.ServerReservationNotification +type mockNtfnManager struct { + sendChan chan *swapserverrpc.ServerReservationNotification } -func (d *dummyReservationNotificationServer) Recv() ( - *swapserverrpc.ServerReservationNotification, error) { +func (m *mockNtfnManager) SubscribeReservations( + ctx context.Context, +) <-chan *swapserverrpc.ServerReservationNotification { - return <-d.SendChan, nil + return m.sendChan } func mustDecodeID(id string) ID {