SubscribeManyNotifyClosed()

This commit is contained in:
fiatjaf
2025-12-04 08:51:34 -03:00
parent eb2be4efe1
commit d5dc3abaf2
2 changed files with 36 additions and 12 deletions

View File

@@ -81,21 +81,17 @@ func loadWalletFromPool(
kinds = append(kinds, 7376) kinds = append(kinds, 7376)
} }
eoseChanE := make(chan struct{}) events, eoseChanE := pool.SubscribeManyNotifyEOSE(
events := pool.SubscribeManyNotifyEOSE(
ctx, ctx,
relays, relays,
nostr.Filter{Kinds: kinds, Authors: []nostr.PubKey{pk}}, nostr.Filter{Kinds: kinds, Authors: []nostr.PubKey{pk}},
eoseChanE,
nostr.SubscriptionOptions{}, nostr.SubscriptionOptions{},
) )
eoseChanD := make(chan struct{}) deletions, eoseChanD := pool.SubscribeManyNotifyEOSE(
deletions := pool.SubscribeManyNotifyEOSE(
ctx, ctx,
relays, relays,
nostr.Filter{Kinds: []nostr.Kind{5}, Tags: nostr.TagMap{"k": []string{"7375"}}, Authors: []nostr.PubKey{pk}}, nostr.Filter{Kinds: []nostr.Kind{5}, Tags: nostr.TagMap{"k": []string{"7375"}}, Authors: []nostr.PubKey{pk}},
eoseChanD,
nostr.SubscriptionOptions{}, nostr.SubscriptionOptions{},
) )

40
pool.go
View File

@@ -234,7 +234,7 @@ func (pool *Pool) SubscribeMany(
filter Filter, filter Filter,
opts SubscriptionOptions, opts SubscriptionOptions,
) chan RelayEvent { ) chan RelayEvent {
return pool.subMany(ctx, urls, filter, nil, opts) return pool.subMany(ctx, urls, filter, nil, nil, opts)
} }
// FetchMany opens a subscription, much like SubscribeMany, but it ends as soon as all Relays // FetchMany opens a subscription, much like SubscribeMany, but it ends as soon as all Relays
@@ -260,16 +260,33 @@ func (pool *Pool) FetchMany(
return pool.subManyEose(ctx, urls, filter, opts) return pool.subManyEose(ctx, urls, filter, opts)
} }
// SubscribeManyNotifyEOSE is like SubscribeMany, but takes a channel that is closed when // SubscribeManyNotifyEOSE is like SubscribeMany, but also returns a channel that is closed when all subscriptions have received an EOSE
// all subscriptions have received an EOSE
func (pool *Pool) SubscribeManyNotifyEOSE( func (pool *Pool) SubscribeManyNotifyEOSE(
ctx context.Context, ctx context.Context,
urls []string, urls []string,
filter Filter, filter Filter,
eoseChan chan struct{},
opts SubscriptionOptions, opts SubscriptionOptions,
) chan RelayEvent { ) (chan RelayEvent, chan struct{}) {
return pool.subMany(ctx, urls, filter, eoseChan, opts) eoseChan := make(chan struct{})
events := pool.subMany(ctx, urls, filter, eoseChan, nil, opts)
return events, eoseChan
}
type RelayClosed struct {
Reason string
Relay *Relay
}
// SubscribeManyNotifyClosed is like SubscribeMany, but also returns a channel that emits every time a subscription receives a CLOSED message
func (pool *Pool) SubscribeManyNotifyClosed(
ctx context.Context,
urls []string,
filter Filter,
opts SubscriptionOptions,
) (chan RelayEvent, chan RelayClosed) {
closedChan := make(chan RelayClosed)
events := pool.subMany(ctx, urls, filter, nil, closedChan, opts)
return events, closedChan
} }
type ReplaceableKey struct { type ReplaceableKey struct {
@@ -382,6 +399,7 @@ func (pool *Pool) subMany(
urls []string, urls []string,
filter Filter, filter Filter,
eoseChan chan struct{}, eoseChan chan struct{},
closedChan chan RelayClosed,
opts SubscriptionOptions, opts SubscriptionOptions,
) chan RelayEvent { ) chan RelayEvent {
ctx, cancel := context.WithCancelCause(ctx) ctx, cancel := context.WithCancelCause(ctx)
@@ -524,6 +542,15 @@ func (pool *Pool) subMany(
} }
} else { } else {
debugLogf("CLOSED from %s: '%s'\n", nm, reason) debugLogf("CLOSED from %s: '%s'\n", nm, reason)
if closedChan != nil {
select {
case closedChan <- RelayClosed{
Reason: reason,
Relay: relay,
}:
default:
}
}
} }
return return
@@ -752,6 +779,7 @@ func (pool *Pool) BatchedSubscribeMany(
[]string{df.Relay}, []string{df.Relay},
df.Filter, df.Filter,
nil, nil,
nil,
opts, opts,
) { ) {
select { select {