TokenAwarePolicy: use token replicas per placement strategy (gocql#1039)

* TokenAwarePolicy: use token replicas per placement strategy

Add support for finding token replicas based on the placement stategy
for each keyspace.

Update host policies to recieve keyspace change updates and the ability
to set the session once they are created.

* rf is a string, parse it and provide helpful error messages

* fix panic when loading ks meta

* fix vet

batch_test.go

@@ -9,12 +9,15 @@ import (
 
 func TestBatch_Errors(t *testing.T) {
 	if *flagProto == 1 {
-		t.Skip("atomic batches not supported. Please use Cassandra >= 2.0")
 	}
 
 	session := createSession(t)
 	defer session.Close()
 
+	if session.cfg.ProtoVersion < protoVersion2 {
+		t.Skip("atomic batches not supported. Please use Cassandra >= 2.0")
+	}
+
 	if err := createTable(session, `CREATE TABLE gocql_test.batch_errors (id int primary key, val inet)`); err != nil {
 		t.Fatal(err)
 	}
@@ -27,13 +30,13 @@ func TestBatch_Errors(t *testing.T) {
 }
 
 func TestBatch_WithTimestamp(t *testing.T) {
-	if *flagProto < protoVersion3 {
-		t.Skip("Batch timestamps are only available on protocol >= 3")
-	}
-
 	session := createSession(t)
 	defer session.Close()
 
+	if session.cfg.ProtoVersion < protoVersion3 {
+		t.Skip("Batch timestamps are only available on protocol >= 3")
+	}
+
 	if err := createTable(session, `CREATE TABLE gocql_test.batch_ts (id int primary key, val text)`); err != nil {
 		t.Fatal(err)
 	}

cassandra_test.go

@@ -83,6 +83,7 @@ func TestEmptyHosts(t *testing.T) {
 }
 
 func TestInvalidPeerEntry(t *testing.T) {
+	t.Skip("dont mutate system tables, rewrite this to test what we mean to test")
 	session := createSession(t)
 
 	// rack, release_version, schema_version, tokens are all null

common_test.go

@@ -94,6 +94,7 @@ func createCluster() *ClusterConfig {
 }
 
 func createKeyspace(tb testing.TB, cluster *ClusterConfig, keyspace string) {
+	// TODO: tb.Helper()
 	c := *cluster
 	c.Keyspace = "system"
 	c.Timeout = 30 * time.Second

conn.go

@@ -895,7 +895,7 @@ func (c *Conn) executeQuery(qry *Query) *Iter {
 		return iter
 	case *resultKeyspaceFrame:
 		return &Iter{framer: framer}
-	case *schemaChangeKeyspace, *schemaChangeTable, *schemaChangeFunction:
+	case *schemaChangeKeyspace, *schemaChangeTable, *schemaChangeFunction, *schemaChangeAggregate, *schemaChangeType:
 		iter := &Iter{framer: framer}
 		if err := c.awaitSchemaAgreement(); err != nil {
 			// TODO: should have this behind a flag

events.go

@@ -80,7 +80,6 @@ func (e *eventDebouncer) debounce(frame frame) {
 }
 
 func (s *Session) handleEvent(framer *framer) {
-	// TODO(zariel): need to debounce events frames, and possible also events
 	defer framerPool.Put(framer)
 
 	frame, err := framer.parseFrame()
@@ -94,9 +93,10 @@ func (s *Session) handleEvent(framer *framer) {
 		Logger.Printf("gocql: handling frame: %v\n", frame)
 	}
 
-	// TODO: handle medatadata events
 	switch f := frame.(type) {
-	case *schemaChangeKeyspace, *schemaChangeFunction, *schemaChangeTable:
+	case *schemaChangeKeyspace, *schemaChangeFunction,
+		*schemaChangeTable, *schemaChangeAggregate, *schemaChangeType:
+
 		s.schemaEvents.debounce(frame)
 	case *topologyChangeEventFrame, *statusChangeEventFrame:
 		s.nodeEvents.debounce(frame)
@@ -106,22 +106,28 @@ func (s *Session) handleEvent(framer *framer) {
 }
 
 func (s *Session) handleSchemaEvent(frames []frame) {
-	s.mu.RLock()
-	defer s.mu.RUnlock()
-
-	if s.schemaDescriber == nil {
-		return
-	}
+	// TODO: debounce events
 	for _, frame := range frames {
 		switch f := frame.(type) {
 		case *schemaChangeKeyspace:
 			s.schemaDescriber.clearSchema(f.keyspace)
+			s.handleKeyspaceChange(f.keyspace, f.change)
 		case *schemaChangeTable:
 			s.schemaDescriber.clearSchema(f.keyspace)
+		case *schemaChangeAggregate:
+			s.schemaDescriber.clearSchema(f.keyspace)
+		case *schemaChangeFunction:
+			s.schemaDescriber.clearSchema(f.keyspace)
+		case *schemaChangeType:
+			s.schemaDescriber.clearSchema(f.keyspace)
 		}
 	}
 }
 
+func (s *Session) handleKeyspaceChange(keyspace, change string) {
+	s.policy.KeyspaceChanged(KeyspaceUpdateEvent{Keyspace: keyspace, Change: change})
+}
+
 func (s *Session) handleNodeEvent(frames []frame) {
 	type nodeEvent struct {
 		change string

frame.go

@@ -1112,6 +1112,14 @@ func (f schemaChangeTable) String() string {
 	return fmt.Sprintf("[event schema_change change=%q keyspace=%q object=%q]", f.change, f.keyspace, f.object)
 }
 
+type schemaChangeType struct {
+	frameHeader
+
+	change   string
+	keyspace string
+	object   string
+}
+
 type schemaChangeFunction struct {
 	frameHeader
 
@@ -1121,6 +1129,15 @@ type schemaChangeFunction struct {
 	args     []string
 }
 
+type schemaChangeAggregate struct {
+	frameHeader
+
+	change   string
+	keyspace string
+	name     string
+	args     []string
+}
+
 func (f *framer) parseResultSchemaChange() frame {
 	if f.proto <= protoVersion2 {
 		change := f.readString()
@@ -1156,7 +1173,7 @@ func (f *framer) parseResultSchemaChange() frame {
 			frame.keyspace = f.readString()
 
 			return frame
-		case "TABLE", "TYPE":
+		case "TABLE":
 			frame := &schemaChangeTable{
 				frameHeader: *f.header,
 				change:      change,
@@ -1166,7 +1183,17 @@ func (f *framer) parseResultSchemaChange() frame {
 			frame.object = f.readString()
 
 			return frame
-		case "FUNCTION", "AGGREGATE":
+		case "TYPE":
+			frame := &schemaChangeType{
+				frameHeader: *f.header,
+				change:      change,
+			}
+
+			frame.keyspace = f.readString()
+			frame.object = f.readString()
+
+			return frame
+		case "FUNCTION":
 			frame := &schemaChangeFunction{
 				frameHeader: *f.header,
 				change:      change,
@@ -1176,6 +1203,17 @@ func (f *framer) parseResultSchemaChange() frame {
 			frame.name = f.readString()
 			frame.args = f.readStringList()
 
+			return frame
+		case "AGGREGATE":
+			frame := &schemaChangeAggregate{
+				frameHeader: *f.header,
+				change:      change,
+			}
+
+			frame.keyspace = f.readString()
+			frame.name = f.readString()
+			frame.args = f.readStringList()
+
 			return frame
 		default:
 			panic(fmt.Errorf("gocql: unknown SCHEMA_CHANGE target: %q change: %q", target, change))