From bbf7f83784e376bafb3ac77ce22d5e0377629dbc Mon Sep 17 00:00:00 2001
From: David Zhao <david@davidzhao.com>
Date: Wed, 9 Jun 2021 09:51:44 -0700
Subject: [PATCH] Improve integration test reliability, wait for RedisRouter to
 start

---
 pkg/routing/redisrouter.go | 17 ++++++++++++++---
 1 file changed, 14 insertions(+), 3 deletions(-)

diff --git a/pkg/routing/redisrouter.go b/pkg/routing/redisrouter.go
index 48f7cf136..2c6ef39fa 100644
--- a/pkg/routing/redisrouter.go
+++ b/pkg/routing/redisrouter.go
@@ -246,9 +246,18 @@ func (r *RedisRouter) Start() error {
 	if !r.isStarted.TrySet(true) {
 		return nil
 	}
+
+	workerStarted := make(chan struct{})
 	go r.statsWorker()
-	go r.redisWorker()
-	return nil
+	go r.redisWorker(workerStarted)
+
+	// wait until worker is running
+	select {
+	case <-workerStarted:
+		return nil
+	case <-time.After(3 * time.Second):
+		return errors.New("Unable to start redis router")
+	}
 }
 
 func (r *RedisRouter) Stop() {
@@ -309,7 +318,7 @@ func (r *RedisRouter) statsWorker() {
 }
 
 // worker that consumes redis messages intended for this node
-func (r *RedisRouter) redisWorker() {
+func (r *RedisRouter) redisWorker(startedChan chan struct{}) {
 	defer func() {
 		logger.Debugw("finishing redisWorker", "node", r.currentNode.Id)
 	}()
@@ -318,6 +327,8 @@ func (r *RedisRouter) redisWorker() {
 	sigChannel := signalNodeChannel(r.currentNode.Id)
 	rtcChannel := rtcNodeChannel(r.currentNode.Id)
 	r.pubsub = r.rc.Subscribe(r.ctx, sigChannel, rtcChannel)
+
+	close(startedChan)
 	for msg := range r.pubsub.Channel() {
 		if msg == nil {
 			return