WIP

Author: OleksiienkoMykyta

cassandra_test.go

@@ -32,6 +32,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"github.com/gocql/gocql/protocol"
 	"github.com/stretchr/testify/require"
 	"io"
 	"math"
@@ -519,23 +520,23 @@ func TestDurationType(t *testing.T) {
 		t.Fatal("create:", err)
 	}
 
-	durations := []Duration{
-		Duration{
+	durations := []protocol.Duration{
+		protocol.Duration{
 			Months:      250,
 			Days:        500,
 			Nanoseconds: 300010001,
 		},
-		Duration{
+		protocol.Duration{
 			Months:      -250,
 			Days:        -500,
 			Nanoseconds: -300010001,
 		},
-		Duration{
+		protocol.Duration{
 			Months:      0,
 			Days:        128,
 			Nanoseconds: 127,
 		},
-		Duration{
+		protocol.Duration{
 			Months:      0x7FFFFFFF,
 			Days:        0x7FFFFFFF,
 			Nanoseconds: 0x7FFFFFFFFFFFFFFF,
@@ -547,7 +548,7 @@ func TestDurationType(t *testing.T) {
 		}
 
 		var id int
-		var duration Duration
+		var duration protocol.Duration
 		if err := session.Query(`SELECT k, v FROM gocql_test.duration_table`).Scan(&id, &duration); err != nil {
 			t.Fatal(err)
 		}

cluster.go

@@ -27,6 +27,7 @@ package gocql
 import (
 	"context"
 	"errors"
+	"github.com/gocql/gocql/consistency"
 	"net"
 	"time"
 )
@@ -114,7 +115,7 @@ type ClusterConfig struct {
 
 	// Default consistency level.
 	// Default: Quorum
-	Consistency Consistency
+	Consistency consistency.Consistency
 
 	// Compression algorithm.
 	// Default: nil
@@ -156,7 +157,7 @@ type ClusterConfig struct {
 
 	// Consistency for the serial part of queries, values can be either SERIAL or LOCAL_SERIAL.
 	// Default: unset
-	SerialConsistency SerialConsistency
+	SerialConsistency consistency.SerialConsistency
 
 	// SslOpts configures TLS use when HostDialer is not set.
 	// SslOpts is ignored if HostDialer is set.

compressor.go

@@ -28,6 +28,7 @@ import (
 	"github.com/golang/snappy"
 )
 
+// Deprecated: use compressor.Compressor instead
 type Compressor interface {
 	Name() string
 	Encode(data []byte) ([]byte, error)

compressor/compressor.go

@@ -0,0 +1,26 @@
+package compressor
+
+import "github.com/golang/snappy"
+
+type Compressor interface {
+	Name() string
+	Encode(data []byte) ([]byte, error)
+	Decode(data []byte) ([]byte, error)
+}
+
+// SnappyCompressor implements the Compressor interface and can be used to
+// compress incoming and outgoing frames. The snappy compression algorithm
+// aims for very high speeds and reasonable compression.
+type SnappyCompressor struct{}
+
+func (s SnappyCompressor) Name() string {
+	return "snappy"
+}
+
+func (s SnappyCompressor) Encode(data []byte) ([]byte, error) {
+	return snappy.Encode(nil, data), nil
+}
+
+func (s SnappyCompressor) Decode(data []byte) ([]byte, error) {
+	return snappy.Decode(nil, data)
+}

conn.go

@@ -30,6 +30,7 @@ import (
 	"crypto/tls"
 	"errors"
 	"fmt"
+	"github.com/gocql/gocql/internal"
 	"io"
 	"io/ioutil"
 	"net"
@@ -1276,7 +1277,7 @@ func (c *Conn) prepareStatement(ctx context.Context, stmt string, tracer Tracer)
 				flight.preparedStatment = &preparedStatment{
 					// defensively copy as we will recycle the underlying buffer after we
 					// return.
-					id: copyBytes(x.preparedID),
+					id: internal.CopyBytes(x.preparedID),
 					// the type info's should _not_ have a reference to the framers read buffer,
 					// therefore we can just copy them directly.
 					request:  x.reqMeta,
@@ -1431,7 +1432,7 @@ func (c *Conn) executeQuery(ctx context.Context, qry *Query) *Iter {
 		if params.skipMeta {
 			if info != nil {
 				iter.meta = info.response
-				iter.meta.pagingState = copyBytes(x.meta.pagingState)
+				iter.meta.pagingState = internal.CopyBytes(x.meta.pagingState)
 			} else {
 				return &Iter{framer: framer, err: errors.New("gocql: did not receive metadata but prepared info is nil")}
 			}
@@ -1442,7 +1443,7 @@ func (c *Conn) executeQuery(ctx context.Context, qry *Query) *Iter {
 		if x.meta.morePages() && !qry.disableAutoPage {
 			newQry := new(Query)
 			*newQry = *qry
-			newQry.pageState = copyBytes(x.meta.pagingState)
+			newQry.pageState = internal.CopyBytes(x.meta.pagingState)
 			newQry.metrics = &queryMetrics{m: make(map[string]*hostMetrics)}
 
 			iter.next = &nextIter{
@@ -1659,7 +1660,7 @@ func (c *Conn) querySystemPeers(ctx context.Context, version cassVersion) *Iter
 
 		err := iter.checkErrAndNotFound()
 		if err != nil {
-			if errFrame, ok := err.(errorFrame); ok && errFrame.code == ErrCodeInvalid { // system.peers_v2 not found, try system.peers
+			if errFrame, ok := err.(errorFrame); ok && errFrame.code == gocql_errors.ErrCodeInvalid { // system.peers_v2 not found, try system.peers
 				c.mu.Lock()
 				c.isSchemaV2 = false
 				c.mu.Unlock()

consistency/consistency.go

@@ -0,0 +1,137 @@
+package consistency
+
+import (
+	"fmt"
+	"strings"
+)
+
+type Consistency uint16
+
+const (
+	Any         Consistency = 0x00
+	One         Consistency = 0x01
+	Two         Consistency = 0x02
+	Three       Consistency = 0x03
+	Quorum      Consistency = 0x04
+	All         Consistency = 0x05
+	LocalQuorum Consistency = 0x06
+	EachQuorum  Consistency = 0x07
+	LocalOne    Consistency = 0x0A
+)
+
+func (c Consistency) String() string {
+	switch c {
+	case Any:
+		return "ANY"
+	case One:
+		return "ONE"
+	case Two:
+		return "TWO"
+	case Three:
+		return "THREE"
+	case Quorum:
+		return "QUORUM"
+	case All:
+		return "ALL"
+	case LocalQuorum:
+		return "LOCAL_QUORUM"
+	case EachQuorum:
+		return "EACH_QUORUM"
+	case LocalOne:
+		return "LOCAL_ONE"
+	default:
+		return fmt.Sprintf("UNKNOWN_CONS_0x%x", uint16(c))
+	}
+}
+
+func (c Consistency) MarshalText() (text []byte, err error) {
+	return []byte(c.String()), nil
+}
+
+func (c *Consistency) UnmarshalText(text []byte) error {
+	switch string(text) {
+	case "ANY":
+		*c = Any
+	case "ONE":
+		*c = One
+	case "TWO":
+		*c = Two
+	case "THREE":
+		*c = Three
+	case "QUORUM":
+		*c = Quorum
+	case "ALL":
+		*c = All
+	case "LOCAL_QUORUM":
+		*c = LocalQuorum
+	case "EACH_QUORUM":
+		*c = EachQuorum
+	case "LOCAL_ONE":
+		*c = LocalOne
+	default:
+		return fmt.Errorf("invalid consistency %q", string(text))
+	}
+
+	return nil
+}
+
+func ParseConsistency(s string) Consistency {
+	var c Consistency
+	if err := c.UnmarshalText([]byte(strings.ToUpper(s))); err != nil {
+		panic(err)
+	}
+	return c
+}
+
+// ParseConsistencyWrapper wraps gocql.ParseConsistency to provide an err
+// return instead of a panic
+func ParseConsistencyWrapper(s string) (consistency Consistency, err error) {
+	err = consistency.UnmarshalText([]byte(strings.ToUpper(s)))
+	return
+}
+
+// MustParseConsistency is the same as ParseConsistency except it returns
+// an error (never). It is kept here since breaking changes are not good.
+// DEPRECATED: use ParseConsistency if you want a panic on parse error.
+func MustParseConsistency(s string) (Consistency, error) {
+	c, err := ParseConsistencyWrapper(s)
+	if err != nil {
+		panic(err)
+	}
+	return c, nil
+}
+
+type SerialConsistency uint16
+
+const (
+	Serial      SerialConsistency = 0x08
+	LocalSerial SerialConsistency = 0x09
+)
+
+func (s SerialConsistency) String() string {
+	switch s {
+	case Serial:
+		return "SERIAL"
+	case LocalSerial:
+		return "LOCAL_SERIAL"
+	default:
+		return fmt.Sprintf("UNKNOWN_SERIAL_CONS_0x%x", uint16(s))
+	}
+}
+
+func (s SerialConsistency) MarshalText() (text []byte, err error) {
+	return []byte(s.String()), nil
+}
+
+func (s *SerialConsistency) UnmarshalText(text []byte) error {
+	switch string(text) {
+	case "SERIAL":
+		*s = Serial
+	case "LOCAL_SERIAL":
+		*s = LocalSerial
+	default:
+		return fmt.Errorf("invalid consistency %q", string(text))
+	}
+
+	return nil
+}