Add event bus

Create a bus that will allow consume different events from the Session.
Targeted event types:
1. Cluster events that comes from Control connection, like: alter schema, topology change, node status change
2. Session events: Control Connection being recreated

Session API should allow to subscribe to events, and unsubscribe, it
should allow event filtering.

Author: dkropachev

cluster.go

@@ -37,6 +37,7 @@ import (
 	"time"
 
 	"github.com/gocql/gocql/internal/debug"
+	"github.com/gocql/gocql/internal/eventbus"
 )
 
 const defaultDriverName = "ScyllaDB GoCQL Driver"
@@ -283,6 +284,8 @@ type ClusterConfig struct {
 	// address in system.local or system.peers returns 127.0.0.1, the peer will be
 	// set to 10.0.0.1 which is what will be used to connect to.
 	IgnorePeerAddr bool
+	// An event bus configuration
+	EventBusConfig eventbus.EventBusConfig
 }
 
 type DNSResolver interface {
@@ -365,6 +368,7 @@ type Dialer interface {
 // resolves to more than 1 IP address then the driver may connect multiple times to
 // the same host, and will not mark the node being down or up from events.
 func NewCluster(hosts ...string) *ClusterConfig {
+	logger := &defaultLogger{}
 	cfg := &ClusterConfig{
 		Hosts:                         hosts,
 		CQLVersion:                    "3.0.0",
@@ -392,8 +396,11 @@ func NewCluster(hosts ...string) *ClusterConfig {
 		MetadataSchemaRequestTimeout:  60 * time.Second,
 		DisableSkipMetadata:           true,
 		WarningsHandlerBuilder:        DefaultWarningHandlerBuilder,
-		Logger:                        &defaultLogger{},
+		Logger:                        logger,
 		DNSResolver:                   defaultDnsResolver,
+		EventBusConfig: eventbus.EventBusConfig{
+			InputEventsQueueSize: 10240,
+		},
 	}
 
 	return cfg

control.go

@@ -37,6 +37,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/gocql/gocql/events"
 	"github.com/gocql/gocql/internal/debug"
 	frm "github.com/gocql/gocql/internal/frame"
 )
@@ -337,8 +338,21 @@ func (c *controlConn) setupConn(conn *Conn) error {
 		conn: conn,
 		host: host,
 	}
-
-	c.conn.Store(ch)
+	old, _ := c.conn.Swap(ch).(*connHost)
+	var oldHost events.HostInfo
+	if old != nil && old.host != nil {
+		oldHost.HostID = old.host.HostID()
+		oldHost.Host = old.host.ConnectAddress()
+		oldHost.Port = old.host.Port()
+	}
+	c.session.publishEvent(&events.ControlConnectionRecreatedEvent{
+		OldHost: oldHost,
+		NewHost: events.HostInfo{
+			HostID: host.HostID(),
+			Host:   host.ConnectAddress(),
+			Port:   host.Port(),
+		},
+	})
 	if c.session.initialized() {
 		// We connected to control conn, so add the connect the host in pool as well.
 		// Notify session we can start trying to connect to the node.
@@ -465,6 +479,13 @@ func (c *controlConn) attemptReconnectToAnyOfHosts(hosts []*HostInfo) error {
 			continue
 		}
 		conn.finalizeConnection()
+		c.session.publishEvent(&events.ControlConnectionRecreatedEvent{
+			NewHost: events.HostInfo{
+				Host:   host.ConnectAddress(),
+				Port:   host.Port(),
+				HostID: host.HostID(),
+			},
+		})
 		return nil
 	}
 	return fmt.Errorf("unable to connect to any known node: %v", hosts)

events.go

@@ -29,6 +29,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/gocql/gocql/events"
 	"github.com/gocql/gocql/internal/debug"
 	frm "github.com/gocql/gocql/internal/frame"
 )
@@ -107,6 +108,16 @@ func (e *eventDebouncer) debounce(frame frame) {
 	e.mu.Unlock()
 }
 
+func (s *Session) publishEvent(event events.Event) {
+	if s.eventBus == nil {
+		return
+	}
+
+	if !s.eventBus.PublishEvent(event) {
+		s.logger.Printf("can't publish event, eventbus is full, increase Cluster.EventBusConfig.InputEventsQueueSize; event is dropped")
+	}
+}
+
 func (s *Session) handleEvent(framer *framer) {
 	frame, err := framer.parseFrame()
 	if err != nil {
@@ -118,6 +129,10 @@ func (s *Session) handleEvent(framer *framer) {
 		s.logger.Printf("gocql: handling frame: %v\n", frame)
 	}
 
+	if event := events.FrameToEvent(frame); event != nil {
+		s.publishEvent(event)
+	}
+
 	switch f := frame.(type) {
 	case *frm.SchemaChangeKeyspace, *frm.SchemaChangeFunction,
 		*frm.SchemaChangeTable, *frm.SchemaChangeAggregate, *frm.SchemaChangeType:

events/event_converter.go

@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package events
+
+import (
+	frm "github.com/gocql/gocql/internal/frame"
+)
+
+// FrameToEvent converts an internal frame to a public Event interface.
+// This function has access to internal frame types and can perform
+// type-safe conversions.
+// Returns nil if the frame is not an event frame.
+func FrameToEvent(f interface{}) Event {
+	if f == nil {
+		return nil
+	}
+
+	switch frame := f.(type) {
+	case *frm.TopologyChangeEventFrame:
+		return &TopologyChangeEvent{
+			Change: frame.Change,
+			Host:   frame.Host,
+			Port:   frame.Port,
+		}
+
+	case *frm.StatusChangeEventFrame:
+		return &StatusChangeEvent{
+			Change: frame.Change,
+			Host:   frame.Host,
+			Port:   frame.Port,
+		}
+
+	case *frm.SchemaChangeKeyspace:
+		return &SchemaChangeKeyspaceEvent{
+			Change:   frame.Change,
+			Keyspace: frame.Keyspace,
+		}
+
+	case *frm.SchemaChangeTable:
+		return &SchemaChangeTableEvent{
+			Change:   frame.Change,
+			Keyspace: frame.Keyspace,
+			Table:    frame.Object,
+		}
+
+	case *frm.SchemaChangeType:
+		return &SchemaChangeTypeEvent{
+			Change:   frame.Change,
+			Keyspace: frame.Keyspace,
+			TypeName: frame.Object,
+		}
+
+	case *frm.SchemaChangeFunction:
+		return &SchemaChangeFunctionEvent{
+			Change:    frame.Change,
+			Keyspace:  frame.Keyspace,
+			Function:  frame.Name,
+			Arguments: frame.Args,
+		}
+
+	case *frm.SchemaChangeAggregate:
+		return &SchemaChangeAggregateEvent{
+			Change:    frame.Change,
+			Keyspace:  frame.Keyspace,
+			Aggregate: frame.Name,
+			Arguments: frame.Args,
+		}
+
+	default:
+		return nil
+	}
+}