register for schema events

Register for schema change events and clear the prepared statement cache
so we don't use stale schema information for queries.

cluster.go

@@ -149,6 +149,8 @@ type ClusterConfig struct {
 		DisableNodeStatusEvents bool
 		// disable registering for topology events (node added/removed/moved)
 		DisableTopologyEvents bool
+		// disable registering for schema events (keyspace/table/function removed/created/updated)
+		DisableSchemaEvents bool
 	}
 
 	// internal config for testing

control.go

@@ -131,6 +131,9 @@ func (c *controlConn) registerEvents(conn *Conn) error {
 	if !c.session.cfg.Events.DisableNodeStatusEvents {
 		events = append(events, "STATUS_CHANGE")
 	}
+	if !c.session.cfg.Events.DisableSchemaEvents {
+		events = append(events, "SCHEMA_CHANGE")
+	}
 
 	if len(events) == 0 {
 		return nil

events.go

@@ -80,6 +80,37 @@ func (e *eventDeouncer) debounce(frame frame) {
 	e.mu.Unlock()
 }
 
+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()
+	if err != nil {
+		// TODO: logger
+		log.Printf("gocql: unable to parse event frame: %v\n", err)
+		return
+	}
+
+	if debug {
+		log.Printf("gocql: handling frame: %v\n", frame)
+	}
+
+	// TODO: handle medatadata events
+	switch f := frame.(type) {
+	case *schemaChangeKeyspace, *schemaChangeFunction, *schemaChangeTable:
+		s.schemaEvents.debounce(frame)
+	case *topologyChangeEventFrame, *statusChangeEventFrame:
+		s.nodeEvents.debounce(frame)
+	default:
+		log.Printf("gocql: invalid event frame (%T): %v\n", f, f)
+	}
+}
+
+func (s *Session) handleSchemaEvent(frames []frame) {
+	// for now we dont care about them, just reset the prepared statements
+	s.stmtsLRU.clear()
+}
+
 func (s *Session) handleNodeEvent(frames []frame) {
 	type nodeEvent struct {
 		change string
@@ -131,34 +162,6 @@ func (s *Session) handleNodeEvent(frames []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()
-	if err != nil {
-		// TODO: logger
-		log.Printf("gocql: unable to parse event frame: %v\n", err)
-		return
-	}
-
-	if debug {
-		log.Printf("gocql: handling frame: %v\n", frame)
-	}
-
-	// TODO: handle medatadata events
-	switch f := frame.(type) {
-	case *schemaChangeKeyspace:
-	case *schemaChangeFunction:
-	case *schemaChangeTable:
-	case *topologyChangeEventFrame, *statusChangeEventFrame:
-		s.nodeEvents.debounce(frame)
-	default:
-		log.Printf("gocql: invalid event frame (%T): %v\n", f, f)
-	}
-
-}
-
 func (s *Session) handleNewNode(host net.IP, port int, waitForBinary bool) {
 	// TODO(zariel): need to be able to filter discovered nodes
 

session.go

@@ -48,7 +48,8 @@ type Session struct {
 	control *controlConn
 
 	// event handlers
-	nodeEvents *eventDeouncer
+	nodeEvents   *eventDeouncer
+	schemaEvents *eventDeouncer
 
 	// ring metadata
 	hosts []HostInfo
@@ -102,6 +103,7 @@ func NewSession(cfg ClusterConfig) (*Session, error) {
 	s.connCfg = connCfg
 
 	s.nodeEvents = newEventDeouncer("NodeEvents", s.handleNodeEvent)
+	s.schemaEvents = newEventDeouncer("SchemaEvents", s.handleSchemaEvent)
 
 	s.routingKeyInfoCache.lru = lru.New(cfg.MaxRoutingKeyInfo)