Topic expiry attempt

server/server_manager.go

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

server/topic.go

@@ -8,7 +8,7 @@ import (
 )
 
 const (
-	rateVisitorExpiryDuration = 12 * time.Hour
+	topicExpiryDuration = 6 * time.Hour
 )
 
 // topic represents a channel to which subscribers can subscribe, and publishers
@@ -17,7 +17,7 @@ type topic struct {
 	ID          string
 	subscribers map[int]*topicSubscriber
 	rateVisitor *visitor
-	rateVisitorExpires time.Time
+	expires     time.Time
 	mu          sync.RWMutex
 }
 
@@ -54,25 +54,18 @@ func (t *topic) Subscribe(s subscriber, visitor *visitor, cancel func()) int {
 func (t *topic) Stale() bool {
 	t.mu.Lock()
 	defer t.mu.Unlock()
-	if t.rateVisitorExpires.Before(time.Now()) {
+	return len(t.subscribers) == 0 && t.expires.Before(time.Now())
-		t.rateVisitor = nil
-	}
-	return len(t.subscribers) == 0 && t.rateVisitor == nil
 }
 
 func (t *topic) SetRateVisitor(v *visitor) {
 	t.mu.Lock()
 	defer t.mu.Unlock()
 	t.rateVisitor = v
-	t.rateVisitorExpires = time.Now().Add(rateVisitorExpiryDuration)
 }
 
 func (t *topic) RateVisitor() *visitor {
 	t.mu.Lock()
 	defer t.mu.Unlock()
-	if t.rateVisitorExpires.Before(time.Now()) {
-		t.rateVisitor = nil
-	}
 	return t.rateVisitor
 }
 
@@ -81,6 +74,9 @@ func (t *topic) Unsubscribe(id int) {
 	t.mu.Lock()
 	defer t.mu.Unlock()
 	delete(t.subscribers, id)
+	if len(t.subscribers) == 0 {
+		t.expires = time.Now().Add(topicExpiryDuration)
+	}
 }
 
 // Publish asynchronously publishes to all subscribers