From f9896e0162fd6ad458d53f30262f36986ba3b5e0 Mon Sep 17 00:00:00 2001
From: Sergiusz Urbaniak <sergiusz.urbaniak@gmail.com>
Date: Mon, 13 Aug 2018 08:14:19 +0200
Subject: [PATCH] provider/mem: cleanup closed listener in GC

... rather than in the Subscribe method. Currently the cleanup for a
given Alert subscription is done in a blocking goroutine, started in
the Subscribe method.

This simplifies it by moving the cleanup to the GC.

Additionally it simplifies the subscribe method by setting up the
buffered channel big enough to fill it up with all pending alerts
preventing the necessity to start a goroutine in Subscribe at all.

Signed-off-by: Sergiusz Urbaniak <sergiusz.urbaniak@gmail.com>
---
 provider/mem/mem.go | 45 ++++++++++++++++++++++++---------------------
 1 file changed, 24 insertions(+), 21 deletions(-)

diff --git a/provider/mem/mem.go b/provider/mem/mem.go
index fc9e1093..7e0875c6 100644
--- a/provider/mem/mem.go
+++ b/provider/mem/mem.go
@@ -75,6 +75,16 @@ func (a *Alerts) runGC() {
 			}
 		}
 
+		for i, l := range a.listeners {
+			select {
+			case <-l.done:
+				delete(a.listeners, i)
+				close(l.alerts)
+			default:
+				// listener is not closed yet, hence proceed.
+			}
+		}
+
 		a.mtx.Unlock()
 	}
 }
@@ -85,15 +95,27 @@ func (a *Alerts) Close() error {
 	return nil
 }
 
+func max(a, b int) int {
+	if a > b {
+		return a
+	}
+	return b
+}
+
 // Subscribe returns an iterator over active alerts that have not been
 // resolved and successfully notified about.
 // They are not guaranteed to be in chronological order.
 func (a *Alerts) Subscribe() provider.AlertIterator {
+	alerts, err := a.getPending()
+
 	var (
-		ch   = make(chan *types.Alert, alertChannelLength)
+		ch   = make(chan *types.Alert, max(len(alerts), alertChannelLength))
 		done = make(chan struct{})
 	)
-	alerts, err := a.getPending()
+
+	for _, a := range alerts {
+		ch <- a
+	}
 
 	a.mtx.Lock()
 	i := a.next
@@ -101,25 +123,6 @@ func (a *Alerts) Subscribe() provider.AlertIterator {
 	a.listeners[i] = listeningAlerts{alerts: ch, done: done}
 	a.mtx.Unlock()
 
-	go func() {
-		defer func() {
-			a.mtx.Lock()
-			delete(a.listeners, i)
-			close(ch)
-			a.mtx.Unlock()
-		}()
-
-		for _, a := range alerts {
-			select {
-			case ch <- a:
-			case <-done:
-				return
-			}
-		}
-
-		<-done
-	}()
-
 	return provider.NewAlertIterator(ch, done, err)
 }