GlcoinCore/loop
1
0
Fork 0
mirror of https://github.com/lightninglabs/loop.git synced 2026-08-13 12:33:03 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

loop: add resume manager

Browse source
This commit is contained in:
Andras Banki-Horvath 2025-10-30 17:55:09 +01:00 • committed by sputn1ck
parent ded8ab5975
commit 460fc116d8
No known key found for this signature in database
GPG key ID: 671103D881A5F0E4
5 changed files with 620 additions and 249 deletions
Download patch file Download diff file Expand all files Collapse all files

57
notifications/manager.go
View file

@ -25,6 +25,10 @@ const (
// NotificationTypeStaticLoopInSweepRequest is the notification type for
// static loop in sweep requests.
NotificationTypeStaticLoopInSweepRequest
// NotificationTypeUnfinishedSwap is the notification type for unfinished
// swap notifications.
NotificationTypeUnfinishedSwap
)
const (
@ -101,13 +105,10 @@ func (m *Manager) SubscribeReservations(ctx context.Context,
m.addSubscriber(NotificationTypeReservation, sub)
// Start a goroutine to remove the subscriber when the context is
// canceled.
go func() {
<-ctx.Done()
context.AfterFunc(ctx, func() {
m.removeSubscriber(NotificationTypeReservation, sub)
close(notifChan)
}()
})
return notifChan
}
@ -120,6 +121,7 @@ func (m *Manager) SubscribeStaticLoopInSweepRequests(ctx context.Context,
notifChan := make(
chan *swapserverrpc.ServerStaticLoopInSweepNotification, 1,
)
sub := subscriber{
subCtx: ctx,
recvChan: notifChan,
@ -127,15 +129,34 @@ func (m *Manager) SubscribeStaticLoopInSweepRequests(ctx context.Context,
m.addSubscriber(NotificationTypeStaticLoopInSweepRequest, sub)
// Start a goroutine to remove the subscriber when the context is
// canceled.
go func() {
<-ctx.Done()
context.AfterFunc(ctx, func() {
m.removeSubscriber(
NotificationTypeStaticLoopInSweepRequest, sub,
NotificationTypeStaticLoopInSweepRequest,
sub,
)
close(notifChan)
}()
})
return notifChan
}
// SubscribeUnfinishedSwaps subscribes to the unfinished swap notifications.
func (m *Manager) SubscribeUnfinishedSwaps(ctx context.Context,
) <-chan *swapserverrpc.ServerUnfinishedSwapNotification {
notifChan := make(
chan *swapserverrpc.ServerUnfinishedSwapNotification, 1,
)
sub := subscriber{
subCtx: ctx,
recvChan: notifChan,
}
m.addSubscriber(NotificationTypeUnfinishedSwap, sub)
context.AfterFunc(ctx, func() {
m.removeSubscriber(NotificationTypeUnfinishedSwap, sub)
close(notifChan)
})
return notifChan
}
@ -293,6 +314,20 @@ func (m *Manager) handleNotification(ntfn *swapserverrpc.
recvChan <- staticLoopInSweepRequestNtfn
}
case *swapserverrpc.SubscribeNotificationsResponse_UnfinishedSwap: // nolint: lll
// We'll forward the unfinished swap notification to all
// subscribers.
unfinishedSwapNtfn := ntfn.GetUnfinishedSwap()
m.Lock()
defer m.Unlock()
for _, sub := range m.subscribers[NotificationTypeUnfinishedSwap] {
recvChan := sub.recvChan.(chan *swapserverrpc.
ServerUnfinishedSwapNotification)
recvChan <- unfinishedSwapNtfn
}
default:
log.Warnf("Received unknown notification type: %v",
ntfn)