Merge branch 'hotfix/v1.1.4'

CHANGELOG.md

@@ -2,6 +2,20 @@
 All notable changes to this project will be documented in this file.
 This project adheres to [Semantic Versioning](http://semver.org/).
 
+## v1.1.4
+### Added
+ - Added root table terms (`r.TableCreate`, `r.TableList` and `r.TableDrop`)
+
+### Removed
+ - Removed `ReadMode` option from `RunOpts` and `ExecOpts` (incorrectly added in v1.1.0)
+
+### Fixed 
+ - Fixed `Decode` no longer setting pointer to nil on document not found
+ - Fixed panic when `fetchMore` returns an error
+ - Fixed deadlock when closing changefeed
+ - Fixed stop query incorrectly waiting for response
+ - Fixed pointers not to be properly decoded
+
 ## v1.1.3
 ### Fixed
  - Fixed pointers not to be properly decoded

README.md

@@ -8,7 +8,7 @@
 
 ![GoRethink Logo](https://raw.github.com/wiki/dancannon/gorethink/gopher-and-thinker-s.png "Golang Gopher and RethinkDB Thinker")
 
-Current version: v1.1.3 (RethinkDB v2.1)
+Current version: v1.1.4 (RethinkDB v2.1)
 
 Please note that this version of the driver only supports versions of RethinkDB using the v0.4 protocol (any versions of the driver older than RethinkDB 2.0 will not work).
 

cursor.go

@@ -119,6 +119,9 @@ func (c *Cursor) Close() error {
 		q := Query{
 			Type:  p.Query_STOP,
 			Token: c.token,
+			Opts: map[string]interface{}{
+				"noreply": true,
+			},
 		}
 
 		_, _, err = conn.Query(q)
@@ -184,10 +187,10 @@ func (c *Cursor) loadNextLocked(dest interface{}) (bool, error) {
 		if c.buffer.Len() == 0 && c.responses.Len() == 0 && !c.finished {
 			c.mu.Unlock()
 			err := c.fetchMore()
+			c.mu.Lock()
 			if err != nil {
 				return false, err
 			}
-			c.mu.Lock()
 		}
 
 		if c.buffer.Len() == 0 && c.responses.Len() == 0 && c.finished {

cursor_test.go

@@ -218,6 +218,13 @@ func (s *RethinkSuite) TestEmptyResults(c *test.C) {
 	res, err = DB("test").Table("test").Filter(obj).Run(session)
 	c.Assert(err, test.IsNil)
 	c.Assert(res.IsNil(), test.Equals, true)
+
+	var objP *object
+
+	res, err = DB("test").Table("test").Get("missing value").Run(session)
+	res.Next(&objP)
+	c.Assert(err, test.IsNil)
+	c.Assert(objP, test.IsNil)
 }
 
 func (s *RethinkSuite) TestCursorAll(c *test.C) {
@@ -338,6 +345,22 @@ func (s *RethinkSuite) TestCursorListen(c *test.C) {
 	})
 }
 
+func (s *RethinkSuite) TestCursorChangesClose(c *test.C) {
+	// Ensure table + database exist
+	DBCreate("test").Exec(session)
+	DB("test").TableDrop("Table3").Exec(session)
+	DB("test").TableCreate("Table3").Exec(session)
+
+	// Test query
+	// res, err := DB("test").Table("Table3").Changes().Run(session)
+	res, err := DB("test").Table("Table3").Changes().Run(session)
+	c.Assert(err, test.IsNil)
+
+	// Ensure that the cursor can be closed
+	err = res.Close()
+	c.Assert(err, test.IsNil)
+}
+
 func (s *RethinkSuite) TestCursorReuseResult(c *test.C) {
 	// Test query
 	query := Expr([]interface{}{

doc.go

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

encoding/decoder.go

@@ -52,11 +52,6 @@ func Decode(dst interface{}, src interface{}) (err error) {
 
 // decode decodes the source value into the destination value
 func decode(dv, sv reflect.Value) {
-	if dv.IsValid() {
-		dv = indirect(dv, false)
-		dv.Set(reflect.Zero(dv.Type()))
-	}
-
 	valueDecoder(dv, sv)(dv, sv)
 }
 
@@ -74,6 +69,11 @@ func valueDecoder(dv, sv reflect.Value) decoderFunc {
 		return invalidValueDecoder
 	}
 
+	if dv.IsValid() {
+		dv = indirect(dv, false)
+		dv.Set(reflect.Zero(dv.Type()))
+	}
+
 	return typeDecoder(dv.Type(), sv.Type())
 }
 

encoding/decoder_types.go

@@ -186,6 +186,9 @@ func interfaceDecoder(dv, sv reflect.Value) {
 
 func interfaceAsTypeDecoder(dv, sv reflect.Value) {
 	if !sv.IsNil() {
+		dv = indirect(dv, false)
+		dv.Set(reflect.Zero(dv.Type()))
+
 		decode(dv, sv.Elem())
 	}
 }

encoding/encoder_test.go

@@ -96,7 +96,7 @@ var optionalsExpected = map[string]interface{}{
 func TestOmitEmpty(t *testing.T) {
 	var o Optionals
 	o.Sw = "something"
-	o.Tr = time.Unix(0, 0)
+	o.Tr = time.Unix(0, 0).In(time.UTC)
 	o.Mr = map[string]interface{}{}
 	o.Mo = map[string]interface{}{}
 

gorethink_test.go

@@ -16,6 +16,7 @@ import (
 
 var session *Session
 var debug = flag.Bool("gorethink.debug", false, "print query trees")
+var testdata = flag.Bool("gorethink.testdata", true, "create test data")
 var url, url1, url2, url3, db, authKey string
 
 func init() {

query.go

@@ -196,7 +196,6 @@ type RunOpts struct {
 	DB             interface{} `gorethink:"db,omitempty"`
 	Db             interface{} `gorethink:"db,omitempty"` // Deprecated
 	Profile        interface{} `gorethink:"profile,omitempty"`
-	ReadMode       interface{} `gorethink:"read_mode,omitempty"`
 	UseOutdated    interface{} `gorethink:"use_outdated,omitempty"` // Deprecated
 	ArrayLimit     interface{} `gorethink:"array_limit,omitempty"`
 	TimeFormat     interface{} `gorethink:"time_format,omitempty"`
@@ -280,7 +279,6 @@ type ExecOpts struct {
 	DB             interface{} `gorethink:"db,omitempty"`
 	Db             interface{} `gorethink:"db,omitempty"` // Deprecated
 	Profile        interface{} `gorethink:"profile,omitempty"`
-	ReadMode       interface{} `gorethink:"read_mode,omitempty"`
 	UseOutdated    interface{} `gorethink:"use_outdated,omitempty"` // Deprecated
 	ArrayLimit     interface{} `gorethink:"array_limit,omitempty"`
 	TimeFormat     interface{} `gorethink:"time_format,omitempty"`

query_table.go

@@ -17,6 +17,18 @@ func (o *TableCreateOpts) toMap() map[string]interface{} {
 	return optArgsToMap(o)
 }
 
+// TableCreate creates a table. A RethinkDB table is a collection of JSON
+// documents.
+//
+// Note: Only alphanumeric characters and underscores are valid for the table name.
+func TableCreate(name interface{}, optArgs ...TableCreateOpts) Term {
+	opts := map[string]interface{}{}
+	if len(optArgs) >= 1 {
+		opts = optArgs[0].toMap()
+	}
+	return constructRootTerm("TableCreate", p.Term_TABLE_CREATE, []interface{}{name}, opts)
+}
+
 // TableCreate creates a table. A RethinkDB table is a collection of JSON
 // documents.
 //
@@ -29,11 +41,21 @@ func (t Term) TableCreate(name interface{}, optArgs ...TableCreateOpts) Term {
 	return constructMethodTerm(t, "TableCreate", p.Term_TABLE_CREATE, []interface{}{name}, opts)
 }
 
+// TableDrop deletes a table. The table and all its data will be deleted.
+func TableDrop(args ...interface{}) Term {
+	return constructRootTerm("TableDrop", p.Term_TABLE_DROP, args, map[string]interface{}{})
+}
+
 // TableDrop deletes a table. The table and all its data will be deleted.
 func (t Term) TableDrop(args ...interface{}) Term {
 	return constructMethodTerm(t, "TableDrop", p.Term_TABLE_DROP, args, map[string]interface{}{})
 }
 
+// TableList lists all table names in a database.
+func TableList(args ...interface{}) Term {
+	return constructRootTerm("TableList", p.Term_TABLE_LIST, args, map[string]interface{}{})
+}
+
 // TableList lists all table names in a database.
 func (t Term) TableList(args ...interface{}) Term {
 	return constructMethodTerm(t, "TableList", p.Term_TABLE_LIST, args, map[string]interface{}{})

query_table_test.go

@@ -18,6 +18,20 @@ func (s *RethinkSuite) TestTableCreate(c *test.C) {
 	c.Assert(response.TablesCreated, jsonEquals, 1)
 }
 
+func (s *RethinkSuite) TestTableCreateSessionDatabase(c *test.C) {
+	session, err := Connect(ConnectOpts{
+		Address: url,
+		AuthKey: authKey,
+	})
+	c.Assert(err, test.IsNil)
+	TableDrop("test").Exec(session)
+
+	// Test database creation
+	response, err := TableCreate("test").RunWrite(session)
+	c.Assert(err, test.IsNil)
+	c.Assert(response.TablesCreated, jsonEquals, 1)
+}
+
 func (s *RethinkSuite) TestTableCreatePrimaryKey(c *test.C) {
 	DB("test").TableDrop("testOpts").Exec(session)
 
@@ -250,9 +264,41 @@ func (s *RethinkSuite) TestTableChangesExit(c *test.C) {
 	for _ = range change {
 		n++
 	}
-	if res.Err() != nil {
-		c.Fatal(res.Err())
+	if res.Err() == nil {
+		c.Fatal("No error returned, expected connection closed")
 	}
 
 	c.Assert(n, test.Equals, 5)
 }
+
+func (s *RethinkSuite) TestTableChangesExitNoResults(c *test.C) {
+	DB("test").TableDrop("changes").Exec(session)
+	DB("test").TableCreate("changes").Exec(session)
+
+	var n int
+
+	res, err := DB("test").Table("changes").Changes().Run(session)
+	if err != nil {
+		c.Fatal(err.Error())
+	}
+	c.Assert(res.Type(), test.Equals, "Feed")
+
+	change := make(chan ChangeResponse)
+
+	// Close cursor after one second
+	go func() {
+		<-time.After(time.Second)
+		res.Close()
+	}()
+
+	// Listen for changes
+	res.Listen(change)
+	for _ = range change {
+		n++
+	}
+	if res.Err() == nil {
+		c.Fatal("No error returned, expected connection closed")
+	}
+
+	c.Assert(n, test.Equals, 0)
+}