Skip to content

Commit 691cfbe

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 aa595b3 commit 691cfbe

File tree

3 files changed

+382
-0
lines changed

3 files changed

+382
-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: 197 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,197 @@
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 Client 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 Client
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. It will keep on running until the
85+
// context is canceled. It will subscribe to notifications and forward them to
86+
// the subscribers. On a first successful connection to the server, it will
87+
// close the readyChan to signal that the manager is ready.
88+
func (m *Manager) Run(ctx context.Context) error {
89+
// Initially we want to immediately try to connect to the server.
90+
waitTime := time.Duration(0)
91+
92+
// Start the notification runloop.
93+
for {
94+
timer := time.NewTimer(waitTime)
95+
// Increase the wait time for the next iteration.
96+
waitTime += time.Second * 10
97+
98+
// Return if the context has been canceled.
99+
select {
100+
case <-ctx.Done():
101+
return nil
102+
103+
case <-timer.C:
104+
}
105+
106+
// In order to create a valid l402 we first are going to call
107+
// the FetchL402 method. As a client might not have outbound capacity
108+
// yet, we'll retry until we get a valid response.
109+
if !m.hasL402 {
110+
err := m.cfg.FetchL402(ctx)
111+
if err != nil {
112+
log.Errorf("Error fetching L402: %v", err)
113+
continue
114+
}
115+
m.hasL402 = true
116+
}
117+
118+
connectedFunc := func() {
119+
// Reset the wait time to 10.
120+
waitTime = time.Second * 10
121+
}
122+
123+
err := m.subscribeNotifications(ctx, connectedFunc)
124+
if err != nil {
125+
log.Errorf("Error subscribing to notifications: %v", err)
126+
}
127+
}
128+
}
129+
130+
// subscribeNotifications subscribes to the notifications from the server.
131+
func (m *Manager) subscribeNotifications(ctx context.Context,
132+
connectedFunc func()) error {
133+
134+
callCtx, cancel := context.WithCancel(ctx)
135+
defer cancel()
136+
137+
notifStream, err := m.cfg.Client.SubscribeNotifications(
138+
callCtx, &swapserverrpc.SubscribeNotificationsRequest{},
139+
)
140+
if err != nil {
141+
return err
142+
}
143+
144+
// Signal that we're connected to the server.
145+
connectedFunc()
146+
log.Debugf("Successfully subscribed to server notifications")
147+
148+
for {
149+
notification, err := notifStream.Recv()
150+
if err == nil && notification != nil {
151+
log.Debugf("Received notification: %v", notification)
152+
m.handleNotification(notification)
153+
continue
154+
}
155+
156+
log.Errorf("Error receiving notification: %v", err)
157+
158+
return err
159+
}
160+
}
161+
162+
// handleNotification handles an incoming notification from the server,
163+
// forwarding it to the appropriate subscribers.
164+
func (m *Manager) handleNotification(notification *swapserverrpc.
165+
SubscribeNotificationsResponse) {
166+
167+
switch notification.Notification.(type) {
168+
case *swapserverrpc.SubscribeNotificationsResponse_ReservationNotification:
169+
// We'll forward the reservation notification to all subscribers.
170+
// Cleaning up any subscribers that have been canceled.
171+
newSubs := make(
172+
[]subscriber, 0, len(m.subscribers[NotificationTypeReservation]),
173+
)
174+
reservationNtfn := notification.GetReservationNotification()
175+
for _, sub := range m.subscribers[NotificationTypeReservation] {
176+
recvChan := sub.recvChan.(chan *swapserverrpc.
177+
ServerReservationNotification)
178+
179+
select {
180+
case <-sub.subCtx.Done():
181+
close(recvChan)
182+
continue
183+
184+
case recvChan <- reservationNtfn:
185+
newSubs = append(newSubs, sub)
186+
}
187+
}
188+
189+
m.Lock()
190+
m.subscribers[NotificationTypeReservation] = newSubs
191+
m.Unlock()
192+
193+
default:
194+
log.Warnf("Received unknown notification type: %v",
195+
notification)
196+
}
197+
}

‎notifications/manager_test.go‎

Lines changed: 159 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,159 @@
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+
go func() {
114+
err := mgr.Run(ctx)
115+
require.NoError(t, err)
116+
}()
117+
118+
// Wait a bit to ensure manager is running and has subscribed
119+
time.Sleep(100 * time.Millisecond)
120+
require.Equal(t, 1, mockClient.timesCalled)
121+
122+
// Send a test notification
123+
testNotif := getTestNotification(testReservationId)
124+
125+
// Send the notification to the recvChan
126+
recvChan <- testNotif
127+
128+
// Collect the notification in the callback
129+
receivedNotification := <-subChan
130+
131+
// Now, check that the notification received in the callback matches the one sent
132+
require.NotNil(t, receivedNotification)
133+
require.Equal(t, testReservationId, receivedNotification.ReservationId)
134+
135+
// Cancel the subscription
136+
subCancel()
137+
138+
// Send another test notification`
139+
testNotif2 := getTestNotification(testReservationId2)
140+
recvChan <- testNotif2
141+
142+
// Wait a bit to ensure the notification is not received
143+
time.Sleep(100 * time.Millisecond)
144+
145+
require.Len(t, mgr.subscribers[NotificationTypeReservation], 0)
146+
147+
// Close the recvChan to stop the manager's receive loop
148+
close(recvChan)
149+
}
150+
151+
func getTestNotification(resId []byte) *swapserverrpc.SubscribeNotificationsResponse {
152+
return &swapserverrpc.SubscribeNotificationsResponse{
153+
Notification: &swapserverrpc.SubscribeNotificationsResponse_ReservationNotification{
154+
ReservationNotification: &swapserverrpc.ServerReservationNotification{
155+
ReservationId: resId,
156+
},
157+
},
158+
}
159+
}

0 commit comments

Comments
 (0)