astra
/
ntfy
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

rename lastVisitor to vRate

pull/609/head
Karmanyaah Malhotra 2023-02-21 20:18:04 -06:00
parent bc3d897d7a
commit 0e4044b747
2 changed files with 17 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
server/server.go
View File

@ -1417,7 +1417,7 @@ func (s *Server) execManager() {
if ev.IsTrace() { if ev.IsTrace() {
expiryMessage := "" expiryMessage := ""
if subs == 0 { if subs == 0 {
expiryTime := time.Until(t.lastVisitorExpires) expiryTime := time.Until(t.vRateExpires)
expiryMessage = ", expires in " + expiryTime.String() expiryMessage = ", expires in " + expiryTime.String()
} }
ev.Trace("- topic %s: %d subscribers%s", t.ID, subs, expiryMessage) ev.Trace("- topic %s: %d subscribers%s", t.ID, subs, expiryMessage)

32
server/topic.go
View File

@ -11,11 +11,11 @@ import (
// topic represents a channel to which subscribers can subscribe, and publishers // topic represents a channel to which subscribers can subscribe, and publishers
// can publish a message // can publish a message
type topic struct { type topic struct {
ID string ID string
subscribers map[int]*topicSubscriber subscribers map[int]*topicSubscriber
lastVisitor *visitor vRate *visitor
lastVisitorExpires time.Time vRateExpires time.Time
mu sync.Mutex mu sync.Mutex
} }
type topicSubscriber struct { type topicSubscriber struct {
@ -49,9 +49,9 @@ func (t *topic) Subscribe(s subscriber, visitor *visitor, cancel func(), subscri
} }
// if no subscriber is already handling the rate limit // if no subscriber is already handling the rate limit
if t.lastVisitor == nil && subscriberRateLimit { if t.vRate == nil && subscriberRateLimit {
t.lastVisitor = visitor t.vRate = visitor
t.lastVisitorExpires = time.Time{} t.vRateExpires = time.Time{}
} }
return subscriberID return subscriberID
@ -61,16 +61,16 @@ func (t *topic) Stale() bool {
t.mu.Lock() t.mu.Lock()
defer t.mu.Unlock() defer t.mu.Unlock()
// if Time is initialized (not the zero value) and the expiry time has passed // if Time is initialized (not the zero value) and the expiry time has passed
if !t.lastVisitorExpires.IsZero() && t.lastVisitorExpires.Before(time.Now()) { if !t.vRateExpires.IsZero() && t.vRateExpires.Before(time.Now()) {
t.lastVisitor = nil t.vRate = nil
} }
return len(t.subscribers) == 0 && t.lastVisitor == nil return len(t.subscribers) == 0 && t.vRate == nil
} }
func (t *topic) Billee() *visitor { func (t *topic) Billee() *visitor {
t.mu.Lock() t.mu.Lock()
defer t.mu.Unlock() defer t.mu.Unlock()
return t.lastVisitor return t.vRate
} }
// Unsubscribe removes the subscription from the list of subscribers // Unsubscribe removes the subscription from the list of subscribers
@ -84,16 +84,16 @@ func (t *topic) Unsubscribe(id int) {
// look for an active subscriber (in random order) that wants to handle the rate limit // look for an active subscriber (in random order) that wants to handle the rate limit
for _, v := range t.subscribers { for _, v := range t.subscribers {
if v.subscriberRateLimit { if v.subscriberRateLimit {
t.lastVisitor = v.visitor t.vRate = v.visitor
t.lastVisitorExpires = time.Time{} t.vRateExpires = time.Time{}
return return
} }
} }
// if no active subscriber is found, count it towards the leaving subscriber // if no active subscriber is found, count it towards the leaving subscriber
if deletingSub.subscriberRateLimit { if deletingSub.subscriberRateLimit {
t.lastVisitor = deletingSub.visitor t.vRate = deletingSub.visitor
t.lastVisitorExpires = time.Now().Add(subscriberBilledValidity) t.vRateExpires = time.Now().Add(subscriberBilledValidity)
} }
} }