for {

// Loop reading messages from conn.Receive() (via

// msgChannel) until the context is cancelled.

select {

case msg, ok := <-msgChannel:

if !ok {

return

}

switch msg := msg.(type) {

case redis.Message:

var res fleet.DistributedQueryResult

err := json.Unmarshal(msg.Data, &res)

if err != nil {

outChannel <- err

}

outChannel <- res

case error:

outChannel <- errors.Wrap(msg, "reading from redis")

}

case <-ctx.Done():

conn.Unsubscribe()

}

}

ctx.Done()