Merge pull request #72 from datastax/use-cluster-cql-version

Return server-side cql version for supported options

Author: mpenick

proxy/proxy.go

@@ -78,7 +78,7 @@ type Proxy struct {
 	systemLocalValues   map[string]message.Column
 }
 
-func (p *Proxy) OnEvent(event interface{}) {
+func (p *Proxy) OnEvent(event proxycore.Event) {
 	switch evt := event.(type) {
 	case *proxycore.SchemaChangeEvent:
 		frm := frame.NewFrame(p.cluster.NegotiatedVersion, -1, evt.Message)
@@ -356,7 +356,10 @@ func (c *client) Receive(reader io.Reader) error {
 
 	switch msg := body.Message.(type) {
 	case *message.Options:
-		c.send(raw.Header, &message.Supported{Options: map[string][]string{"CQL_VERSION": {"3.0.0"}, "COMPRESSION": {}}})
+		c.send(raw.Header, &message.Supported{Options: map[string][]string{
+			"CQL_VERSION": {c.proxy.cluster.Info.CQLVersion},
+			"COMPRESSION": {},
+		}})
 	case *message.Startup:
 		c.send(raw.Header, &message.Ready{})
 	case *message.Register:

proxycore/cluster.go

@@ -33,33 +33,65 @@ const (
 	DefaultRefreshTimeout = 5 * time.Second
 )
 
+type Event interface {
+	isEvent() // Marker method for the event interface
+}
+
 type AddEvent struct {
 	Host *Host
 }
 
+func (a AddEvent) isEvent() {
+	panic("do not call")
+}
+
 type RemoveEvent struct {
 	Host *Host
 }
 
+func (r RemoveEvent) isEvent() {
+	panic("do not call")
+}
+
+type UpEvent struct {
+	Host *Host
+}
+
+func (a UpEvent) isEvent() {
+	panic("do not call")
+}
+
 type BootstrapEvent struct {
 	Hosts []*Host
 }
 
+func (b BootstrapEvent) isEvent() {
+	panic("do not call")
+}
+
 type SchemaChangeEvent struct {
 	Message *message.SchemaChangeEvent
 }
 
+func (s SchemaChangeEvent) isEvent() {
+	panic("do not call")
+}
+
 type ReconnectEvent struct {
 	Endpoint
 }
 
+func (r ReconnectEvent) isEvent() {
+	panic("do not call")
+}
+
 type ClusterListener interface {
-	OnEvent(event interface{})
+	OnEvent(event Event)
 }
 
-type ClusterListenerFunc func(event interface{})
+type ClusterListenerFunc func(event Event)
 
-func (f ClusterListenerFunc) OnEvent(event interface{}) {
+func (f ClusterListenerFunc) OnEvent(event Event) {
 	f(event)
 }
 
@@ -236,7 +268,7 @@ func (c *Cluster) mergeHosts(hosts []*Host) error {
 	return nil
 }
 
-func (c *Cluster) sendEvent(event interface{}) {
+func (c *Cluster) sendEvent(event Event) {
 	for _, listener := range c.listeners {
 		listener.OnEvent(event)
 	}

proxycore/cluster_test.go

@@ -56,7 +56,7 @@ func TestConnectCluster(t *testing.T) {
 
 	events := make(chan interface{})
 
-	err = cluster.Listen(ClusterListenerFunc(func(event interface{}) {
+	err = cluster.Listen(ClusterListenerFunc(func(event Event) {
 		events <- event
 	}))
 	require.NoError(t, err)

proxycore/lb.go

@@ -42,7 +42,7 @@ type roundRobinLoadBalancer struct {
 	mu    *sync.Mutex
 }
 
-func (l *roundRobinLoadBalancer) OnEvent(event interface{}) {
+func (l *roundRobinLoadBalancer) OnEvent(event Event) {
 	l.mu.Lock()
 	defer l.mu.Unlock()
 

proxycore/session.go

@@ -106,7 +106,7 @@ func (s *Session) leastBusyConn(host *Host) *ClientConn {
 	return nil
 }
 
-func (s *Session) OnEvent(event interface{}) {
+func (s *Session) OnEvent(event Event) {
 	switch evt := event.(type) {
 	case *BootstrapEvent:
 		go func() {