func TestGetTimeRangeVersions(t *testing.T) {
	key := "TestGetTimeRangeVersions"
	c := gohbase.NewClient(*host)
	err := insertKeyValue(c, key, "cf", []byte("1"), hrpc.Timestamp(time.Unix(0, 50*1e6)))
	if err != nil {
		t.Fatalf("Put failed: %s", err)
	}
	err = insertKeyValue(c, key, "cf", []byte("1"), hrpc.Timestamp(time.Unix(0, 51*1e6)))
	if err != nil {
		t.Fatalf("Put failed: %s", err)
	}
	err = insertKeyValue(c, key, "cf", []byte("1"), hrpc.Timestamp(time.Unix(0, 49*1e6)))
	if err != nil {
		t.Fatalf("Put failed: %s", err)
	}

	var maxVersions uint32 = 2
	get, err := hrpc.NewGetStr(context.Background(), table, key,
		hrpc.Families(map[string][]string{"cf": nil}), hrpc.TimeRange(time.Unix(0, 0),
			time.Unix(0, 51*1e6)), hrpc.MaxVersions(maxVersions))
	rsp, err := c.Get(get)
	if err != nil {
		t.Fatalf("Get failed: %s", err)
	}
	if uint32(len(rsp.Cells)) != maxVersions {
		t.Fatalf("Expected versions: %d, Got versions: %d", maxVersions, len(rsp.Cells))
	}
	getTs1 := *rsp.Cells[0].Timestamp
	if getTs1 != 50 {
		t.Errorf("Timestamps are not the same. Expected Time: %v, Got Time: %v",
			50, getTs1)
	}
	getTs2 := *rsp.Cells[1].Timestamp
	if getTs2 != 49 {
		t.Errorf("Timestamps are not the same. Expected Time: %v, Got Time: %v",
			49, getTs2)
	}

	// get with no versions set
	get, err = hrpc.NewGetStr(context.Background(), table, key,
		hrpc.Families(map[string][]string{"cf": nil}), hrpc.TimeRange(time.Unix(0, 0),
			time.Unix(0, 51*1e6)))
	rsp, err = c.Get(get)
	if err != nil {
		t.Fatalf("Get failed: %s", err)
	}
	if uint32(len(rsp.Cells)) != 1 {
		t.Fatalf("Expected versions: %d, Got versions: %d", 1, len(rsp.Cells))
	}
	getTs1 = *rsp.Cells[0].Timestamp
	if getTs1 != 50 {
		t.Errorf("Timestamps are not the same. Expected Time: %v, Got Time: %v",
			50, getTs1)
	}
}

func TestPutMultipleCells(t *testing.T) {
	key := "row2.5"
	values := map[string]map[string][]byte{"cf": map[string][]byte{}, "cf2": map[string][]byte{}}
	values["cf"]["a"] = []byte("a")
	values["cf"]["b"] = []byte("b")
	values["cf2"]["a"] = []byte("a")
	c := gohbase.NewClient(*host)
	putRequest, err := hrpc.NewPutStr(context.Background(), table, key, values)
	_, err = c.Put(putRequest)
	if err != nil {
		t.Errorf("Put returned an error: %v", err)
	}
	families := map[string][]string{"cf": nil, "cf2": nil}
	get, err := hrpc.NewGetStr(context.Background(), table, key, hrpc.Families(families))
	rsp, err := c.Get(get)
	if err != nil {
		t.Errorf("Get returned an error: %v", err)
	}
	cells := rsp.GetResult().Cell
	if len(cells) != 3 {
		t.Errorf("Get expected 3 cells. Received: %d", len(cells))
	}
	for _, cell := range cells {
		if !bytes.Equal(cell.GetQualifier(), cell.GetValue()) {
			t.Errorf("Get returned an incorrect result. Expected: %v, Received: %v",
				cell.GetQualifier(), cell.GetValue())
		}
	}

}

func TestDeleteTimestamp(t *testing.T) {
	key := "TestDeleteTimestamp"
	c := gohbase.NewClient(*host)
	var putTs uint64 = 50
	timestamp := time.Unix(0, int64(putTs*1e6))
	err := insertKeyValue(c, key, "cf", []byte("1"), hrpc.Timestamp(timestamp))
	if err != nil {
		t.Fatalf("Put failed: %s", err)
	}
	deleteRequest, err := hrpc.NewDelStr(context.Background(), table, key,
		map[string]map[string][]byte{"cf": map[string][]byte{"a": nil}},
		hrpc.Timestamp(timestamp))
	_, err = c.Delete(deleteRequest)
	if err != nil {
		t.Fatalf("Delete failed: %s", err)
	}
	get, err := hrpc.NewGetStr(context.Background(), table, key,
		hrpc.Families(map[string][]string{"cf": nil}))
	rsp, err := c.Get(get)
	if err != nil {
		t.Fatalf("Get failed: %s", err)
	}
	if len(rsp.Cells) != 0 {
		t.Errorf("Timestamp wasn't deleted, get result length: %d", len(rsp.Cells))
	}
}

func TestMultiplePutsGetsSequentially(t *testing.T) {
	const num_ops = 100
	keyPrefix := "row3"
	headers := map[string][]string{"cf": nil}
	c := gohbase.NewClient(*host)
	err := performNPuts(keyPrefix, num_ops)
	if err != nil {
		t.Errorf("Put returned an error: %v", err)
	}
	for i := num_ops - 1; i >= 0; i-- {
		key := keyPrefix + fmt.Sprintf("%d", i)
		get, err := hrpc.NewGetStr(context.Background(), table, key, hrpc.Families(headers))
		rsp, err := c.Get(get)
		if err != nil {
			t.Errorf("Get returned an error: %v", err)
		}
		if len(rsp.Result.Cell) != 1 {
			t.Errorf("Incorrect number of cells returned by Get: %d", len(rsp.Result.Cell))
		}
		rsp_value := rsp.Result.Cell[0].GetValue()
		if !bytes.Equal(rsp_value, []byte(fmt.Sprintf("%d", i))) {
			t.Errorf("Get returned an incorrect result. Expected: %v, Got: %v",
				[]byte(fmt.Sprintf("%d", i)), rsp_value)
		}
	}
}

// Note: This function currently causes an infinite loop in the client throwing the error -
// 2015/06/19 14:34:11 Encountered an error while reading: Failed to read from the RS: EOF
func TestChangingRegionServers(t *testing.T) {
	key := "row8"
	val := []byte("1")
	headers := map[string][]string{"cf": nil}
	if host == nil {
		t.Fatal("Host is not set!")
	}
	c := gohbase.NewClient(*host)
	err := insertKeyValue(c, key, "cf", val)
	if err != nil {
		t.Errorf("Put returned an error: %v", err)
	}

	// RegionServer 1 hosts all the current regions.
	// Now launch servers 2,3
	test.LaunchRegionServers([]string{"2", "3"})

	// Now (gracefully) stop servers 1,2.
	// All regions should now be on server 3.
	test.StopRegionServers([]string{"1", "2"})
	get, err := hrpc.NewGetStr(context.Background(), table, key, hrpc.Families(headers))
	rsp, err := c.Get(get)
	if err != nil {
		t.Errorf("Get returned an error: %v", err)
	}
	rsp_value := rsp.Result.Cell[0].GetValue()
	if !bytes.Equal(rsp_value, val) {
		t.Errorf("Get returned an incorrect result. Expected: %v, Received: %v",
			val, rsp_value)
	}

	// Clean up by re-launching RS1 and closing RS3
	test.LaunchRegionServers([]string{"1"})
	test.StopRegionServers([]string{"3"})
}

func TestGetMultipleCells(t *testing.T) {
	key := "row1.75"
	c := gohbase.NewClient(*host, gohbase.FlushInterval(time.Millisecond*2))
	err := insertKeyValue(c, key, "cf", []byte("cf"))
	if err != nil {
		t.Errorf("Put returned an error: %v", err)
	}
	err = insertKeyValue(c, key, "cf2", []byte("cf2"))
	if err != nil {
		t.Errorf("Put returned an error: %v", err)
	}

	families := map[string][]string{"cf": nil, "cf2": nil}
	get, err := hrpc.NewGetStr(context.Background(), table, key, hrpc.Families(families))
	rsp, err := c.Get(get)
	cells := rsp.GetResult().Cell
	num_results := len(cells)
	if num_results != 2 {
		t.Errorf("Get expected 2 cells. Received: %d", num_results)
	}
	for _, cell := range cells {
		if !bytes.Equal(cell.GetFamily(), cell.GetValue()) {
			t.Errorf("Get returned an incorrect result. Expected: %v, Received: %v",
				cell.GetFamily(), cell.GetValue())
		}
	}
}

// CheckTable returns an error if the given table name doesn't exist.
func (c *Client) CheckTable(ctx context.Context, table string) error {
	getStr, err := hrpc.NewGetStr(ctx, table, "theKey")
	if err == nil {
		_, err = c.SendRPC(getStr)
	}
	return err
}

// CheckTable returns an error if the given table name doesn't exist.
func (c *Client) CheckTable(ctx context.Context, table string) (*pb.GetResponse, error) {
	getStr, _ := hrpc.NewGetStr(ctx, table, "theKey")
	resp, err := c.sendRPC(getStr)
	if err != nil {
		return nil, err
	}
	return resp.(*pb.GetResponse), err
}

func TestGet(t *testing.T) {
	key := "row1"
	val := []byte("1")
	headers := map[string][]string{"cf": nil}
	if host == nil {
		t.Fatal("Host is not set!")
	}

	c := gohbase.NewClient(*host)
	err := insertKeyValue(c, key, "cf", val)
	if err != nil {
		t.Errorf("Put returned an error: %v", err)
	}

	get, err := hrpc.NewGetStr(context.Background(), table, key, hrpc.Families(headers))
	if err != nil {
		t.Fatalf("Failed to create Get request: %s", err)
	}
	rsp, err := c.Get(get)
	if err != nil {
		t.Errorf("Get returned an error: %v", err)
	}
	rsp_value := rsp.Result.Cell[0].GetValue()
	if !bytes.Equal(rsp_value, val) {
		t.Errorf("Get returned an incorrect result. Expected: %v, Got: %v",
			val, rsp_value)
	}

	get.ExistsOnly()
	rsp, err = c.Get(get)
	if err != nil {
		t.Errorf("Get returned an error: %v", err)
	} else if !*rsp.Result.Exists {
		t.Error("Get claimed that our row didn't exist")
	}

	ctx, _ := context.WithTimeout(context.Background(), 0)
	get, err = hrpc.NewGetStr(ctx, table, key, hrpc.Families(headers))
	if err != nil {
		t.Fatalf("Failed to create Get request: %s", err)
	}
	_, err = c.Get(get)
	if err != gohbase.ErrDeadline {
		t.Errorf("Get ignored the deadline")
	}
}

func TestTimeRangeError(t *testing.T) {
	key := "TestTimeRangeError"
	table := "test"
	_, err := hrpc.NewGetStr(context.Background(), table, key,
		hrpc.Families(map[string][]string{"cf": nil}), hrpc.TimeRange(time.Unix(51, 0),
			time.Unix(51, 0)))
	expErr := "'from' timestamp (51000ms) is greater or equal to 'to' timestamp (51000ms)"
	if err.Error() != expErr {
		t.Errorf("Expected error: %s, Got error: %s", expErr, err)
	}
	_, err = hrpc.NewGetStr(context.Background(), table, key,
		hrpc.Families(map[string][]string{"cf": nil}), hrpc.TimeRange(time.Unix(52, 0),
			time.Unix(51, 0)))
	expErr = "'from' timestamp (52000ms) is greater or equal to 'to' timestamp (51000ms)"
	if err.Error() != expErr {
		t.Errorf("Expected error: %s, Got error: %s", expErr, err)
	}
}

func doTest(client gohbase.Client, action string) (err error) {

	switch action {
	case "get":
		var getReq *hrpc.Get
		getReq, err = hrpc.NewGetStr(context.Background(), "t1", "10001")
		if err != nil {
			fmt.Printf("hrpc.NewGetStr failed, err is %v", err)
			return
		}
		_, err = client.Get(getReq)
		if err != nil {
			fmt.Printf("client.Get failed, err is %v", err)
			return
		}
	case "put":
		var putReq *hrpc.Mutate
		rowID, _ := IntRange(0, 3000000)
		putReq, err = hrpc.NewPutStr(context.Background(), "t1", strconv.Itoa(rowID), map[string]map[string][]byte{
			"fam1": map[string][]byte{
				"col1": []byte("testvalue"),
			},
		})
		if err != nil {
			fmt.Printf("hrpc.NewPutStr failed, err is %v", err)
			return
		}
		_, err = client.Put(putReq)
		if err != nil {
			fmt.Printf("client.Put failed, err is %v", err)
			return
		}
	case "delete":
		var delReq *hrpc.Mutate
		rowID, _ := IntRange(0, 3000000)
		delReq, err = hrpc.NewDelStr(context.Background(), "t1", strconv.Itoa(rowID), map[string]map[string][]byte{
			"fam1": map[string][]byte{
				"col1": []byte("testvalue"),
			},
		})
		if err != nil {
			fmt.Printf("hrpc.NewDelStr failed, err is %v", err)
			return
		}
		_, err = client.Delete(delReq)
		if err != nil {
			fmt.Printf("client.Del failed, err is %v", err)
			return
		}
	}
	return
}

func BenchmarkGet(b *testing.B) {
	b.ReportAllocs()
	keyPrefix := "row10"
	err := performNPuts(keyPrefix, b.N)
	if err != nil {
		b.Errorf("Put returned an error: %v", err)
	}
	c := gohbase.NewClient(*host)
	b.ResetTimer()
	headers := map[string][]string{"cf": nil}
	for i := 0; i < b.N; i++ {
		key := keyPrefix + fmt.Sprintf("%d", i)
		get, _ := hrpc.NewGetStr(context.Background(), table, key, hrpc.Families(headers))
		c.Get(get)
	}
}

func TestGetBadColumnFamily(t *testing.T) {
	key := "row1.625"
	c := gohbase.NewClient(*host)
	err := insertKeyValue(c, key, "cf", []byte("Bad!"))
	if err != nil {
		t.Errorf("Put returned an error: %v", err)
	}
	families := map[string][]string{"badcf": nil}
	get, err := hrpc.NewGetStr(context.Background(), table, key, hrpc.Families(families))
	rsp, err := c.Get(get)
	if err == nil {
		t.Errorf("Get didn't return an error! (It should have)")
	}
	if rsp.GetResult() != nil {
		t.Errorf("Get expected no result. Received: %v", rsp.GetResult())
	}
}

func TestGetDoesntExist(t *testing.T) {
	key := "row1.5"
	c := gohbase.NewClient(*host)
	headers := map[string][]string{"cf": nil}
	get, err := hrpc.NewGetStr(context.Background(), table, key, hrpc.Families(headers))
	rsp, err := c.Get(get)
	if err != nil {
		t.Errorf("Get returned an error: %v", err)
	} else if results := len(rsp.GetResult().Cell); results != 0 {
		t.Errorf("Get expected 0 cells. Received: %d", results)
	}

	get.ExistsOnly()
	rsp, err = c.Get(get)
	if err != nil {
		t.Errorf("Get returned an error: %v", err)
	} else if *rsp.Result.Exists {
		t.Error("Get claimed that our non-existent row exists")
	}
}

func TestPutTimestamp(t *testing.T) {
	key := "TestPutTimestamp"
	c := gohbase.NewClient(*host)
	var putTs uint64 = 50
	timestamp := time.Unix(0, int64(putTs*1e6))
	err := insertKeyValue(c, key, "cf", []byte("1"), hrpc.Timestamp(timestamp))
	if err != nil {
		t.Fatalf("Put failed: %s", err)
	}
	get, err := hrpc.NewGetStr(context.Background(), table, key,
		hrpc.Families(map[string][]string{"cf": nil}))
	rsp, err := c.Get(get)
	if err != nil {
		t.Fatalf("Get failed: %s", err)
	}
	getTs := *rsp.Cells[0].Timestamp
	if getTs != putTs {
		t.Errorf("Timestamps are not the same. Put Time: %v, Get Time: %v",
			putTs, getTs)
	}
}

func TestAppend(t *testing.T) {
	key := "row7"
	c := gohbase.NewClient(*host)
	// Inserting "Hello"
	insertErr := insertKeyValue(c, key, "cf", []byte("Hello"))
	if insertErr != nil {
		t.Errorf("Put returned an error: %v", insertErr)
	}
	// Appending " my name is Dog."
	values := map[string]map[string][]byte{"cf": map[string][]byte{}}
	values["cf"]["a"] = []byte(" my name is Dog.")
	appRequest, err := hrpc.NewAppStr(context.Background(), table, key, values)
	appRsp, err := c.Append(appRequest)
	if err != nil {
		t.Errorf("Append returned an error: %v", err)
	}
	if appRsp.GetResult() == nil {
		t.Errorf("Append doesn't return updated value.")
	}
	// Verifying new result is "Hello my name is Dog."
	result := appRsp.GetResult().Cell[0].GetValue()
	if !bytes.Equal([]byte("Hello my name is Dog."), result) {
		t.Errorf("Append returned an incorrect result. Expected: %v, Receieved: %v",
			[]byte("Hello my name is Dog."), result)
	}

	// Make sure the change was actually committed.
	headers := map[string][]string{"cf": nil}
	get, err := hrpc.NewGetStr(context.Background(), table, key, hrpc.Families(headers))
	rsp, err := c.Get(get)
	cells := rsp.GetResult().Cell
	if len(cells) != 1 {
		t.Errorf("Get expected 1 cells. Received: %d", len(cells))
	}
	result = cells[0].GetValue()
	if !bytes.Equal([]byte("Hello my name is Dog."), result) {
		t.Errorf("Append returned an incorrect result. Expected: %v, Receieved: %v",
			[]byte("Hello my name is Dog."), result)
	}
}

func TestTimestampIncreasing(t *testing.T) {
	key := "row4"
	c := gohbase.NewClient(*host)
	var oldTime uint64 = 0
	headers := map[string][]string{"cf": nil}
	for i := 0; i < 10; i++ {
		insertKeyValue(c, key, "cf", []byte("1"))
		get, err := hrpc.NewGetStr(context.Background(), table, key, hrpc.Families(headers))
		rsp, err := c.Get(get)
		if err != nil {
			t.Errorf("Get returned an error: %v", err)
			break
		}
		newTime := rsp.GetResult().Cell[0].GetTimestamp()
		if newTime <= oldTime {
			t.Errorf("Timestamps are not increasing. Old Time: %v, New Time: %v",
				oldTime, newTime)
		}
		oldTime = newTime
		time.Sleep(time.Millisecond)
	}
}

func TestMultiplePutsGetsParallel(t *testing.T) {
	const num_ops = 1000
	keyPrefix := "row3.5"
	headers := map[string][]string{"cf": nil}
	c := gohbase.NewClient(*host)
	// TODO: Currently have to CheckTable before initiating the N requests
	// 	 otherwise we face runaway client generation - one for each request.
	c.CheckTable(context.Background(), table)
	var wg sync.WaitGroup
	for i := 0; i < num_ops; i++ {
		wg.Add(1)
		go func(client *gohbase.Client, key string) {
			defer wg.Done()
			err := insertKeyValue(client, key, "cf", []byte(key))
			if err != nil {
				t.Errorf("(Parallel) Put returned an error: %v", err)
			}
		}(c, keyPrefix+fmt.Sprintf("%d", i))
	}
	wg.Wait()
	// All puts are complete. Now do the same for gets.
	for i := num_ops - 1; i >= 0; i-- {
		wg.Add(1)
		go func(client *gohbase.Client, key string) {
			defer wg.Done()
			get, err := hrpc.NewGetStr(context.Background(), table, key, hrpc.Families(headers))
			rsp, err := c.Get(get)
			if err != nil {
				t.Errorf("(Parallel) Get returned an error: %v", err)
			} else {
				rsp_value := rsp.Result.Cell[0].GetValue()
				if !bytes.Equal(rsp_value, []byte(key)) {
					t.Errorf("Get returned an incorrect result.")
				}
			}
		}(c, keyPrefix+fmt.Sprintf("%d", i))
	}
	wg.Wait()
}

func main() {
	client := gohbase.NewClient("localhost", gohbase.RpcQueueSize(1), gohbase.FlushInterval(time.Nanosecond))

	t := time.Now()
	for i := 0; i < 100; i++ {
		// Values maps a ColumnFamily -> Qualifiers -> Values.
		values := map[string]map[string][]byte{"cf": map[string][]byte{"ab": []byte("kmztest")}}
		putRequest, err := hrpc.NewPutStr(context.Background(), "test", "key"+strconv.Itoa(i), values)
		checkErr(err)
		_, err = client.Put(putRequest)
		checkErr(err)
	}
	fmt.Println(float64(time.Now().UnixNano()-t.UnixNano()) / 1000000)

	fmt.Println("=====================")

	getRequest, err := hrpc.NewGetStr(context.Background(), "test", "row")
	checkErr(err)
	getRsp, err := client.Get(getRequest)
	checkErr(err)
	fmt.Println(getRsp)

}

func testTsunaHbaseClient(concurrency, eachCount int) {
	client := gohbase.NewClient("172.16.13.94:2181,172.16.13.94:2182,172.16.13.94:2183")
	getReq, err := hrpc.NewGetStr(context.Background(), "t", "10001")
	if err != nil {
		fmt.Printf("hrpc.NewGetStr failed, err is %v", err)
		return
	}
	st := time.Now().Unix()
	wg := &sync.WaitGroup{}
	wg.Add(concurrency)
	failCount := make([]int, concurrency)
	for j := 0; j < concurrency; j++ {
		go func(seq int) {
			curFailCount := 0
			defer wg.Done()
			for i := 0; i < eachCount; i++ {
				getRes, err := client.Get(getReq)
				if err != nil {
					fmt.Printf("client.Get failed, err is %v", err)
					curFailCount += 1
					// return
				}
				// fmt.Printf("getRes is %v", getRes)
				_ = getRes
			}
			failCount[seq] = curFailCount
		}(j)
	}
	wg.Wait()
	failCountSum := 0
	for _, curFailCount := range failCount {
		failCountSum += curFailCount
	}
	et := time.Now().Unix()
	fmt.Printf("cost: %d, total: %d, failed: %d\n", et-st, concurrency*eachCount, failCountSum)
}