Skip to content

Commit

Permalink
Merge branch 'release/v0.4.0'
Browse files Browse the repository at this point in the history
  • Loading branch information
Daniel Cannon committed Aug 31, 2014
2 parents 6759a9c + 6b03a6b commit 36ef8b1
Show file tree
Hide file tree
Showing 25 changed files with 1,344 additions and 981 deletions.
19 changes: 19 additions & 0 deletions CHANGELOG.md
View file
Original file line number Diff line number Diff line change
@@ -1,5 +1,24 @@
# Changelog

## v0.4.0 - 13 Aug 2014

- Updated the driver to support RethinkDB v1.14 (#116)
- Added the Binary data type
- Added the Binary command which takes a `[]byte`, `io.Reader` or `bytes.Buffer{}` as an argument.
- Added the `BinaryFormat` optional argument to `RunOpts`
- Added the `GroupFormat` optional argument to `RunOpts`
- Added the `ArrayLimit` optional argument to `RunOpts`
- Renamed the `ReturnVals` optional argument to `ReturnChanges`
- Renamed the `Upsert` optional argument to `Conflict`
- Added the `IndexRename` command
- Updated `Distinct` to now take the `Index` optional argument (using `DistinctOpts`)

### Internal Changes

- Updated to use the new JSON protocol
- Switched the connection pool code to use github.com/fatih/pool
- Added some benchmarks

## v0.3.2 - 17 Aug 2014

- Fixed issue causing connections not to be closed correctly (#109)
Expand Down
2 changes: 1 addition & 1 deletion README.md
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ GoRethink - RethinkDB Driver for Go [![wercker status](https://app.wercker.com/s
[Go](http://golang.org/) driver for [RethinkDB](http://www.rethinkdb.com/) made by [Daniel Cannon](http://github.com/dancannon) and based off of Christopher Hesse's [RethinkGo](https://github.com/christopherhesse/rethinkgo) driver.


Current version: v0.3.2 (RethinkDB v1.13)
Current version: v0.4.0 (RethinkDB v1.14)

**Version 0.3 introduced some API changes, for more information check the [change log](CHANGELOG.md)**

Expand Down
207 changes: 109 additions & 98 deletions connection.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,19 +3,29 @@ package gorethink
import (
"bufio"
"encoding/binary"
"encoding/json"
"fmt"
"io"
"net"
"sync"
"time"

"code.google.com/p/goprotobuf/proto"
"gopkg.in/fatih/pool.v1"

p "github.com/dancannon/gorethink/ql2"
)

type Response struct {
Token int64
Type p.Response_ResponseType `json:"t"`
Responses []interface{} `json:"r"`
Backtrace []interface{} `json:"b"`
Profile interface{} `json:"p"`
}

type Conn interface {
SendQuery(s *Session, q *p.Query, t Term, opts map[string]interface{}, async bool) (*Cursor, error)
ReadResponse(s *Session, token int64) (*p.Response, error)
ReadResponse(s *Session, token int64) (*Response, error)
Close() error
}

Expand All @@ -31,55 +41,54 @@ type Connection struct {
}

// Dial closes the previous connection and attempts to connect again.
func Dial(s *Session) (*Connection, error) {
conn, err := net.Dial("tcp", s.address)
if err != nil {
return nil, RqlConnectionError{err.Error()}
}

// Send the protocol version to the server as a 4-byte little-endian-encoded integer
if err := binary.Write(conn, binary.LittleEndian, p.VersionDummy_V0_3); err != nil {
return nil, RqlConnectionError{err.Error()}
}
func Dial(s *Session) pool.Factory {
return func() (net.Conn, error) {
conn, err := net.Dial("tcp", s.address)
if err != nil {
return nil, RqlConnectionError{err.Error()}
}

// Send the length of the auth key to the server as a 4-byte little-endian-encoded integer
if err := binary.Write(conn, binary.LittleEndian, uint32(len(s.authkey))); err != nil {
return nil, RqlConnectionError{err.Error()}
}
// Send the protocol version to the server as a 4-byte little-endian-encoded integer
if err := binary.Write(conn, binary.LittleEndian, p.VersionDummy_V0_3); err != nil {
return nil, RqlConnectionError{err.Error()}
}

// Send the auth key as an ASCII string
// If there is no auth key, skip this step
if s.authkey != "" {
if _, err := io.WriteString(conn, s.authkey); err != nil {
// Send the length of the auth key to the server as a 4-byte little-endian-encoded integer
if err := binary.Write(conn, binary.LittleEndian, uint32(len(s.authkey))); err != nil {
return nil, RqlConnectionError{err.Error()}
}
}

// Send the protocol type as a 4-byte little-endian-encoded integer
if err := binary.Write(conn, binary.LittleEndian, p.VersionDummy_PROTOBUF); err != nil {
return nil, RqlConnectionError{err.Error()}
}
// Send the auth key as an ASCII string
// If there is no auth key, skip this step
if s.authkey != "" {
if _, err := io.WriteString(conn, s.authkey); err != nil {
return nil, RqlConnectionError{err.Error()}
}
}

// read server response to authorization key (terminated by NUL)
reader := bufio.NewReader(conn)
line, err := reader.ReadBytes('\x00')
if err != nil {
if err == io.EOF {
return nil, fmt.Errorf("Unexpected EOF: %s", string(line))
// Send the protocol type as a 4-byte little-endian-encoded integer
if err := binary.Write(conn, binary.LittleEndian, p.VersionDummy_JSON); err != nil {
return nil, RqlConnectionError{err.Error()}
}
return nil, RqlDriverError{err.Error()}
}
// convert to string and remove trailing NUL byte
response := string(line[:len(line)-1])
if response != "SUCCESS" {
// we failed authorization or something else terrible happened
return nil, RqlDriverError{fmt.Sprintf("Server dropped connection with message: \"%s\"", response)}
}

return &Connection{
s: s,
Conn: conn,
}, nil
// read server response to authorization key (terminated by NUL)
reader := bufio.NewReader(conn)
line, err := reader.ReadBytes('\x00')
if err != nil {
if err == io.EOF {
return nil, fmt.Errorf("Unexpected EOF: %s", string(line))
}
return nil, RqlDriverError{err.Error()}
}
// convert to string and remove trailing NUL byte
response := string(line[:len(line)-1])
if response != "SUCCESS" {
// we failed authorization or something else terrible happened
return nil, RqlDriverError{fmt.Sprintf("Server dropped connection with message: \"%s\"", response)}
}

return conn, nil
}
}

func TestOnBorrow(c *Connection, t time.Time) error {
Expand All @@ -97,26 +106,36 @@ func TestOnBorrow(c *Connection, t time.Time) error {
return nil
}

func (c *Connection) ReadResponse(s *Session, token int64) (*p.Response, error) {
func (c *Connection) ReadResponse(s *Session, token int64) (*Response, error) {
for {
// Read the 8-byte token of the query the response corresponds to.
var responseToken int64
if err := binary.Read(c, binary.LittleEndian, &responseToken); err != nil {
return nil, RqlConnectionError{err.Error()}
}

// Read the length of the JSON-encoded response as a 4-byte
// little-endian-encoded integer.
var messageLength uint32
if err := binary.Read(c, binary.LittleEndian, &messageLength); err != nil {
c.Close()
return nil, RqlConnectionError{err.Error()}
}

buffer := make([]byte, messageLength)
if _, err := io.ReadFull(c, buffer); err != nil {
c.Close()
// Read the JSON encoding of the Response itself.
b := make([]byte, messageLength)
if _, err := io.ReadFull(c, b); err != nil {
return nil, RqlDriverError{err.Error()}
}

response := &p.Response{}
if err := proto.Unmarshal(buffer, response); err != nil {
// Decode the response
var response = new(Response)
response.Token = responseToken
err := json.Unmarshal(b, response)
if err != nil {
return nil, RqlDriverError{err.Error()}
}

if response.GetToken() == token {
if responseToken == token {
return response, nil
} else if cursor, ok := s.checkCache(token); ok {
// Handle batch response
Expand All @@ -127,10 +146,15 @@ func (c *Connection) ReadResponse(s *Session, token int64) (*p.Response, error)
}
}

func (c *Connection) SendQuery(s *Session, q *p.Query, t Term, opts map[string]interface{}, async bool) (*Cursor, error) {
var data []byte
func (c *Connection) SendQuery(s *Session, q Query, opts map[string]interface{}, async bool) (*Cursor, error) {
var err error

// Build query
b, err := json.Marshal(q.build())
if err != nil {
return nil, RqlDriverError{"Error building query"}
}

// Ensure that the connection is not closed
if s.closed {
return nil, RqlDriverError{"Connection is closed"}
Expand All @@ -143,63 +167,52 @@ func (c *Connection) SendQuery(s *Session, q *p.Query, t Term, opts map[string]i
c.SetDeadline(time.Now().Add(s.timeout))
}

// Send query
if data, err = proto.Marshal(q); err != nil {
return nil, RqlDriverError{err.Error()}
}
if err = binary.Write(c, binary.LittleEndian, uint32(len(data))); err != nil {
// Send a unique 8-byte token
if err = binary.Write(c, binary.LittleEndian, q.Token); err != nil {
c.Close()
return nil, RqlConnectionError{err.Error()}
}

if err = binary.Write(c, binary.BigEndian, data); err != nil {
c.Close()
// Send the length of the JSON-encoded query as a 4-byte
// little-endian-encoded integer.
if err = binary.Write(c, binary.LittleEndian, uint32(len(b))); err != nil {
return nil, RqlConnectionError{err.Error()}
}

// Send the JSON encoding of the query itself.
if err = binary.Write(c, binary.BigEndian, b); err != nil {
return nil, RqlConnectionError{err.Error()}
}

// Return immediately if the noreply option was set
if noreply, ok := opts["noreply"]; ok && noreply.(bool) {
c.Close()
return nil, nil
} else if async {
if noreply, ok := opts["noreply"]; (ok && noreply.(bool)) || async {
return nil, nil
}

// Get response
response, err := c.ReadResponse(s, *q.Token)
response, err := c.ReadResponse(s, q.Token)
if err != nil {
return nil, err
}

err = checkErrorResponse(response, t)
err = checkErrorResponse(response, q.Term)
if err != nil {
return nil, err
}

// De-construct the profile datum if it exists
var profile interface{}
if response.GetProfile() != nil {
var err error

profile, err = deconstructDatum(response.GetProfile(), opts)
if err != nil {
return nil, RqlDriverError{err.Error()}
}
}

// De-construct datum and return a cursor
switch response.GetType() {
switch response.Type {
case p.Response_SUCCESS_PARTIAL, p.Response_SUCCESS_SEQUENCE, p.Response_SUCCESS_FEED:
cursor := &Cursor{
session: s,
conn: c,
query: q,
term: t,
term: *q.Term,
opts: opts,
profile: profile,
profile: response.Profile,
}

s.setCache(*q.Token, cursor)
s.setCache(q.Token, cursor)

cursor.extend(response)

Expand All @@ -208,17 +221,15 @@ func (c *Connection) SendQuery(s *Session, q *p.Query, t Term, opts map[string]i
var value []interface{}
var err error

if len(response.GetResponse()) < 1 {
if len(response.Responses) < 1 {
value = []interface{}{}
} else if response.GetResponse()[0].GetType() == p.Datum_R_ARRAY {
value, err = deconstructDatums(response.GetResponse()[0].GetRArray(), opts)
if err != nil {
return nil, err
}
} else {
var v interface{}

v, err = deconstructDatum(response.GetResponse()[0], opts)
v, err = recursivelyConvertPseudotype(response.Responses[0], opts)
if err != nil {
return nil, err
}
if err != nil {
return nil, RqlDriverError{err.Error()}
}
Expand All @@ -236,9 +247,9 @@ func (c *Connection) SendQuery(s *Session, q *p.Query, t Term, opts map[string]i
session: s,
conn: c,
query: q,
term: t,
term: *q.Term,
opts: opts,
profile: profile,
profile: response.Profile,
buffer: value,
finished: true,
}
Expand All @@ -247,7 +258,7 @@ func (c *Connection) SendQuery(s *Session, q *p.Query, t Term, opts map[string]i
case p.Response_WAIT_COMPLETE:
return nil, nil
default:
return nil, RqlDriverError{fmt.Sprintf("Unexpected response type received: %s", response.GetType())}
return nil, RqlDriverError{fmt.Sprintf("Unexpected response type received: %s", response.Type)}
}
}

Expand All @@ -265,28 +276,28 @@ func (c *Connection) CloseNoWait() error {
c.closed = true
c.Unlock()

return c.Conn.Close()
return c.s.pool.Put(c.Conn)
}

// noreplyWaitQuery sends the NOREPLY_WAIT query to the server.
// TODO: Removed duplicated functions in connection and session
// for NoReplyWait
func (c *Connection) NoreplyWait() error {
q := &p.Query{
Type: p.Query_NOREPLY_WAIT.Enum(),
Token: proto.Int64(c.s.nextToken()),
q := Query{
Type: p.Query_NOREPLY_WAIT,
Token: c.s.nextToken(),
}

_, err := c.SendQuery(c.s, q, Term{}, map[string]interface{}{}, false)
_, err := c.SendQuery(c.s, q, map[string]interface{}{}, false)
if err != nil {
return err
}

return nil
}

func checkErrorResponse(response *p.Response, t Term) error {
switch response.GetType() {
func checkErrorResponse(response *Response, t *Term) error {
switch response.Type {
case p.Response_CLIENT_ERROR:
return RqlClientError{rqlResponseError{response, t}}
case p.Response_COMPILE_ERROR:
Expand Down
Loading

0 comments on commit 36ef8b1

Please sign in to comment.