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 = "NTF"
+
+// 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,214 @@
+package notifications
+
+import (
+	"context"
+	"crypto/rand"
+	"encoding/hex"
+	"time"
+
+	"github.com/lightninglabs/loop/swapserverrpc"
+)
+
+type NotificationType int
+
+const (
+	NotificationTypeUnknown NotificationType = iota
+	NotificationTypeReservation
+)
+
+// 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 swapserverrpc.SwapServerClient
+
+	// 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
+
+	runCtx context.Context
+
+	subscribers map[NotificationType]map[string]func(*swapserverrpc.SubscribeNotificationsResponse) //nolint:lll
+}
+
+// NewManager creates a new notification manager.
+func NewManager(cfg *Config) *Manager {
+	return &Manager{
+		cfg:         cfg,
+		subscribers: make(map[NotificationType]map[string]func(*swapserverrpc.SubscribeNotificationsResponse)), //nolint:lll
+	}
+}
+
+// SubscribeReservations subscribes to the reservation notifications.
+func (n *Manager) SubscribeReservations(
+	callback func(*swapserverrpc.ServerReservationNotification)) string {
+
+	fn := func(notif *swapserverrpc.SubscribeNotificationsResponse) {
+		callback(notif.GetReservationNotification())
+	}
+	return n.subscribe(NotificationTypeReservation, fn)
+}
+
+// subscribe subscribes to a notification type and returns a unique ID that can
+// be used to unsubscribe.
+func (n *Manager) subscribe(notificationType NotificationType,
+	callback func(*swapserverrpc.SubscribeNotificationsResponse)) string {
+	if n.subscribers[notificationType] == nil {
+		n.subscribers[notificationType] = make(map[string]func(*swapserverrpc.SubscribeNotificationsResponse)) //nolint:lll
+	}
+
+	// Generate a unique ID for this subscription
+	id := generateUniqueID()
+
+	n.subscribers[notificationType][id] = callback
+
+	return id
+}
+
+// Unsubscribe removes a subscription based on its ID.
+func (n *Manager) Unsubscribe(id string) {
+	for _, subscribers := range n.subscribers {
+		delete(subscribers, id)
+	}
+}
+
+// Run starts the notification manager.
+func (n *Manager) Run(ctx context.Context, readyChan chan struct{},
+) error {
+
+	n.runCtx = ctx
+
+	err := n.RegisterNotifications()
+	if err != nil {
+		return err
+	}
+
+	// Notify the caller that we're ready, this means we have successfully
+	// fetched a l402 token and have subscribed to the notifications at least
+	// once.
+	close(readyChan)
+
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		}
+	}
+}
+
+// RegisterNotifications registers a new server notification stream.
+func (n *Manager) RegisterNotifications() 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 !n.hasL402 {
+		n.fetchL402(n.runCtx)
+	}
+
+	ctx, cancel := context.WithCancel(n.runCtx)
+
+	// We'll now subscribe to the swap server notifications.
+	notifStream, err := n.cfg.Client.SubscribeNotifications(
+		ctx, &swapserverrpc.SubscribeNotificationsRequest{},
+	)
+	if err != nil {
+		cancel()
+		return err
+	}
+
+	log.Debugf("Successfully subscribed to server notifications")
+
+	// We'll now start a goroutine that will handle all the incoming
+	// notifications.
+	go func() {
+		for {
+			notification, err := notifStream.Recv()
+			if err == nil && notification != nil {
+				log.Debugf("Received notification: %v", notification)
+				n.handleNotification(notification)
+				continue
+			}
+			log.Errorf("Error receiving "+
+				"notification: %v", err)
+
+			cancel()
+
+			// If we encounter an error, we'll
+			// try to reconnect.
+			for {
+				select {
+				case <-n.runCtx.Done():
+					return
+
+				case <-time.After(time.Second * 10):
+					log.Debugf("Reconnecting to server notifications")
+					err = n.RegisterNotifications()
+					if err != nil {
+						log.Errorf("Error reconnecting: %v", err)
+						continue
+					}
+
+					// If we were able to reconnect, we'll
+					// return.
+					return
+				}
+			}
+		}
+	}()
+
+	return 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
+		}
+	}
+}
+
+// handleNotification handles an incoming notification from the server,
+// forwarding it to the appropriate subscribers.
+func (n *Manager) handleNotification(notification *swapserverrpc.
+	SubscribeNotificationsResponse) {
+
+	switch notification.Notification.(type) {
+	case *swapserverrpc.SubscribeNotificationsResponse_ReservationNotification:
+		for _, fn := range n.subscribers[NotificationTypeReservation] {
+			fn(notification)
+		}
+	}
+}
+
+// generateUniqueID generates a unique ID for a subscription.
+func generateUniqueID() string {
+	var b [16]byte
+	_, err := rand.Read(b[:])
+	if err != nil {
+		panic(err)
+	}
+	return hex.EncodeToString(b[:])
+}