func (c *resultsConn) sendResult(v []interface{}, wg *sync.WaitGroup) {
	defer wg.Done()
	var rp message.ResPayload
	if err := unmarshalBRPOPValue(&rp, v); err != nil {
		if c.vars != nil {
			c.vars.Add("FailedResPayloadUnmarshals", 1)
		}
		logf(c.logFn, "Results: BRPOP failed to unmarshal result payload: %v", err)
		return
	}
	k := fmt.Sprintf(resTimeoutKey, rp.ConnUUID, rp.MsgUUID)
	rc := c.pool.Get()
	defer rc.Close()
	rc = clusterifyConn(rc, k)
	pttl, err := redis.Int(delAndPTTLScript.Do(rc, k))
	if err != nil {
		if c.vars != nil {
			c.vars.Add("FailedPTTLResults", 1)
		}
		logf(c.logFn, "Results: DEL/PTTL failed: %v", err)
		return
	}
	if pttl <= 0 {
		if c.vars != nil {
			c.vars.Add("ExpiredResults", 1)
		}
		logf(c.logFn, "Results: message %v expired, dropping call", rp.MsgUUID)
		return
	}
	c.ch <- &rp
	if c.vars != nil {
		c.vars.Add("Results", 1)
	}
}
receives the raw value v retured from BRPOP.
package redisbroker
import (
	"expvar"
	"fmt"
	"sync"
	"time"
	"github.com/mna/juggler/broker"
	"github.com/mna/juggler/message"
	"github.com/garyburd/redigo/redis"
	"github.com/pborman/uuid"
)
var _ broker.ResultsConn = (*resultsConn)(nil)
type resultsConn struct {
	c        redis.Conn
	pool     Pool
	connUUID uuid.UUID
	timeout  time.Duration
	logFn    func(string, ...interface{})
	vars     *expvar.Map
	once sync.Once
	ch   chan *message.ResPayload
	errmu sync.Mutex
	err   error
}
func (c *resultsConn) Close() error {
	return c.c.Close()
}
func (c *resultsConn) ResultsErr() error {
	c.errmu.Lock()
	err := c.err
	c.errmu.Unlock()
	return err
}
func (c *resultsConn) Results() <-chan *message.ResPayload {
	c.once.Do(func() {
		c.ch = make(chan *message.ResPayload)
		key := fmt.Sprintf(resKey, c.connUUID)
		to := int(c.timeout / time.Second)
		rc := clusterifyConn(c.c, key)
		go c.pollResults(rc, key, to)
	})
	return c.ch
}
func (c *resultsConn) pollResults(pollConn redis.Conn, key string, timeout int) {
	defer close(c.ch)
	wg := sync.WaitGroup{}
	for {
		v, err := redis.Values(pollConn.Do("BRPOP", key, timeout))
		if err != nil {
			if err == redis.ErrNil {
				continue
			}
			c.errmu.Lock()
			c.err = err
			c.errmu.Unlock()
			wg.Wait()
			return
		}
		wg.Add(1)
		go c.sendResult(v, &wg)
	}
}