notifications: add notification manager

This commit adds a generic notification manager
that can be used to subscribe to different types
of notifications.

Author: sputn1ck

notifications/log.go

@@ -0,0 +1,26 @@
+package notifications
+
+import (
+	"github.com/btcsuite/btclog"
+	"github.com/lightningnetwork/lnd/build"
+)
+
+// Subsystem defines the sub system name of this package.
+const Subsystem = "NTFNS"
+
+// log is a logger that is initialized with no output filters.  This
+// means the package will not perform any logging by default until the caller
+// requests it.
+var log btclog.Logger
+
+// The default amount of logging is none.
+func init() {
+	UseLogger(build.NewSubLogger(Subsystem, nil))
+}
+
+// UseLogger uses a specified Logger to output package logging info.
+// This should be used in preference to SetLogWriter if the caller is also
+// using btclog.
+func UseLogger(logger btclog.Logger) {
+	log = logger
+}

notifications/manager.go

@@ -0,0 +1,222 @@
+package notifications
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"github.com/lightninglabs/loop/swapserverrpc"
+	"google.golang.org/grpc"
+)
+
+type NotificationType int
+
+const (
+	NotificationTypeUnknown NotificationType = iota
+	NotificationTypeReservation
+)
+
+type Client interface {
+	SubscribeNotifications(ctx context.Context,
+		in *swapserverrpc.SubscribeNotificationsRequest,
+		opts ...grpc.CallOption) (
+		swapserverrpc.SwapServer_SubscribeNotificationsClient, error)
+}
+
+// Config contains all the services that the notification manager needs to
+// operate.
+type Config struct {
+	// Client is the client used to communicate with the swap server.
+	Client Client
+
+	// FetchL402 is the function used to fetch the l402 token.
+	FetchL402 func(context.Context) error
+}
+
+// Manager is a manager for notifications that the swap server sends to the
+// client.
+type Manager struct {
+	cfg *Config
+
+	hasL402 bool
+
+	subscribers map[NotificationType][]subscriber
+	sync.Mutex
+}
+
+// NewManager creates a new notification manager.
+func NewManager(cfg *Config) *Manager {
+	return &Manager{
+		cfg:         cfg,
+		subscribers: make(map[NotificationType][]subscriber),
+	}
+}
+
+type subscriber struct {
+	subCtx   context.Context
+	recvChan interface{}
+}
+
+// SubscribeReservations subscribes to the reservation notifications.
+func (m *Manager) SubscribeReservations(ctx context.Context,
+) <-chan *swapserverrpc.ServerReservationNotification {
+
+	// We'll create a channel that we'll use to send the notifications to the
+	// caller.
+	notifChan := make(
+		chan *swapserverrpc.ServerReservationNotification, //nolint:lll
+	)
+	sub := subscriber{
+		subCtx:   ctx,
+		recvChan: notifChan,
+	}
+
+	m.Lock()
+	m.subscribers[NotificationTypeReservation] = append(
+		m.subscribers[NotificationTypeReservation],
+		sub,
+	)
+	m.Unlock()
+
+	return notifChan
+}
+
+// Run starts the notification manager. It will keep on running until the
+// context is canceled. It will subscribe to notifications and forward them to
+// the subscribers. On a first successful connection to the server, it will
+// close the readyChan to signal that the manager is ready.
+func (m *Manager) Run(ctx context.Context, readyChan chan<- struct{}) 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(ctx)
+	}
+
+	closeChanFunc := sync.OnceFunc(func() {
+		close(readyChan)
+	})
+
+	// Initially we want to immediately try to connect to the server.
+	waitTime := time.Duration(0)
+
+	// Start the notification runloop.
+	for {
+		timer := time.NewTimer(waitTime)
+		// Increase the wait time for the next iteration.
+		waitTime += time.Second * 10
+
+		// Return if the context has been canceled.
+		select {
+		case <-ctx.Done():
+			return nil
+
+		case <-timer.C:
+		}
+
+		connectedFunc := func() {
+			closeChanFunc()
+			// Reset the wait time to 10.
+			waitTime = time.Second * 10
+		}
+
+		err := m.subscribeNotifications(ctx, connectedFunc)
+		if err != nil {
+			log.Errorf("Error subscribing to notifications: %v", err)
+		}
+	}
+}
+
+// subscribeNotifications subscribes to the notifications from the server.
+func (m *Manager) subscribeNotifications(ctx context.Context,
+	connectedFunc func()) error {
+
+	callCtx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	notifStream, err := m.cfg.Client.SubscribeNotifications(
+		callCtx, &swapserverrpc.SubscribeNotificationsRequest{},
+	)
+	if err != nil {
+		return err
+	}
+
+	// Signal that we're connected to the server.
+	connectedFunc()
+	log.Debugf("Successfully subscribed to server notifications")
+
+	for {
+		notification, err := notifStream.Recv()
+		if err == nil && notification != nil {
+			log.Debugf("Received notification: %v", notification)
+			m.handleNotification(notification)
+			continue
+		}
+
+		log.Errorf("Error receiving notification: %v", err)
+
+		return err
+	}
+}
+
+// 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.
+	waitTime := time.Duration(0)
+	timer := time.NewTimer(waitTime)
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-timer.C:
+			err := m.cfg.FetchL402(ctx)
+			if err != nil {
+				log.Warnf("Error fetching L402: %v", err)
+				waitTime += time.Second * 10
+				timer.Reset(waitTime)
+				continue
+			}
+			m.hasL402 = true
+
+			return
+		}
+	}
+}
+
+// handleNotification handles an incoming notification from the server,
+// forwarding it to the appropriate subscribers.
+func (m *Manager) handleNotification(notification *swapserverrpc.
+	SubscribeNotificationsResponse) {
+
+	switch notification.Notification.(type) {
+	case *swapserverrpc.SubscribeNotificationsResponse_ReservationNotification:
+		// We'll forward the reservation notification to all subscribers.
+		// Cleaning up any subscribers that have been canceled.
+		newSubs := make(
+			[]subscriber, 0, len(m.subscribers[NotificationTypeReservation]),
+		)
+		reservationNtfn := notification.GetReservationNotification()
+		for _, sub := range m.subscribers[NotificationTypeReservation] {
+			recvChan := sub.recvChan.(chan *swapserverrpc.
+				ServerReservationNotification)
+
+			select {
+			case <-sub.subCtx.Done():
+				close(recvChan)
+				continue
+
+			case recvChan <- reservationNtfn:
+				newSubs = append(newSubs, sub)
+			}
+		}
+
+		m.Lock()
+		m.subscribers[NotificationTypeReservation] = newSubs
+		m.Unlock()
+
+	default:
+		log.Warnf("Received unknown notification type: %v",
+			notification)
+	}
+}