func InitHub(url string) error {
	c, err := redis.DialURL(url)
	if err != nil {
		return err
	}
	pubconn = c

	c, err = redis.DialURL(url)
	if err != nil {
		return err
	}
	subconn = redis.PubSubConn{c}

	go func() {
		for {
			switch v := subconn.Receive().(type) {
			case redis.Message:
				EmitLocal(v.Channel, string(v.Data))

			case error:
				panic(v)
			}
		}
	}()

	return nil
}

func redisconn(redisdb *redis.Conn) {
	if *redisdb == nil {
		conn, err := redis.DialURL(os.Getenv("REDIS_URL"))
		checkErr(err)
		*redisdb = conn
	}
	_, err := (*redisdb).Do("PING")
	if err != nil {
		conn, err := redis.DialURL(os.Getenv("REDIS_URL"))
		checkErr(err)
		*redisdb = conn
	}
}

//Open 2 connections to redis
func Connect(url string) error {
	c, err := redis.DialURL(url)
	if err != nil {
		return err
	}
	pubconn = c

	c, err = redis.DialURL(url)
	if err != nil {
		return err
	}
	subconn = redis.PubSubConn{c}
	return nil
}

func TestMain(m *testing.M) {
	// clean test dir
	os.RemoveAll("/tmp/clusterTest")

	// initialize backend if redis-server found
	initialize()

	conn, err := redis.DialURL(config.ClusterConnection, redis.DialConnectTimeout(30*time.Second), redis.DialPassword(config.ClusterToken))
	if err != nil {
		return
	}
	hostname, _ := os.Hostname()
	self := fmt.Sprintf("%v:%v", hostname, config.ApiPort)
	defer conn.Do("SREM", "members", self)
	defer conn.Close()

	rtn := m.Run()

	// clean test dir
	os.RemoveAll("/tmp/clusterTest")
	// just in case, ensure clean members
	conn.Do("SREM", "members", self)
	conn.Close()

	os.Exit(rtn)
}

func (site Site) redisdb() redis.Conn {
	redisdb, err := redis.DialURL(site.redisURL())
	if err != nil {
		panic(err)
	}
	return redisdb
}

func getRedisConn(redisURI string) redis.Conn {
	redisConn, err := redis.DialURL(redisURI)
	if err != nil {
		log.Panicln("Error with redis.DialURL", err.Error())
	}
	return redisConn
}

// updateMaster gets the address of the master node from sentinel
func updateMaster() error {
	config.Log.Debug("Contacting sentinel for address of master...")

	// connect to sentinel in order to query for the master address
	r, err := redis.DialURL("redis://"+config.SentinelAddress, redis.DialConnectTimeout(config.TimeoutNotReady), redis.DialReadTimeout(config.TimeoutSentinelPoll), redis.DialPassword(config.SentinelPassword))
	if err != nil {
		return fmt.Errorf("Failed to reach sentinel - %v", err)
	}

	// retrieve the master redis address
	addr, err := redis.Strings(r.Do("SENTINEL", "get-master-addr-by-name", config.MonitorName))
	if err != nil {
		return fmt.Errorf("Failed to get-master-addr-by-name - %v", err)
	}

	// cleanup after ourselves
	r.Close()

	// construct a useable address from sentinel's response
	masterAddr = fmt.Sprintf("%v:%v", addr[0], addr[1])
	config.Log.Debug("Master address: '%v'", masterAddr)

	// wait for redis to transition to master
	if err = verifyMaster(masterAddr, config.SentinelPassword); err != nil {
		return fmt.Errorf("Could not verify master - %v", err)
	}

	return nil
}

// Connect to remote instance of Redis using a URL.
func ExampleDialURL() {
	c, err := redis.DialURL(os.Getenv("REDIS_URL"))
	if err != nil {
		// handle connection error
	}
	defer c.Close()
}

// verifyMaster verifies that the decided master node has fully transitioned
func verifyMaster(addr, pass string) error {
	// connect to redis in order to verify its state
	r, err := redis.DialURL("redis://"+addr, redis.DialConnectTimeout(config.TimeoutNotReady), redis.DialPassword(pass))
	if err != nil {
		return fmt.Errorf("Failed to reach redis at: '%v'", addr)
	}

	// give redis some time to transition
	timeout := time.After(config.TimeoutMasterWait)

	for {
		select {
		case <-timeout:
			return fmt.Errorf("Timed out waiting for redis to transition to master")
		default:
			// retrieve the redis node's role
			info, err := redis.Bytes(r.Do("INFO", "replication"))
			if err != nil {
				return fmt.Errorf("Failed to get INFO - %v", err)
			}

			// check if node is master
			if strings.Contains(string(info), "role:master") {
				return nil
			}
		}
	}

	// cleanup after ourselves
	r.Close()

	return nil
}

/*
AnalysePool can be calls to process redis queue(s).
analyzerID will set which redis AnalysePool will connect to (redis:=pool[len(urls)%AnalysePool])

exitOnEmpty is a closure function which will control inner loop of AnalysePool when queue is empty.
	exitOnEmpty := func() bool {
		return true
	}
analyzer is a closure function which will be called for processing the tasks popped from queue.
	analyzer := func(id int, task chan string, success chan bool, next chan bool) {
		for {
			select {
			case msg := <-task:
				if id == 2 {
					time.Sleep(20 * time.Millisecond)
				}
				fmt.Println(id, msg)
				if msg == "stop" {
					<-next
					success <- true
					return
				}
			case <-time.After(2 * time.Second):
				fmt.Println("no new event for 2 seconds for ID", id)
				<-next
				success <- false
				return
			}
		}
	}
Analyser clousre must be able to accept the new Tasks without delay and if needed process them concurrently. Delay in accepting new Task will block AnalysePool.
*/
func (q *Queue) AnalysePool(analyzerID int, exitOnEmpty func() bool, analyzer func(int, chan string, chan bool, chan bool)) {
	redisdb, _ := redis.DialURL(q.urls[q.redisID(analyzerID)])

	next := make(chan bool, q.analyzerBuff())
	pool := make(map[int]chan string)
	for {
		id, task := q.removeTask(redisdb, q.queueName(analyzerID))
		if task == "" {
			if exitOnEmpty() {
				break
			} else {
				time.Sleep(100 * time.Millisecond)
			}
		} else {
			if pool[id] == nil {
				pool[id] = make(chan string)
				success := make(chan bool)
				go analyzer(id, pool[id], success, next)
				go q.waitforSuccess(id, success, pool)
				pool[id] <- task
				next <- true
			} else {
				pool[id] <- task
			}
		}
	}

	for i := 0; i < q.analyzerBuff(); i++ {
		next <- true
	}
}

// Partitions will define the number of redis partitions for queue. It is useful if one redis for any reason cannot handle the load of analyser.
// This function will accept a slice of redisURL to set the redis pool.
func Partitions(urls []string) {
	redisParitions = len(urls)
	redisPool = redisPool[:0]
	for _, v := range urls {
		r, _ := redis.DialURL(v)
		redisPool = append(redisPool, redisStruct{r, v})
	}
}

func NewCacheRedis(address string) *CacheRedis {
	c, err := redis.DialURL(address)
	if err != nil {
		panic(err)
	}

	return &CacheRedis{conn: c}
}

func NewClient(address string, options Options) *Client {
	pool := &redis.Pool{
		Dial: func() (redis.Conn, error) {
			return redis.DialURL(address)
		},
	}
	return NewClientWithPool(pool, options)
}

func TestDialURLErrors(t *testing.T) {
	for _, d := range dialErrors {
		_, err := redis.DialURL(d.rawurl)
		if err == nil || !strings.Contains(err.Error(), d.expectedError) {
			t.Errorf("DialURL did not return expected error (expected %v to contain %s)", err, d.expectedError)
		}
	}
}

func waitforSuccess(n int, id int, success chan bool, pool map[int]chan string) {
	redisdb, _ := redis.DialURL(redisPool[id%redisParitions].url)
	redisdb.Do("SET", "PENDING::"+strconv.Itoa(id), 1)
	r := <-success
	if r {
		delete(pool, id)
		redisdb.Do("DEL", "PENDING::"+strconv.Itoa(id))
	}
}

func (q *Queue) waitforSuccess(id int, success chan bool, pool map[int]chan string) {
	redisdb, _ := redis.DialURL(q.urls[q.redisID(id)])
	redisdb.Do("SET", q.pendingKeyName(id), 1)
	r := <-success
	if r {
		delete(pool, id)
		redisdb.Do("DEL", q.pendingKeyName(id))
	}
}

// NewMemoryStore creates the new store.
func NewRedisStore(rawurl string, options ...redis.DialOption) *RedisStore {
	c, err := redis.DialURL(rawurl, options...)
	if err != nil {
		fmt.Println("NewRedisStore Error", err)
		return nil
	}
	s := RedisStore{c: c}
	return &s
}

func runLogger(redisURI, queueName string, logChannel chan []byte) {
	redisConn, err := redis.DialURL(redisURI)
	logError("redis.DialURL Failed: %v\n", err)

	for {
		logEntryBytes := <-logChannel
		_, err = redisConn.Do("lpush", queueName, logEntryBytes)
		logError("Redis LPUSH failed: %v\n", err)
	}
}

// Urls will accept a slice of redis connection URLS. This slice will setup the connections and also set how many redis paritions will be used.
// Setting more than one redis is useful in some cases that a single redis can't handle a the queue load either because of IO and memory restrictions or if possible CPU.
func (q *Queue) Urls(urls []string) {
	q.urls = q.urls[:0]
	q.pool = q.pool[:0]
	for _, v := range urls {
		c, e := redis.DialURL(v)
		checkErr(e)
		q.urls = append(q.urls, v)
		q.pool = append(q.pool, c)
	}
}

func (r *redisStore) initRedisConn() {
	fmt.Println("链接redis服务器")
	c, err := redis.DialURL("redis://192.168.6.108")

	if err != nil {
		fmt.Println("redis初始化失败")
		fmt.Println(err)
	}
	r.redis = c
}