func (m *RedisEngine) RunTask() {
	client, err := redis.Dial("tcp", m.RedisServer)
	if err != nil {
		log.Println("redis connection err", err)
	}
	defer client.Close()
	for {
		select {
		case <-m.ExitChannel:
			return
		case request := <-m.CmdChannel:
			err = hystrix.Do("RedisCmd", func() error {
				reply := client.Cmd(request.Cmd, request.Args...)
				if reply.Err != nil {
					return reply.Err
				}
				request.ReplyChannel <- reply
				return nil
			}, func(error) error {
				if err != nil {
					log.Println(request.Cmd, request.Args, err)
				}
				client.Close()
				client, err = redis.Dial("tcp", m.RedisServer)
				return err
			})

		}
	}
}

func InitDB() {
	var err error
	Db, err = redis.Dial(NETWORK, Config.DB)
	if err != nil {
		panic(fmt.Sprintf("%v", err))
	}
}

func New(url string) *RedisStore {
	client, err := radix.Dial("tcp", url)
	if err != nil {
		panic(err)
	}
	return &RedisStore{client}
}

func (suite *HandlerSuite) SetupTest() {
	var err error
	df := func(network, addr string) (*redis.Client, error) {
		client, err := redis.Dial(network, addr)
		// fmt.Println("DIaling")
		if err != nil {
			return nil, err
		}
		err = client.Cmd("SELECT", 8).Err
		if err != nil {
			return nil, err
		}
		err = client.Cmd("FLUSHDB").Err
		if err != nil {
			return nil, err
		}
		return client, nil
	}
	redisPool, err = pool.NewCustomPool("tcp", redisURL, 1, df)
	if err != nil {
		panic(err)
	}

	c, err := redisPool.Get()
	errorHandler(err)
	defer redisPool.Put(c)
	SetDomain("xyz1234567890", "peterbe.com", c)

}

func PushNode(settings *yaml.Yaml, message Message) {
	debug := settings.Get("debug").(bool)
	redis_connection := settings.Get("redis_connection").(string)
	redis_list := settings.Get("redis_list").(string)
	redis_db := settings.Get("redis_db").(int)

	t := time.Now()
	ts := t.Format("Mon Jan 2 15:04:05 -0700 MST 2006")
	rc, err := redis.Dial("tcp", redis_connection)
	if err != nil {
		fmt.Printf("[%s] ERROR Error received: %s\n", ts, err)
	}
	if debug {
		fmt.Printf("[%s] INFO Connection made to Redis server %s\n", ts, redis_connection)
	}
	r := rc.Cmd("SELECT", redis_db)
	if r.Err != nil {
		fmt.Printf("[%s] ERROR Error received: %s\n", ts, r.Err)
	} else {
		if debug {
			fmt.Printf("[%s] INFO Redis database selected: %d\n", ts, redis_db)
		}
	}
	j, errj := json.Marshal(message)
	if errj != nil {
		fmt.Printf("[%s] ERROR Error received: %s\n", ts, errj)
	}
	r = rc.Cmd("LPUSH", redis_list, j)
	rc.Close()
}

func RedisClient() *redis.Client {
	client, err := redis.Dial("tcp", redisURL())
	if err != nil {
		panic("Could not connect to redis on " + redisURL())
	}
	return client
}

// Create a new Output for redis processing.
func NewOutput(u *url.URL) (io.Output, error) {
	if client, err := redis.Dial("tcp", u.Host); err == nil {
		return &Output{client: client}, nil
	} else {
		return nil, err
	}
}

func AssignmentsHandler(w http.ResponseWriter, r *http.Request) {

	r.ParseForm()
	timeslot_id := r.FormValue("assignment[timeslot_id]")
	boat_id := r.FormValue("assignment[boat_id]")

	redisClient, _ := redis.Dial("tcp", "127.0.0.1:6379")

	data_ts, _ := redisClient.Cmd("GET", "ts:"+timeslot_id).Bytes()
	data_boat, _ := redisClient.Cmd("GET", "boat:"+boat_id).Bytes()

	var boat Boat
	json.Unmarshal(data_boat, &boat)

	var timestamp Timeslot
	json.Unmarshal(data_ts, &timestamp)

	timestamp.Boats = append(timestamp.Boats, boat)
	timestamp.Availability = max(timestamp.Availability, boat.Capacity)

	ts, _ := json.Marshal(timestamp)
	redisClient.Cmd("SET", "ts:"+timeslot_id, ts)
	redisClient.Cmd("SADD", "asmt:"+boat_id, "ts:"+timeslot_id)

	redisClient.Close()
}

func TestReset(t *T) {
	// Simply initializing a cluster proves Reset works to some degree, since
	// NewCluster calls Reset
	cluster := getCluster(t)
	old7000 := cluster.clients["127.0.0.1:7000"]
	old7001 := cluster.clients["127.0.0.1:7001"]

	// We make a bogus client and add it to the cluster to prove that it gets
	// removed, since it's not needed
	client, err := redis.Dial("tcp", "127.0.0.1:6379")
	assert.Nil(t, err)
	cluster.clients["127.0.0.1:6379"] = client

	err = cluster.Reset()
	assert.Nil(t, err)

	// Prove that the bogus client is closed
	closedErr := errors.New("use of closed network connection")
	assert.Equal(t, closedErr, client.Cmd("PING").Err)

	// Prove that the remaining two addresses are still in clients, were not
	// reconnected, and still work
	assert.Equal(t, 2, len(cluster.clients))
	assert.Equal(t, old7000, cluster.clients["127.0.0.1:7000"])
	assert.Equal(t, old7001, cluster.clients["127.0.0.1:7001"])
	assert.NotNil(t, cluster.clients["127.0.0.1:7000"])
	assert.NotNil(t, cluster.clients["127.0.0.1:7001"])
	assert.Nil(t, cluster.Cmd("GET", "foo").Err)
	assert.Nil(t, cluster.Cmd("GET", "bar").Err)
}

func TimeslotHandler(w http.ResponseWriter, r *http.Request) {

	r.ParseForm()
	start_time, _ := strconv.ParseInt(r.FormValue("timeslot[start_time]"), 10, 0)
	dur, _ := strconv.ParseInt(r.FormValue("timeslot[duration]"), 10, 0)
	date := r.FormValue("date")
	var timeslot []byte

	redisClient, _ := redis.Dial("tcp", "127.0.0.1:6379")

	if len(date) == 0 {

		t1 := time.Unix(start_time, 0)
		dateKey := fmt.Sprintf("%d-%02d-%02d", t1.Year(), t1.Month(), t1.Day())

		v := &Timeslot{
			Id:             uuid.New(),
			Start_time:     start_time,
			Duration:       dur,
			Availability:   0,
			Customer_count: 0,
			Boats:          []Boat{},
		}
		timeslot, _ = json.Marshal(v)

		redisClient.Cmd("MULTI")
		redisClient.Cmd("SET", "ts:"+v.Id, timeslot)
		redisClient.Cmd("SADD", dateKey, "ts:"+v.Id)
		redisClient.Cmd("EXEC")
	} else {
		var ts []interface{}
		r := redisClient.Cmd("SMEMBERS", date)

		for i := range r.Elems {
			elemStr, _ := r.Elems[i].Str()
			data, _ := redisClient.Cmd("GET", elemStr).Bytes()

			var timestamp interface{}
			json.Unmarshal(data, &timestamp)

			bs := timestamp.(map[string]interface{})["Boats"]
			for i = 0; i < len(bs.([]interface{})); i++ {
				Id := (bs.([]interface{})[i]).(map[string]interface{})["Id"]
				delete((bs.([]interface{})[i]).(map[string]interface{}), "Id")
				delete((bs.([]interface{})[i]).(map[string]interface{}), "Name")
				delete((bs.([]interface{})[i]).(map[string]interface{}), "Capacity")
				bs.([]interface{})[i] = Id
			}
			ts = append(ts, timestamp)
		}
		timeslot, _ = json.Marshal(ts)
	}

	redisClient.Close()
	status := 200
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(status)
	w.Write([]byte(timeslot))
}

func SetupRedisConnection() *redis.Client {
	c, e := redis.Dial("tcp", ":6379")

	if e != nil {
		panic(e)
	}
	return c
}

// Retrieves an available redis client. If there are none available it will
// create a new one on the fly
func (p *Pool) Get() (*redis.Client, error) {
	select {
	case conn := <-p.pool:
		return conn, nil
	default:
		return redis.Dial(p.network, p.addr)
	}
}

func initWebRedis() {
	var err error
	webDb, err = redis.Dial("tcp", "127.0.0.1:6379")
	if err != nil {
		fmt.Println(err)
		os.Exit(1)
	}
}

func NewRedisQueue(name string) *RedisQueue {
	client, err := redis.Dial("tcp", "localhost:6379")
	if err != nil {
		panic(err)
	}
	return &RedisQueue{
		c:    client,
		name: name,
	}
}

func RedisQueueManagerFactory(qfactory QueueFactory) QueueManager {
	client, err := redis.Dial("tcp", "localhost:6379")
	if err != nil {
		panic(err)
	}
	return &RedisQueueManager{
		c:        client,
		qfactory: qfactory,
	}
}

func connect(host string, port int) *redis.Client {
	connectionString := fmt.Sprintf("%s:%d", host, port)
	conn, err := redis.Dial("tcp", connectionString)

	if err != nil {
		fmt.Println("Cannot connect to Redis, exiting.")
		os.Exit(int(syscall.ECONNREFUSED))
	}

	return conn
}

func main() {
	logFilePtr := flag.String("file", "", "Path to log file")
	thresholdPtr := flag.Int("threshold", 10, "Threshold for failed logins")
	flag.Parse()

	if *logFilePtr != "" {
		file, err := os.Open(*logFilePtr)
		if err != nil {
			fmt.Println("Couldn't open logfile:", err)
			os.Exit(1)
		}
		defer file.Close()

		var lines []string
		scanner := bufio.NewScanner(file)
		for scanner.Scan() {
			lines = append(lines, scanner.Text())
		}

		var entries map[string]int
		entries = make(map[string]int)

		for _, entry := range lines {
			parts := strings.Split(entry, " ")
			l := logEntry{
				Address:      parts[0],
				Method:       strings.Replace(parts[5], "\"", "", 1),
				Uri:          parts[6],
				ResponseCode: parts[8],
			}
			if l.Method == "POST" && l.ResponseCode == "200" {
				entries[l.Address] += 1
			}
		}

		if len(entries) > 0 {
			connection, err := redis.Dial("tcp", "localhost:6379")
			if err != nil {
				fmt.Println("Couldn't connect to Redis:", err)
				os.Exit(1)
			}

			connection.Cmd("MULTI")
			for k, v := range entries {
				if v >= *thresholdPtr {
					fmt.Printf("Blacklisting %s. Threshold: %d, Actual: %d\n", k, *thresholdPtr, v)
					actorString := fmt.Sprintf("%s:repsheet:ip:blacklisted", k)
					connection.Cmd("SET", actorString, "Failed login processor")
				}
			}
			connection.Cmd("EXEC")
		}
	}
}

func clientOrAbort(t *testing.T) *redis.Client {
	c, err := redis.Dial("tcp", os.Getenv(RedisEnv))
	if err != nil {
		t.Log("could not connect to a redis instance")
		t.SkipNow()
	}
	if err := c.Cmd("FLUSHALL").Err; err != nil {
		t.Fatal(err)
	}
	return c
}

func RedisListLpush(lname, value string) bool {
	client, err := redis.Dial("tcp", redisIp)
	errHndlr(err)
	defer client.Close()

	// select database
	r := client.Cmd("select", redisDb)
	errHndlr(r.Err)

	result, _ := client.Cmd("lpush", lname, value).Bool()
	return result
}

func RedisRemoveHashField(hkey, field string) bool {
	client, err := redis.Dial("tcp", redisIp)
	errHndlr(err)
	defer client.Close()

	// select database
	r := client.Cmd("select", redisDb)
	errHndlr(r.Err)

	result, _ := client.Cmd("hdel", hkey, field).Bool()
	return result
}