Coracle/nostrlib
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

return IncomingEvent struct from SimplePool methods containing the relay.

Browse Source
This commit is contained in:
fiatjaf
2023-09-30 19:16:30 -03:00
parent 18cee7421a
commit 978d7825b5
1 changed files with 14 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
pool.go
View File

@@ -16,6 +16,11 @@ type SimplePool struct {
cancel context.CancelFunc cancel context.CancelFunc
} }
type IncomingEvent struct {
*Event
Relay *Relay
}
func NewSimplePool(ctx context.Context) *SimplePool { func NewSimplePool(ctx context.Context) *SimplePool {
ctx, cancel := context.WithCancel(ctx) ctx, cancel := context.WithCancel(ctx)
@@ -52,8 +57,8 @@ func (pool *SimplePool) EnsureRelay(url string) (*Relay, error) {
// SubMany opens a subscription with the given filters to multiple relays // SubMany opens a subscription with the given filters to multiple relays
// the subscriptions only end when the context is canceled // the subscriptions only end when the context is canceled
func (pool *SimplePool) SubMany(ctx context.Context, urls []string, filters Filters) chan *Event { func (pool *SimplePool) SubMany(ctx context.Context, urls []string, filters Filters) chan IncomingEvent {
uniqueEvents := make(chan *Event) uniqueEvents := make(chan IncomingEvent)
seenAlready := xsync.NewMapOf[bool]() seenAlready := xsync.NewMapOf[bool]()
pending := xsync.NewCounter() pending := xsync.NewCounter()
@@ -74,7 +79,7 @@ func (pool *SimplePool) SubMany(ctx context.Context, urls []string, filters Filt
for evt := range sub.Events { for evt := range sub.Events {
// dispatch unique events to client // dispatch unique events to client
if _, ok := seenAlready.LoadOrStore(evt.ID, true); !ok { if _, ok := seenAlready.LoadOrStore(evt.ID, true); !ok {
uniqueEvents <- evt uniqueEvents <- IncomingEvent{Event: evt, Relay: relay}
} }
} }
@@ -89,10 +94,10 @@ func (pool *SimplePool) SubMany(ctx context.Context, urls []string, filters Filt
} }
// SubManyEose is like SubMany, but it stops subscriptions and closes the channel when gets a EOSE // SubManyEose is like SubMany, but it stops subscriptions and closes the channel when gets a EOSE
func (pool *SimplePool) SubManyEose(ctx context.Context, urls []string, filters Filters) chan *Event { func (pool *SimplePool) SubManyEose(ctx context.Context, urls []string, filters Filters) chan IncomingEvent {
ctx, cancel := context.WithCancel(ctx) ctx, cancel := context.WithCancel(ctx)
uniqueEvents := make(chan *Event) uniqueEvents := make(chan IncomingEvent)
seenAlready := xsync.NewMapOf[bool]() seenAlready := xsync.NewMapOf[bool]()
wg := sync.WaitGroup{} wg := sync.WaitGroup{}
wg.Add(len(urls)) wg.Add(len(urls))
@@ -133,7 +138,7 @@ func (pool *SimplePool) SubManyEose(ctx context.Context, urls []string, filters
// dispatch unique events to client // dispatch unique events to client
if _, ok := seenAlready.LoadOrStore(evt.ID, true); !ok { if _, ok := seenAlready.LoadOrStore(evt.ID, true); !ok {
select { select {
case uniqueEvents <- evt: case uniqueEvents <- IncomingEvent{Event: evt, Relay: relay}:
case <-ctx.Done(): case <-ctx.Done():
return return
} }
@@ -147,11 +152,11 @@ func (pool *SimplePool) SubManyEose(ctx context.Context, urls []string, filters
} }
// QuerySingle returns the first event returned by the first relay, cancels everything else. // QuerySingle returns the first event returned by the first relay, cancels everything else.
func (pool *SimplePool) QuerySingle(ctx context.Context, urls []string, filter Filter) *Event { func (pool *SimplePool) QuerySingle(ctx context.Context, urls []string, filter Filter) *IncomingEvent {
ctx, cancel := context.WithCancel(ctx) ctx, cancel := context.WithCancel(ctx)
defer cancel() defer cancel()
for evt := range pool.SubManyEose(ctx, urls, Filters{filter}) { for ievt := range pool.SubManyEose(ctx, urls, Filters{filter}) {
return evt return &ievt
} }
return nil return nil
} }