refactor SimplePool automatic reconnection.

This commit is contained in:
fiatjaf
2023-12-02 14:59:16 -03:00
parent a65bd94a00
commit 0e0ecb2412

30
pool.go
View File

@@ -79,15 +79,6 @@ func (pool *SimplePool) subMany(ctx context.Context, urls []string, filters Filt
ticker := time.NewTicker(seenAlreadyDropTick) ticker := time.NewTicker(seenAlreadyDropTick)
eose := false eose := false
updateSince := func() {
// After reconnection, update the since in the filter so that
// old events are not retrieved.
now := Now()
for i := range filters {
filters[i].Since = &now
}
}
pending := xsync.NewCounter() pending := xsync.NewCounter()
pending.Add(int64(len(urls))) pending.Add(int64(len(urls)))
for _, url := range urls { for _, url := range urls {
@@ -107,26 +98,25 @@ func (pool *SimplePool) subMany(ctx context.Context, urls []string, filters Filt
default: default:
} }
var sub *Subscription
relay, err := pool.EnsureRelay(nm) relay, err := pool.EnsureRelay(nm)
if err != nil { if err != nil {
time.Sleep(3 * time.Second) time.Sleep(3 * time.Second)
updateSince() goto reconnect
continue
} }
sub, err := relay.Subscribe(ctx, filters) sub, err = relay.Subscribe(ctx, filters)
if err != nil { if err != nil {
time.Sleep(3 * time.Second) time.Sleep(3 * time.Second)
updateSince() goto reconnect
continue
} }
loop:
for { for {
select { select {
case evt, more := <-sub.Events: case evt, more := <-sub.Events:
if !more { if !more {
break loop goto reconnect
} }
if unique { if unique {
if _, seen := seenAlready.LoadOrStore(evt.ID, evt.CreatedAt); seen { if _, seen := seenAlready.LoadOrStore(evt.ID, evt.CreatedAt); seen {
@@ -156,7 +146,13 @@ func (pool *SimplePool) subMany(ctx context.Context, urls []string, filters Filt
return return
} }
} }
updateSince()
reconnect:
// when attempting to reconnect update the `since` in filters so old events are not retrieved
now := Now()
for i := range filters {
filters[i].Since = &now
}
} }
}(NormalizeURL(url)) }(NormalizeURL(url))
} }