From bdf08bfd177a29399b4277899ebaf6c781e230cb Mon Sep 17 00:00:00 2001
From: disksing <i@disksing.com>
Date: Thu, 19 Nov 2015 13:52:41 +0800
Subject: [PATCH] *: address comments

---
 store/hbase/hbase_test.go | 42 +++++++++++++++++++++++++++++++
 store/hbase/kv.go         | 52 +++++++++++++++++++++------------------
 2 files changed, 70 insertions(+), 24 deletions(-)
 create mode 100644 store/hbase/hbase_test.go

diff --git a/store/hbase/hbase_test.go b/store/hbase/hbase_test.go
new file mode 100644
index 0000000000000..ad0deb316e77c
--- /dev/null
+++ b/store/hbase/hbase_test.go
@@ -0,0 +1,42 @@
+// Copyright 2015 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package hbasekv
+
+import (
+	"strings"
+
+	. "github.com/pingcap/check"
+)
+
+var _ = Suite(&testHBaseSuite{})
+
+type testHBaseSuite struct {
+}
+
+func (t *testHBaseSuite) TestParseDSN(c *C) {
+	zks, oracle, table, err := parseDSN("zk1.com,zk2,192.168.0.1|localhost:1234/tidb")
+	c.Assert(strings.Join(zks, " "), Equals, "zk1.com zk2 192.168.0.1")
+	c.Assert(oracle, Equals, "localhost:1234")
+	c.Assert(table, Equals, "tidb")
+	c.Assert(err, IsNil)
+
+	zks, oracle, table, err = parseDSN("zk1,zk2/tidb")
+	c.Assert(strings.Join(zks, " "), Equals, "zk1 zk2")
+	c.Assert(oracle, Equals, "")
+	c.Assert(table, Equals, "tidb")
+	c.Assert(err, IsNil)
+
+	_, _, _, err = parseDSN("zk1,zk2|localhost:1234")
+	c.Assert(err, NotNil)
+}
diff --git a/store/hbase/kv.go b/store/hbase/kv.go
index aa01bb8444078..356abc1fd65a9 100644
--- a/store/hbase/kv.go
+++ b/store/hbase/kv.go
@@ -49,8 +49,8 @@ var (
 )
 
 var (
-	// ErrDsnInvalid is returned when store dsn is invalid.
-	ErrDsnInvalid = errors.New("dsn invalid")
+	// ErrInvalidDSN is returned when store dsn is invalid.
+	ErrInvalidDSN = errors.New("invalid dsn")
 )
 
 type storeCache struct {
@@ -141,30 +141,16 @@ func (d Driver) Open(dsn string) (kv.Storage, error) {
 	mc.mu.Lock()
 	defer mc.mu.Unlock()
 
-	if len(dsn) == 0 {
-		return nil, errors.Trace(ErrDsnInvalid)
-	}
-	pos := strings.LastIndex(dsn, "/")
-	if pos == -1 {
-		return nil, errors.Trace(ErrDsnInvalid)
-	}
-	tableName := dsn[pos+1:]
-	addrs := dsn[:pos]
-
-	var tsoAddr string
-	pos = strings.LastIndex(addrs, "|")
-	if pos != -1 {
-		tsoAddr = addrs[pos+1:]
-		addrs = addrs[:pos]
-	}
-
-	zks := strings.Split(addrs, ",")
-
 	if store, ok := mc.cache[dsn]; ok {
 		// TODO: check the cache store has the same engine with this Driver.
 		return store, nil
 	}
 
+	zks, oracleAddr, tableName, err := parseDSN(dsn)
+	if err != nil {
+		return nil, errors.Trace(err)
+	}
+
 	// create buffered HBase connections, HBaseClient is goroutine-safe, so
 	// it's OK to redistribute to transactions.
 	conns := make([]hbase.HBaseClient, 0, hbaseConnPoolSize)
@@ -190,10 +176,10 @@ func (d Driver) Open(dsn string) (kv.Storage, error) {
 	}
 
 	var ora oracle.Oracle
-	if tsoAddr == "" {
-		ora = &oracles.LocalOracle{}
+	if oracleAddr == "" {
+		ora = oracles.NewLocalOracle()
 	} else {
-		ora = oracles.NewTsoOracle(tsoAddr)
+		ora = oracles.NewRemoteOracle(oracleAddr)
 	}
 
 	s := &hbaseStore{
@@ -205,3 +191,21 @@ func (d Driver) Open(dsn string) (kv.Storage, error) {
 	mc.cache[dsn] = s
 	return s, nil
 }
+
+func parseDSN(dsn string) (zks []string, oracleAddr, tableName string, err error) {
+	pos := strings.LastIndex(dsn, "/")
+	if pos == -1 {
+		err = errors.Trace(ErrInvalidDSN)
+		return
+	}
+	tableName = dsn[pos+1:]
+	addrs := dsn[:pos]
+
+	pos = strings.LastIndex(addrs, "|")
+	if pos != -1 {
+		oracleAddr = addrs[pos+1:]
+		addrs = addrs[:pos]
+	}
+	zks = strings.Split(addrs, ",")
+	return
+}