Merge tag 'v0.7.0' into develop

Released v0.7.0

.travis.yml

@@ -1,8 +1,10 @@
 language: go
 
+gobuild_args: -tags='cluster'
+
 go:
   - 1.4
-  # - tip
+  - tip
 
 before_install:
   - wget http://download.rethinkdb.com/dev/2.0.0-0RC1/rethinkdb_2.0.0%2b0RC1~0precise_amd64.deb
@@ -12,4 +14,7 @@ before_script:
 #   - sudo add-apt-repository ppa:rethinkdb/ppa -y
 #   - sudo apt-get update -q
 #   - sudo apt-get install rethinkdb
-  - rethinkdb --bind all &
+  - rethinkdb > /dev/null 2>&1 &
+  - rethinkdb --port-offset 1 --directory rethinkdb_data1 --join localhost:29016 > /dev/null 2>&1 &
+  - rethinkdb --port-offset 2 --directory rethinkdb_data2 --join localhost:29016 > /dev/null 2>&1 &
+  - rethinkdb --port-offset 3 --directory rethinkdb_data3 --join localhost:29016 > /dev/null 2>&1 &

CHANGELOG.md

@@ -2,6 +2,41 @@
 All notable changes to this project will be documented in this file.
 This project adheres to [Semantic Versioning](http://semver.org/).
 
+## v0.7.0 - 2015-03-30
+
+This release includes support for RethinkDB 2.0 and connecting to clusters. To connect to a cluster you should use the new `Addresses` field in `ConnectOpts`, for example:
+
+```go
+session, err := r.Connect(r.ConnectOpts{
+    Addresses: []string{"localhost:28015", "localhost:28016"},
+})
+if err != nil {
+    log.Fatalln(err.Error())
+}
+```
+
+Also added was the ability to read from a cursor using a channel, this is especially useful when using changefeeds. For more information see this [gist](https://gist.github.com/dancannon/2865686d163ed78bbc3c)
+
+```go
+cursor, err := r.Table("items").Changes()
+ch := make(chan map[string]interface{})
+cursor.Listen(ch)
+```
+
+For more details checkout the [README](https://github.com/dancannon/gorethink/blob/master/README.md) and [godoc](https://godoc.org/github.com/dancannon/gorethink). As always if you have any further questions send me a message on [Gitter](https://gitter.im/dancannon/gorethink).
+
+- Added the ability to connect to multiple nodes, queries are then distributed between these nodes. If a node stops responding then queries stop being sent to this node.
+- Added the `DiscoverHosts` optional argument to `ConnectOpts`, when this value is `true` the driver will listen for new nodes added to the cluster.
+- Added the `Addresses` optional argument to `ConnectOpts`, this allows the driver to connect to multiple nodes in a cluster.
+- Added the `IncludeStates` optional argument to `Changes`.
+- Added `MinVal` and `MaxVal` which represent the smallest and largest possible values.
+- Added the `Listen` cursor helper function which publishes database results to a channel.
+- Added support for optional  arguments for the `Wait` function.
+- Added the `Type` function to the `Cursor`, by default this value will be "Cursor" unless using a changefeed.
+- Changed the `IndexesOf` function to `OffsetsOf` .
+- Changed driver to use the v0.4 protocol (used to use v0.3).
+- Fixed geometry tests not properly checking the expected results.
+- Fixed bug causing nil pointer panics when using an `Unmarshaler`
 
 ## v0.6.3 - 2015-03-04
 ### Added

README.md

@@ -2,14 +2,12 @@
 
 [![GitHub tag](https://img.shields.io/github/tag/dancannon/gorethink.svg?style=flat)]()
 [![GoDoc](https://godoc.org/github.com/dancannon/gorethink?status.png)](https://godoc.org/github.com/dancannon/gorethink)
-[![wercker status](https://app.wercker.com/status/e315e764041af8e80f0c68280d4b4de2/s/master "wercker status")](https://app.wercker.com/project/bykey/e315e764041af8e80f0c68280d4b4de2) 
+[![build status](https://img.shields.io/travis/dancannon/gorethink/master.svg "build status")](https://travis-ci.org/dancannon/gorethink) 
 
 [Go](http://golang.org/) driver for [RethinkDB](http://www.rethinkdb.com/) 
 
 
-Current version: v0.6.3 (RethinkDB v1.16) 
-
-**Version 0.6 introduced some small API changes and some significant internal changes, for more information check the [change log](CHANGELOG.md) and please be aware the driver is not yet stable**
+Current version: v0.7.0 (RethinkDB v2.0) 
 
 [![Gitter](https://badges.gitter.im/Join%20Chat.svg)](https://gitter.im/dancannon/gorethink?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge)
 
@@ -47,25 +45,40 @@ The driver uses a connection pool at all times, by default it creates and frees
 To configure the connection pool `MaxIdle`, `MaxOpen` and `IdleTimeout` can be specified during connection. If you wish to change the value of `MaxIdle` or `MaxOpen` during runtime then the functions `SetMaxIdleConns` and `SetMaxOpenConns` can be used.
 
 ```go
-import (
-    r "github.com/dancannon/gorethink"
-)
-
 var session *r.Session
 
-session, err := r.ConnectWithOpts(r.ConnectOpts{
+session, err := r.Connect(r.ConnectOpts{
+    Address: "localhost:28015",
     Database: "test",
     MaxIdle: 10,
     MaxOpen: 10,
-}, "localhost:28015")
+})
 if err != nil {
     log.Fatalln(err.Error())
 }
 
 session.SetMaxOpenConns(5)
 ```
 
-A pre-configured [Pool](http://godoc.org/github.com/dancannon/gorethink#Pool) instance can also be passed to Connect().
+### Connect to a cluster
+
+To connect to a RethinkDB cluster which has multiple nodes you can use the following syntax. 
+
+```go
+var session *r.Session
+
+session, err := r.Conenct(r.ConnectOpts{
+    Addresses: []string{"localhost:28015", "localhost:28016"},
+    Database: "test",
+    AuthKey:  "14daak1cad13dj",
+    DiscoverHosts: true,
+}, "localhost:28015")
+if err != nil {
+    log.Fatalln(err.Error())
+}
+```
+
+When `DiscoverHosts` is true any nodes are added to the cluster after the initial connection then the new node will be added to the pool of available nodes used by GoRethink.
 
 
 ## Query Functions

cluster.go

@@ -160,7 +160,7 @@ func (c *Cluster) listenForNodeChanges() error {
 			b.MaxElapsedTime = time.Second * 5
 
 			backoff.Retry(func() error {
-				node, err := c.connectNode(result.NewVal)
+				node, err := c.connectNodeWithStatus(result.NewVal)
 				if err == nil {
 					if !c.nodeExists(node) {
 						c.addNode(node)
@@ -204,14 +204,27 @@ func (c *Cluster) connectNodes(hosts []Host) {
 			continue
 		}
 
-		var results []nodeStatus
-		err = cursor.All(&results)
-		if err != nil {
-			continue
-		}
+		if c.opts.DiscoverHosts {
+			var results []nodeStatus
+			err = cursor.All(&results)
+			if err != nil {
+				continue
+			}
 
-		for _, result := range results {
-			node, err := c.connectNode(result)
+			for _, result := range results {
+				node, err := c.connectNodeWithStatus(result)
+				if err == nil {
+					if _, ok := nodeSet[node.ID]; !ok {
+						log.WithFields(logrus.Fields{
+							"id":   node.ID,
+							"host": node.Host.String(),
+						}).Debug("Connected to node")
+						nodeSet[node.ID] = node
+					}
+				}
+			}
+		} else {
+			node, err := c.connectNode(host.String(), []Host{host})
 			if err == nil {
 				if _, ok := nodeSet[node.ID]; !ok {
 					log.WithFields(logrus.Fields{
@@ -232,13 +245,16 @@ func (c *Cluster) connectNodes(hosts []Host) {
 	c.setNodes(nodes)
 }
 
-func (c *Cluster) connectNode(s nodeStatus) (*Node, error) {
+func (c *Cluster) connectNodeWithStatus(s nodeStatus) (*Node, error) {
 	aliases := make([]Host, len(s.Network.CanonicalAddresses))
 	for i, aliasAddress := range s.Network.CanonicalAddresses {
 		aliases[i] = NewHost(aliasAddress.Host, int(s.Network.ReqlPort))
 	}
 
-	// Keep trying to connect to one of the host aliases
+	return c.connectNode(s.ID, aliases)
+}
+
+func (c *Cluster) connectNode(id string, aliases []Host) (*Node, error) {
 	var pool *Pool
 	var err error
 
@@ -266,7 +282,7 @@ func (c *Cluster) connectNode(s nodeStatus) (*Node, error) {
 		return nil, ErrInvalidNode
 	}
 
-	return newNode(s.ID, aliases, c, pool), nil
+	return newNode(id, aliases, c, pool), nil
 }
 
 // IsConnected returns true if cluster has nodes and is not already closed.

cluster_integration_test.go

@@ -10,7 +10,7 @@ import (
 )
 
 func (s *RethinkSuite) TestClusterDetectNewNode(c *test.C) {
-	session, err := ConnectClusterWithOpts(ConnectOpts{
+	session, err := Connect(ConnectOpts{
 		DiscoverHosts:       true,
 		NodeRefreshInterval: time.Second,
 	}, url, url2)
@@ -31,30 +31,8 @@ func (s *RethinkSuite) TestClusterDetectNewNode(c *test.C) {
 	}
 }
 
-func (s *RethinkSuite) TestClusterDetectRemovedNode(c *test.C) {
-	session, err := ConnectClusterWithOpts(ConnectOpts{
-		DiscoverHosts:       true,
-		NodeRefreshInterval: time.Second,
-	}, url, url2, url3)
-	c.Assert(err, test.IsNil)
-
-	t := time.NewTimer(time.Minute * 5)
-	for {
-		select {
-		// Fail if deadline has passed
-		case <-t.C:
-			c.Fatal("No node was removed from the cluster")
-		default:
-			// Pass if another node was added
-			if len(session.cluster.GetNodes()) < 3 {
-				return
-			}
-		}
-	}
-}
-
 func (s *RethinkSuite) TestClusterNodeHealth(c *test.C) {
-	session, err := ConnectClusterWithOpts(ConnectOpts{
+	session, err := Connect(ConnectOpts{
 		DiscoverHosts:       true,
 		NodeRefreshInterval: time.Second,
 		MaxIdle:             50,

cluster_test.go

@@ -10,10 +10,12 @@ import (
 )
 
 func (s *RethinkSuite) TestClusterConnect(c *test.C) {
-	cluster, err := ConnectCluster(url, url2, url3)
+	session, err := Connect(ConnectOpts{
+		Addresses: []string{url1, url2, url3},
+	})
 	c.Assert(err, test.IsNil)
 
-	row, err := Expr("Hello World").Run(cluster)
+	row, err := Expr("Hello World").Run(session)
 	c.Assert(err, test.IsNil)
 
 	var response string
@@ -23,11 +25,13 @@ func (s *RethinkSuite) TestClusterConnect(c *test.C) {
 }
 
 func (s *RethinkSuite) TestClusterMultipleQueries(c *test.C) {
-	cluster, err := ConnectCluster(url, url2, url3)
+	session, err := Connect(ConnectOpts{
+		Addresses: []string{url1, url2, url3},
+	})
 	c.Assert(err, test.IsNil)
 
 	for i := 0; i < 1000; i++ {
-		row, err := Expr(fmt.Sprintf("Hello World", i)).Run(cluster)
+		row, err := Expr(fmt.Sprintf("Hello World", i)).Run(session)
 		c.Assert(err, test.IsNil)
 
 		var response string
@@ -39,16 +43,18 @@ func (s *RethinkSuite) TestClusterMultipleQueries(c *test.C) {
 
 func (s *RethinkSuite) TestClusterConnectError(c *test.C) {
 	var err error
-	_, err = ConnectClusterWithOpts(ConnectOpts{
-		Timeout: time.Second,
-	}, "nonexistanturl")
+	_, err = Connect(ConnectOpts{
+		Addresses: []string{"nonexistanturl"},
+		Timeout:   time.Second,
+	})
 	c.Assert(err, test.NotNil)
 }
 
 func (s *RethinkSuite) TestClusterConnectDatabase(c *test.C) {
-	session, err := ConnectClusterWithOpts(ConnectOpts{
-		Database: "test2",
-	}, url, url2, url3)
+	session, err := Connect(ConnectOpts{
+		Addresses: []string{url1, url2, url3},
+		Database:  "test2",
+	})
 	c.Assert(err, test.IsNil)
 
 	_, err = Table("test2").Run(session)

cursor.go

@@ -412,6 +412,10 @@ type queue struct {
 }
 
 func (q *queue) Len() int {
+	if len(q.elems) == 0 {
+		return 0
+	}
+
 	return q.nelems
 }
 func (q *queue) Push(elem interface{}) {

doc.go

@@ -1,6 +1,6 @@
-// Go driver for RethinkDB
+// Package gorethink implements a Go driver for RethinkDB
 //
-// Current version: v0.6.3 (RethinkDB v1.16)
+// Current version: v0.7.0 (RethinkDB v2.0)
 // For more in depth information on how to use RethinkDB check out the API docs
 // at http://rethinkdb.com/api
 package gorethink

example_query_select_test.go

@@ -15,9 +15,10 @@ func Example_Get() {
 		Gender    string `gorethink:"gender"`
 	}
 
-	sess, err := r.ConnectWithOpts(r.ConnectOpts{
+	sess, err := r.Connect(r.ConnectOpts{
+		Address: url,
 		AuthKey: authKey,
-	}, url)
+	})
 	if err != nil {
 		log.Fatalf("Error connecting to DB: %s", err)
 	}
@@ -57,9 +58,10 @@ func Example_GetAll_Compound() {
 		Gender    string `gorethink:"gender"`
 	}
 
-	sess, err := r.ConnectWithOpts(r.ConnectOpts{
+	sess, err := r.Connect(r.ConnectOpts{
+		Address: url,
 		AuthKey: authKey,
-	}, url)
+	})
 	if err != nil {
 		log.Fatalf("Error connecting to DB: %s", err)
 	}

example_query_table_test.go

@@ -8,9 +8,10 @@ import (
 )
 
 func Example_TableCreate() {
-	sess, err := r.ConnectWithOpts(r.ConnectOpts{
+	sess, err := r.Connect(r.ConnectOpts{
+		Address: url,
 		AuthKey: authKey,
-	}, url)
+	})
 	if err != nil {
 		log.Fatalf("Error connecting to DB: %s", err)
 	}
@@ -30,9 +31,10 @@ func Example_TableCreate() {
 }
 
 func Example_IndexCreate() {
-	sess, err := r.ConnectWithOpts(r.ConnectOpts{
+	sess, err := r.Connect(r.ConnectOpts{
+		Address: url,
 		AuthKey: authKey,
-	}, url)
+	})
 	if err != nil {
 		log.Fatalf("Error connecting to DB: %s", err)
 	}
@@ -53,9 +55,10 @@ func Example_IndexCreate() {
 }
 
 func Example_IndexCreate_compound() {
-	sess, err := r.ConnectWithOpts(r.ConnectOpts{
+	sess, err := r.Connect(r.ConnectOpts{
+		Address: url,
 		AuthKey: authKey,
-	}, url)
+	})
 	if err != nil {
 		log.Fatalf("Error connecting to DB: %s", err)
 	}

example_test.go

@@ -23,9 +23,10 @@ func init() {
 }
 
 func Example() {
-	session, err := r.ConnectWithOpts(r.ConnectOpts{
+	session, err := r.Connect(r.ConnectOpts{
+		Address: url,
 		AuthKey: authKey,
-	}, url)
+	})
 	if err != nil {
 		log.Fatalf("Error connecting to DB: %s", err)
 	}