aljaz/lndhub.go
1
1
Fork 0
mirror of https://github.com/getAlby/lndhub.go.git synced 2025-12-20 14:14:47 +01:00
Code Issues Packages Projects Releases Wiki Activity

background routine: add context cancel support

Browse Source
This commit is contained in:
kiwiidb
2022-04-22 14:32:48 +02:00
parent 8dd8c14cd7
commit 624909484e
2 changed files with 31 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
integration_tests/websocket_test.go
View File

@@ -271,6 +271,7 @@ func (suite *WebSocketTestSuite) TestWebSocketMissingInvoice() {
func (suite *WebSocketTestSuite) TearDownSuite() { func (suite *WebSocketTestSuite) TearDownSuite() {
suite.invoiceUpdateSubCancelFn() suite.invoiceUpdateSubCancelFn()
suite.websocketServer.Close() suite.websocketServer.Close()
clearTable(suite.service, "invoices")
} }
func TestWebSocketSuite(t *testing.T) { func TestWebSocketSuite(t *testing.T) {

54
lib/service/invoicesubscription.go
View File

@@ -4,6 +4,7 @@ import (
"context" "context"
"database/sql" "database/sql"
"encoding/hex" "encoding/hex"
"fmt"
"strings" "strings"
"time" "time"
@@ -122,32 +123,37 @@ func (svc *LndhubService) InvoiceUpdateSubscription(ctx context.Context) error {
return err return err
} }
for { for {
// receive the next invoice update select {
rawInvoice, err := invoiceSubscriptionStream.Recv() case <-ctx.Done():
if err != nil { return fmt.Errorf("Context was canceled")
svc.Logger.Errorf("Error processing invoice update subscription: %v", err) default:
sentry.CaptureException(err) // receive the next invoice update
// TODO: close the stream somehoe before retrying? rawInvoice, err := invoiceSubscriptionStream.Recv()
// Wait 30 seconds and try to reconnect if err != nil {
// TODO: implement some backoff svc.Logger.Errorf("Error processing invoice update subscription: %v", err)
time.Sleep(30 * time.Second) sentry.CaptureException(err)
invoiceSubscriptionStream, _ = svc.ConnectInvoiceSubscription(ctx) // TODO: close the stream somehoe before retrying?
continue // Wait 30 seconds and try to reconnect
} // TODO: implement some backoff
time.Sleep(30 * time.Second)
invoiceSubscriptionStream, _ = svc.ConnectInvoiceSubscription(ctx)
continue
}
// Ignore updates for open invoices // Ignore updates for open invoices
// We store the invoice details in the AddInvoice call // We store the invoice details in the AddInvoice call
// Processing open invoices here could cause a race condition: // Processing open invoices here could cause a race condition:
// We could get this notification faster than we finish the AddInvoice call // We could get this notification faster than we finish the AddInvoice call
if rawInvoice.State == lnrpc.Invoice_OPEN { if rawInvoice.State == lnrpc.Invoice_OPEN {
svc.Logger.Infof("Invoice state is open. Ignoring update. r_hash:%v", hex.EncodeToString(rawInvoice.RHash)) svc.Logger.Infof("Invoice state is open. Ignoring update. r_hash:%v", hex.EncodeToString(rawInvoice.RHash))
continue continue
} }
processingError := svc.ProcessInvoiceUpdate(ctx, rawInvoice) processingError := svc.ProcessInvoiceUpdate(ctx, rawInvoice)
if processingError != nil { if processingError != nil {
svc.Logger.Error(processingError) svc.Logger.Error(processingError)
sentry.CaptureException(processingError) sentry.CaptureException(processingError)
}
} }
} }
} }