added more docs

README.md

@@ -1,7 +1,7 @@
 gocql
 =====
 
-Package gocql implements a fast and robust Cassandra driver for the
+Package gocql implements a fast and robust Cassandra client for the
 Go programming language.
 
 Installation
@@ -13,9 +13,19 @@ Installation
 Features
 --------
 
-* Modern Cassandra client for Cassandra 2.0
-* Built-In support for UUIDs (version 1 and 4)
-
+* modern Cassandra client for Cassandra 1.2 and 2.0
+* automatic type conversations between Cassandra and Go
+  * support for all common types including sets, lists and maps
+  * custom types can implement a `Marshaler` and `Unmarshaler` interface
+  * strict type conversations without any loss of precision
+  * built-In support for UUIDs (version 1 and 4)
+* support for logged, unlogged and counter batches
+* cluster management
+  * automatic reconnect on connection failures with exponential falloff
+  * round robin distribution of queries to different hosts
+  * round robin distribution of queries to different connections on a host
+  * each connection can execute up to 128 concurrent queries
+* automatic query preparation
 
 Example
 -------
@@ -25,54 +35,44 @@ package main
 
 import (
 	"fmt"
-	"github.com/tux21b/gocql"
 	"log"
+
+	"github.com/tux21b/gocql"
+	"github.com/tux21b/gocql/uuid"
 )
 
 func main() {
-	// connect to your cluster
-	db := gocql.NewSession(gocql.Config{
-		Nodes: []string{
-			"192.168.1.1",
-			"192.168.1.2",
-			"192.168.1.3",
-		},
-		Keyspace:    "example",       // (optional)
-		Consistency: gocql.ConQuorum, // (optional)
-	})
-	defer db.Close()
-
-	// simple query
-	var title, text string
-	if err := db.Query("SELECT title, text FROM posts WHERE title = ?",
-		"Lorem Ipsum").Scan(&title, &text); err != nil {
+	// connect to the cluster
+	cluster := gocql.NewCluster("192.168.1.1", "192.168.1.2", "192.168.1.3")
+	cluster.Keyspace = "example"
+	cluster.Consistency = gocql.Quorum
+	session := cluster.CreateSession()
+	defer session.Close()
+
+	// insert a tweet
+	if err := session.Query(`INSERT INTO tweet (timeline, id, text) VALUES (?, ?, ?)`,
+		"me", uuid.TimeUUID(), "hello world").Exec(); err != nil {
 		log.Fatal(err)
 	}
-	fmt.Println(title, text)
 
-	// iterator example
-	var titles []string
-	iter := db.Query("SELECT title FROM posts").Iter()
-	for iter.Scan(&title) {
-		titles = append(titles, title)
-	}
-	if err := iter.Close(); err != nil {
-		log.Fatal(err)
-	}
-	fmt.Println(titles)
+	var id uuid.UUID
+	var text string
 
-	// insertion example (with custom consistency level)
-	if err := db.Query("INSERT INTO posts (title, text) VALUES (?, ?)",
-		"New Title", "foobar").Consistency(gocql.ConAny).Exec(); err != nil {
+	// select a single tweet
+	if err := session.Query(`SELECT id, text FROM tweet WHERE timeline = ? LIMIT 1`,
+		"me").Consistency(gocql.One).Scan(&id, &text); err != nil {
 		log.Fatal(err)
 	}
+	fmt.Println("Tweet:", id, text)
 
-	// prepared queries
-	query := gocql.NewQuery("SELECT text FROM posts WHERE title = ?")
-	if err := db.Do(query, "New Title").Scan(&text); err != nil {
+	// list all tweets
+	iter := session.Query(`SELECT id, text FROM tweet WHERE timeline = ?`, "me").Iter()
+	for iter.Scan(&id, &text) {
+		fmt.Println("Tweet:", id, text)
+	}
+	if err := iter.Close(); err != nil {
 		log.Fatal(err)
 	}
-	fmt.Println(text)
 }
 ```
 

cluster.go

@@ -11,26 +11,26 @@ import (
 	"time"
 )
 
-// Cluster sets up and maintains the node configuration of a Cassandra
-// cluster.
-//
-// It has a varity of attributes that can be used to modify the behavior
-// to fit the most common use cases. Applications that requre a different
-// a setup should compose the nodes on their own.
+// ClusterConfig is a struct to configure the default cluster implementation
+// of gocoql. It has a varity of attributes that can be used to modify the
+// behavior to fit the most common use cases. Applications that requre a
+// different setup must implement their own cluster.
 type ClusterConfig struct {
-	Hosts        []string
-	CQLVersion   string
-	ProtoVersion int
-	Timeout      time.Duration
-	DefaultPort  int
-	Keyspace     string
-	NumConns     int
-	NumStreams   int
-	DelayMin     time.Duration
-	DelayMax     time.Duration
-	StartupMin   int
+	Hosts        []string      // addresses for the initial connections
+	CQLVersion   string        // CQL version (default: 3.0.0)
+	ProtoVersion int           // version of the native protocol (default: 2)
+	Timeout      time.Duration // connection timeout (default: 200ms)
+	DefaultPort  int           // default port (default: 9042)
+	Keyspace     string        // initial keyspace (optional)
+	NumConns     int           // number of connections per host (default: 2)
+	NumStreams   int           // number of streams per connection (default: 128)
+	DelayMin     time.Duration // minimum reconnection delay (default: 1s)
+	DelayMax     time.Duration // maximum reconnection delay (default: 10min)
+	StartupMin   int           // wait for StartupMin hosts (default: len(Hosts)/2+1)
+	Consistency  Consistency   // default consistency level (default: Quorum)
 }
 
+// NewCluster generates a new config for the default cluster implementation.
 func NewCluster(hosts ...string) *ClusterConfig {
 	cfg := &ClusterConfig{
 		Hosts:        hosts,
@@ -43,18 +43,30 @@ func NewCluster(hosts ...string) *ClusterConfig {
 		DelayMin:     1 * time.Second,
 		DelayMax:     10 * time.Minute,
 		StartupMin:   len(hosts)/2 + 1,
+		Consistency:  Quorum,
 	}
 	return cfg
 }
 
+// CreateSession initializes the cluster based on this config and returns a
+// session object that can be used to interact with the database.
 func (cfg *ClusterConfig) CreateSession() *Session {
 	impl := &clusterImpl{
 		cfg:      *cfg,
 		hostPool: NewRoundRobin(),
 		connPool: make(map[string]*RoundRobin),
+		conns:    make(map[*Conn]struct{}),
 	}
 	impl.wgStart.Add(1)
-	impl.startup()
+	for i := 0; i < len(impl.cfg.Hosts); i++ {
+		addr := strings.TrimSpace(impl.cfg.Hosts[i])
+		if strings.IndexByte(addr, ':') < 0 {
+			addr = fmt.Sprintf("%s:%d", addr, impl.cfg.DefaultPort)
+		}
+		for j := 0; j < impl.cfg.NumConns; j++ {
+			go impl.connect(addr)
+		}
+	}
 	impl.wgStart.Wait()
 	return NewSession(impl)
 }
@@ -63,30 +75,16 @@ type clusterImpl struct {
 	cfg      ClusterConfig
 	hostPool *RoundRobin
 	connPool map[string]*RoundRobin
-	mu       sync.RWMutex
-
-	conns []*Conn
+	conns    map[*Conn]struct{}
+	keyspace string
+	mu       sync.Mutex
 
 	started bool
 	wgStart sync.WaitGroup
 
 	quit     bool
 	quitWait chan bool
 	quitOnce sync.Once
-
-	keyspace string
-}
-
-func (c *clusterImpl) startup() {
-	for i := 0; i < len(c.cfg.Hosts); i++ {
-		addr := strings.TrimSpace(c.cfg.Hosts[i])
-		if strings.IndexByte(addr, ':') < 0 {
-			addr = fmt.Sprintf("%s:%d", addr, c.cfg.DefaultPort)
-		}
-		for j := 0; j < c.cfg.NumConns; j++ {
-			go c.connect(addr)
-		}
-	}
 }
 
 func (c *clusterImpl) connect(addr string) {
@@ -136,6 +134,7 @@ func (c *clusterImpl) addConn(conn *Conn, keyspace string) {
 		return
 	}
 	if keyspace != c.keyspace && c.keyspace != "" {
+		// change the keyspace before adding the node to the pool
 		go c.changeKeyspace(conn, c.keyspace, false)
 		return
 	}
@@ -150,7 +149,7 @@ func (c *clusterImpl) addConn(conn *Conn, keyspace string) {
 		}
 	}
 	connPool.AddNode(conn)
-	c.conns = append(c.conns, conn)
+	c.conns[conn] = struct{}{}
 }
 
 func (c *clusterImpl) removeConn(conn *Conn) {
@@ -165,21 +164,16 @@ func (c *clusterImpl) removeConn(conn *Conn) {
 	if connPool.Size() == 0 {
 		c.hostPool.RemoveNode(connPool)
 	}
-	for i := 0; i < len(c.conns); i++ {
-		if c.conns[i] == conn {
-			last := len(c.conns) - 1
-			c.conns[i], c.conns[last] = c.conns[last], c.conns[i]
-			c.conns = c.conns[:last]
-		}
-	}
+	delete(c.conns, conn)
 }
 
 func (c *clusterImpl) HandleError(conn *Conn, err error, closed bool) {
 	if !closed {
+		// ignore all non-fatal errors
 		return
 	}
 	c.removeConn(conn)
-	go c.connect(conn.Address())
+	go c.connect(conn.Address()) // reconnect
 }
 
 func (c *clusterImpl) HandleKeyspace(conn *Conn, keyspace string) {
@@ -189,10 +183,13 @@ func (c *clusterImpl) HandleKeyspace(conn *Conn, keyspace string) {
 		return
 	}
 	c.keyspace = keyspace
-	conns := make([]*Conn, len(c.conns))
-	copy(conns, c.conns)
+	conns := make([]*Conn, 0, len(c.conns))
+	for conn := range c.conns {
+		conns = append(conns, conn)
+	}
 	c.mu.Unlock()
 
+	// change the keyspace of all other connections too
 	for i := 0; i < len(conns); i++ {
 		if conns[i] == conn {
 			continue
@@ -202,10 +199,16 @@ func (c *clusterImpl) HandleKeyspace(conn *Conn, keyspace string) {
 }
 
 func (c *clusterImpl) ExecuteQuery(qry *Query) (*Iter, error) {
+	if qry.Cons == 0 {
+		qry.Cons = c.cfg.Consistency
+	}
 	return c.hostPool.ExecuteQuery(qry)
 }
 
 func (c *clusterImpl) ExecuteBatch(batch *Batch) error {
+	if batch.Cons == 0 {
+		batch.Cons = c.cfg.Consistency
+	}
 	return c.hostPool.ExecuteBatch(batch)
 }
 
@@ -215,8 +218,8 @@ func (c *clusterImpl) Close() {
 		defer c.mu.Unlock()
 		c.quit = true
 		close(c.quitWait)
-		for i := 0; i < len(c.conns); i++ {
-			c.conns[i].Close()
+		for conn := range c.conns {
+			conn.Close()
 		}
 	})
 }

conn.go

@@ -26,7 +26,6 @@ type Cluster interface {
 type ConnConfig struct {
 	ProtoVersion int
 	CQLVersion   string
-	Keyspace     string
 	Timeout      time.Duration
 	NumStreams   int
 }

gocql_test/main.go

@@ -126,7 +126,7 @@ func getPage(title string, revid uuid.UUID) (*Page, error) {
 	p := new(Page)
 	err := session.Query(`SELECT title, revid, body, views, protected, modified,
 			tags, attachments
-			FROM page WHERE title = ? AND revid = ?`, title, revid).Scan(
+			FROM page WHERE title = ? AND revid = ? LIMIT 1`, title, revid).Scan(
 		&p.Title, &p.RevId, &p.Body, &p.Views, &p.Protected, &p.Modified,
 		&p.Tags, &p.Attachments)
 	return p, err
@@ -180,3 +180,42 @@ func main() {
 	}
 
 }
+
+func main2() {
+	// connect to the cluster
+	cluster := gocql.NewCluster("192.168.1.1", "192.168.1.2", "192.168.1.3")
+	cluster.Keyspace = "example"
+	cluster.Consistency = gocql.Quorum
+	session := cluster.CreateSession()
+	defer session.Close()
+
+	// insert a tweet
+	if err := session.Query(`INSERT INTO tweet
+		(timeline, id, text) VALUES (?, ?, ?)`,
+		"me", uuid.TimeUUID(), "hello world").Exec(); err != nil {
+		log.Fatal("insert tweet: ", err)
+	}
+
+	var id uuid.UUID
+	var text string
+
+	// select a single tweet
+	if err := session.Query(`SELECT id, text
+		FROM tweet
+		WHERE timeline = ?
+		LIMIT 1`,
+		"me").Consistency(gocql.One).Scan(&id, &text); err != nil {
+		log.Fatal("get tweet: ", err)
+	}
+	fmt.Println("Tweet:", id, text)
+
+	// list all tweets
+	iter := session.Query(`SELECT id, text FROM tweet
+		WHERE timeline = ?`, "me").Iter()
+	for iter.Scan(&id, &text) {
+		fmt.Println("Tweet:", id, text)
+	}
+	if err := iter.Close(); err != nil {
+		log.Fatal("list tweets:", err)
+	}
+}

session.go

@@ -79,10 +79,14 @@ type QueryBuilder struct {
 	ctx Node
 }
 
+// Args specifies the query parameters.
 func (b QueryBuilder) Args(args ...interface{}) {
 	b.qry.Args = args
 }
 
+// Consistency sets the consistency level for this query. If no consistency
+// level have been set, the default consistency level of the cluster
+// is used.
 func (b QueryBuilder) Consistency(cons Consistency) QueryBuilder {
 	b.qry.Cons = cons
 	return b

topology.go

@@ -51,14 +51,6 @@ func (r *RoundRobin) Size() int {
 	return n
 }
 
-func (r *RoundRobin) GetPool() []Node {
-	r.mu.RLock()
-	pool := make([]Node, len(r.pool))
-	copy(pool, r.pool)
-	r.mu.RUnlock()
-	return pool
-}
-
 func (r *RoundRobin) ExecuteQuery(qry *Query) (*Iter, error) {
 	node := r.pick()
 	if node == nil {