simplify subscription closing.

This commit is contained in:
fiatjaf
2023-08-22 10:41:58 -03:00
parent cd86ee2514
commit 446b104990
2 changed files with 22 additions and 25 deletions

View File

@ -490,7 +490,6 @@ func (r *Relay) PrepareSubscription(ctx context.Context, filters Filters, opts .
events: make(chan *Event), events: make(chan *Event),
EndOfStoredEvents: make(chan struct{}), EndOfStoredEvents: make(chan struct{}),
Filters: filters, Filters: filters,
closeEventsChannel: make(chan struct{}),
} }
for _, opt := range opts { for _, opt := range opts {

View File

@ -31,7 +31,6 @@ type Subscription struct {
live atomic.Bool live atomic.Bool
eosed atomic.Bool eosed atomic.Bool
closeEventsChannel chan struct{}
cancel context.CancelFunc cancel context.CancelFunc
} }
@ -78,14 +77,14 @@ func (sub *Subscription) start() {
mu.Unlock() mu.Unlock()
}() }()
case <-sub.Context.Done(): case <-sub.Context.Done():
// the subscription ends once the context is canceled // the subscription ends once the context is canceled (if not already)
sub.Unsub() sub.Unsub() // this will set sub.live to false
return
case <-sub.closeEventsChannel: // do this so we don't have the possibility of closing the Events channel and then trying to send to it
// this is called only once on .Unsub() and closes the .Events channel
mu.Lock() mu.Lock()
close(sub.Events) close(sub.Events)
mu.Unlock() mu.Unlock()
return return
} }
} }
@ -94,17 +93,16 @@ func (sub *Subscription) start() {
// Unsub closes the subscription, sending "CLOSE" to relay as in NIP-01. // Unsub closes the subscription, sending "CLOSE" to relay as in NIP-01.
// Unsub() also closes the channel sub.Events and makes a new one. // Unsub() also closes the channel sub.Events and makes a new one.
func (sub *Subscription) Unsub() { func (sub *Subscription) Unsub() {
// cancel the context (if it's not canceled already)
sub.cancel() sub.cancel()
// naïve sync.Once implementation: // mark subscription as closed and send a CLOSE to the relay (naïve sync.Once implementation)
if sub.live.CompareAndSwap(true, false) { if sub.live.CompareAndSwap(true, false) {
go sub.Close() sub.Close()
id := sub.GetID()
sub.Relay.Subscriptions.Delete(id)
// do this so we don't have the possibility of closing the Events channel and then trying to send to it
close(sub.closeEventsChannel)
} }
// remove subscription from our map
sub.Relay.Subscriptions.Delete(sub.GetID())
} }
// Close just sends a CLOSE message. You probably want Unsub() instead. // Close just sends a CLOSE message. You probably want Unsub() instead.