Skip to content

Commit 2e81d93

Browse files
sputn1cksputn1ck
committed
notifications: add notification manager
This commit adds a generic notification manager that can be used to subscribe to different types of notifications.
1 parent aea7578 commit 2e81d93

File tree

3 files changed

+380
-0
lines changed

3 files changed

+380
-0
lines changed

‎notifications/log.go‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,26 @@
1+
package notifications
2+
3+
import (
4+
"github.com/btcsuite/btclog"
5+
"github.com/lightningnetwork/lnd/build"
6+
)
7+
8+
// Subsystem defines the sub system name of this package.
9+
const Subsystem = "NTFNS"
10+
11+
// log is a logger that is initialized with no output filters. This
12+
// means the package will not perform any logging by default until the caller
13+
// requests it.
14+
var log btclog.Logger
15+
16+
// The default amount of logging is none.
17+
func init() {
18+
UseLogger(build.NewSubLogger(Subsystem, nil))
19+
}
20+
21+
// UseLogger uses a specified Logger to output package logging info.
22+
// This should be used in preference to SetLogWriter if the caller is also
23+
// using btclog.
24+
func UseLogger(logger btclog.Logger) {
25+
log = logger
26+
}

‎notifications/manager.go‎

Lines changed: 193 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,193 @@
1+
package notifications
2+
3+
import (
4+
"context"
5+
"sync"
6+
"time"
7+
8+
"github.com/lightninglabs/loop/swapserverrpc"
9+
"google.golang.org/grpc"
10+
)
11+
12+
type NotificationType int
13+
14+
const (
15+
NotificationTypeUnknown NotificationType = iota
16+
NotificationTypeReservation
17+
)
18+
19+
type NotificationsClient interface {
20+
SubscribeNotifications(ctx context.Context,
21+
in *swapserverrpc.SubscribeNotificationsRequest,
22+
opts ...grpc.CallOption) (
23+
swapserverrpc.SwapServer_SubscribeNotificationsClient, error)
24+
}
25+
26+
// Config contains all the services that the notification manager needs to
27+
// operate.
28+
type Config struct {
29+
// Client is the client used to communicate with the swap server.
30+
Client NotificationsClient
31+
32+
// FetchL402 is the function used to fetch the l402 token.
33+
FetchL402 func(context.Context) error
34+
}
35+
36+
// Manager is a manager for notifications that the swap server sends to the
37+
// client.
38+
type Manager struct {
39+
cfg *Config
40+
41+
hasL402 bool
42+
43+
subscribers map[NotificationType][]subscriber
44+
sync.Mutex
45+
}
46+
47+
// NewManager creates a new notification manager.
48+
func NewManager(cfg *Config) *Manager {
49+
return &Manager{
50+
cfg: cfg,
51+
subscribers: make(map[NotificationType][]subscriber),
52+
}
53+
}
54+
55+
type subscriber struct {
56+
subCtx context.Context
57+
recvChan interface{}
58+
}
59+
60+
// SubscribeReservations subscribes to the reservation notifications.
61+
func (m *Manager) SubscribeReservations(ctx context.Context,
62+
) <-chan *swapserverrpc.ServerReservationNotification {
63+
64+
// We'll create a channel that we'll use to send the notifications to the
65+
// caller.
66+
notifChan := make(
67+
chan *swapserverrpc.ServerReservationNotification, //nolint:lll
68+
)
69+
sub := subscriber{
70+
subCtx: ctx,
71+
recvChan: notifChan,
72+
}
73+
74+
m.Lock()
75+
m.subscribers[NotificationTypeReservation] = append(
76+
m.subscribers[NotificationTypeReservation],
77+
sub,
78+
)
79+
m.Unlock()
80+
81+
return notifChan
82+
}
83+
84+
// Run starts the notification manager.
85+
func (n *Manager) Run(ctx context.Context, readyChan chan<- struct{}) error {
86+
// In order to create a valid l402 we first are going to call
87+
// the FetchL402 method. As a client might not have outbound capacity
88+
// yet, we'll retry until we get a valid response.
89+
if !n.hasL402 {
90+
n.fetchL402(ctx)
91+
}
92+
93+
var closedChan bool
94+
95+
// Start the notification runloop.
96+
for {
97+
// Return if the context has been canceled.
98+
select {
99+
case <-ctx.Done():
100+
return nil
101+
default:
102+
}
103+
104+
callCtx, cancel := context.WithCancel(ctx)
105+
notifStream, err := n.cfg.Client.SubscribeNotifications(
106+
callCtx, &swapserverrpc.SubscribeNotificationsRequest{},
107+
)
108+
if err != nil {
109+
cancel()
110+
log.Errorf("Error subscribing to notifications: %v", err)
111+
continue
112+
}
113+
if !closedChan {
114+
close(readyChan)
115+
}
116+
117+
log.Debugf("Successfully subscribed to server notifications")
118+
119+
for {
120+
notification, err := notifStream.Recv()
121+
if err == nil && notification != nil {
122+
log.Debugf("Received notification: %v", notification)
123+
n.handleNotification(notification)
124+
continue
125+
}
126+
log.Errorf("Error receiving "+
127+
"notification: %v", err)
128+
129+
cancel()
130+
// Wait for a bit before reconnecting.
131+
<-time.After(time.Second * 10)
132+
break
133+
}
134+
}
135+
}
136+
137+
// fetchL402 fetches the L402 from the server. This method will keep on
138+
// retrying until it gets a valid response.
139+
func (m *Manager) fetchL402(ctx context.Context) {
140+
// Add a 0 timer so that we initially fetch the L402 immediately.
141+
timer := time.NewTimer(0)
142+
for {
143+
select {
144+
case <-ctx.Done():
145+
return
146+
147+
case <-timer.C:
148+
err := m.cfg.FetchL402(ctx)
149+
if err != nil {
150+
log.Warnf("Error fetching L402: %v", err)
151+
timer.Reset(time.Second * 10)
152+
continue
153+
}
154+
m.hasL402 = true
155+
156+
return
157+
}
158+
}
159+
}
160+
161+
// handleNotification handles an incoming notification from the server,
162+
// forwarding it to the appropriate subscribers.
163+
func (n *Manager) handleNotification(notification *swapserverrpc.
164+
SubscribeNotificationsResponse) {
165+
166+
switch notification.Notification.(type) {
167+
case *swapserverrpc.SubscribeNotificationsResponse_ReservationNotification:
168+
// We'll forward the reservation notification to all subscribers.
169+
// Cleaning up any subscribers that have been canceled.
170+
newSubs := []subscriber{}
171+
reservationNtfn := notification.GetReservationNotification()
172+
for _, sub := range n.subscribers[NotificationTypeReservation] {
173+
recvChan := sub.recvChan.(chan *swapserverrpc.
174+
ServerReservationNotification)
175+
176+
select {
177+
case <-sub.subCtx.Done():
178+
close(recvChan)
179+
continue
180+
case recvChan <- reservationNtfn:
181+
newSubs = append(newSubs, sub)
182+
}
183+
}
184+
185+
n.Lock()
186+
n.subscribers[NotificationTypeReservation] = newSubs
187+
n.Unlock()
188+
189+
default:
190+
log.Warnf("Received unknown notification type: %v",
191+
notification)
192+
}
193+
}

‎notifications/manager_test.go‎

Lines changed: 161 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,161 @@
1+
package notifications
2+
3+
import (
4+
"context"
5+
"io"
6+
"testing"
7+
"time"
8+
9+
"github.com/lightninglabs/loop/swapserverrpc"
10+
"github.com/stretchr/testify/require"
11+
"google.golang.org/grpc"
12+
"google.golang.org/grpc/metadata"
13+
)
14+
15+
var (
16+
testReservationId = []byte{0x01, 0x02}
17+
testReservationId2 = []byte{0x01, 0x02}
18+
)
19+
20+
// mockNotificationsClient implements the NotificationsClient interface for testing.
21+
type mockNotificationsClient struct {
22+
mockStream swapserverrpc.SwapServer_SubscribeNotificationsClient
23+
subscribeErr error
24+
timesCalled int
25+
}
26+
27+
func (m *mockNotificationsClient) SubscribeNotifications(ctx context.Context,
28+
in *swapserverrpc.SubscribeNotificationsRequest,
29+
opts ...grpc.CallOption) (
30+
swapserverrpc.SwapServer_SubscribeNotificationsClient, error) {
31+
32+
m.timesCalled++
33+
if m.subscribeErr != nil {
34+
return nil, m.subscribeErr
35+
}
36+
return m.mockStream, nil
37+
}
38+
39+
// mockSubscribeNotificationsClient simulates the server stream.
40+
type mockSubscribeNotificationsClient struct {
41+
grpc.ClientStream
42+
recvChan chan *swapserverrpc.SubscribeNotificationsResponse
43+
recvErrChan chan error
44+
}
45+
46+
func (m *mockSubscribeNotificationsClient) Recv() (
47+
*swapserverrpc.SubscribeNotificationsResponse, error) {
48+
49+
select {
50+
case err := <-m.recvErrChan:
51+
return nil, err
52+
case notif, ok := <-m.recvChan:
53+
if !ok {
54+
return nil, io.EOF
55+
}
56+
return notif, nil
57+
}
58+
}
59+
60+
func (m *mockSubscribeNotificationsClient) Header() (metadata.MD, error) {
61+
return nil, nil
62+
}
63+
64+
func (m *mockSubscribeNotificationsClient) Trailer() metadata.MD {
65+
return nil
66+
}
67+
68+
func (m *mockSubscribeNotificationsClient) CloseSend() error {
69+
return nil
70+
}
71+
72+
func (m *mockSubscribeNotificationsClient) Context() context.Context {
73+
return context.TODO()
74+
}
75+
76+
func (m *mockSubscribeNotificationsClient) SendMsg(interface{}) error {
77+
return nil
78+
}
79+
80+
func (m *mockSubscribeNotificationsClient) RecvMsg(interface{}) error {
81+
return nil
82+
}
83+
84+
func TestManager_ReservationNotification(t *testing.T) {
85+
// Create a mock notification client
86+
recvChan := make(chan *swapserverrpc.SubscribeNotificationsResponse, 1)
87+
errChan := make(chan error, 1)
88+
mockStream := &mockSubscribeNotificationsClient{
89+
recvChan: recvChan,
90+
recvErrChan: errChan,
91+
}
92+
mockClient := &mockNotificationsClient{
93+
mockStream: mockStream,
94+
}
95+
96+
// Create a Manager with the mock client
97+
mgr := NewManager(&Config{
98+
Client: mockClient,
99+
FetchL402: func(ctx context.Context) error {
100+
// Simulate successful fetching of L402
101+
return nil
102+
},
103+
})
104+
105+
// Subscribe to reservation notifications.
106+
subCtx, subCancel := context.WithCancel(context.Background())
107+
subChan := mgr.SubscribeReservations(subCtx)
108+
109+
// Run the manager.
110+
ctx, cancel := context.WithCancel(context.Background())
111+
defer cancel()
112+
113+
rdyChan := make(chan struct{})
114+
go func() {
115+
err := mgr.Run(ctx, rdyChan)
116+
require.NoError(t, err)
117+
}()
118+
<-rdyChan
119+
120+
// Wait a bit to ensure manager is running and has subscribed
121+
time.Sleep(100 * time.Millisecond)
122+
require.Equal(t, 1, mockClient.timesCalled)
123+
124+
// Send a test notification
125+
testNotif := getTestNotification(testReservationId)
126+
127+
// Send the notification to the recvChan
128+
recvChan <- testNotif
129+
130+
// Collect the notification in the callback
131+
receivedNotification := <-subChan
132+
133+
// Now, check that the notification received in the callback matches the one sent
134+
require.NotNil(t, receivedNotification)
135+
require.Equal(t, testReservationId, receivedNotification.ReservationId)
136+
137+
// Cancel the subscription
138+
subCancel()
139+
140+
// Send another test notification`
141+
testNotif2 := getTestNotification(testReservationId2)
142+
recvChan <- testNotif2
143+
144+
// Wait a bit to ensure the notification is not received
145+
time.Sleep(100 * time.Millisecond)
146+
147+
require.Len(t, mgr.subscribers[NotificationTypeReservation], 0)
148+
149+
// Close the recvChan to stop the manager's receive loop
150+
close(recvChan)
151+
}
152+
153+
func getTestNotification(resId []byte) *swapserverrpc.SubscribeNotificationsResponse {
154+
return &swapserverrpc.SubscribeNotificationsResponse{
155+
Notification: &swapserverrpc.SubscribeNotificationsResponse_ReservationNotification{
156+
ReservationNotification: &swapserverrpc.ServerReservationNotification{
157+
ReservationId: resId,
158+
},
159+
},
160+
}
161+
}

0 commit comments

Comments
 (0)