cluster.go

// Copyright (c) 2012 The gocql Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.

package gocql

import (
	"errors"
	"github.com/golang/groupcache/lru"
	"sync"
	"time"
)

//Package global reference to Prepared Statements LRU
var stmtsLRU preparedLRU

//preparedLRU is the prepared statement cache
type preparedLRU struct {
	lru *lru.Cache
	mu  sync.Mutex
}

//Max adjusts the maximum size of the cache and cleans up the oldest records if
//the new max is lower than the previous value. Not concurrency safe.
func (p *preparedLRU) Max(max int) {
	for p.lru.Len() > max {
		p.lru.RemoveOldest()
	}
	p.lru.MaxEntries = max
}

// PoolHandlers are used for asynchronous events such as opened and closed connections
type PoolHandler func(ConnectionPool)

// ClusterConfig is a struct to configure the default cluster implementation
// of gocoql. It has a varity of attributes that can be used to modify the
// behavior to fit the most common use cases. Applications that requre a
// different setup must implement their own cluster.
type ClusterConfig struct {
	Hosts             []string      // addresses for the initial connections
	CQLVersion        string        // CQL version (default: 3.0.0)
	ProtoVersion      int           // version of the native protocol (default: 2)
	Timeout           time.Duration // connection timeout (default: 600ms)
	DefaultPort       int           // default port (default: 9042)
	Keyspace          string        // initial keyspace (optional)
	NumConns          int           // number of connections per host (default: 2)
	NumStreams        int           // number of streams per connection (default: 128)
	Consistency       Consistency   // default consistency level (default: Quorum)
	Compressor        Compressor    // compression algorithm (default: nil)
	Authenticator     Authenticator // authenticator (default: nil)
	RetryPolicy       RetryPolicy   // Default retry policy to use for queries (default: 0)
	SocketKeepalive   time.Duration // The keepalive period to use, enabled if > 0 (default: 0)
	ConnPoolType      NewPoolFunc   // The function used to create the connection pool for the session (default: NewSimplePool)
	DiscoverHosts     bool          // If set, gocql will attempt to automatically discover other members of the Cassandra cluster (default: false)
	MaxPreparedStmts  int           // Sets the maximum cache size for prepared statements globally for gocql (default: 1000)
	ConnectHandler    PoolHandler   // Called when a connection is added to a pool
	DisconnectHandler PoolHandler   // Called when a connection is removed from a pool
}

// NewCluster generates a new config for the default cluster implementation.
func NewCluster(hosts ...string) *ClusterConfig {
	cfg := &ClusterConfig{
		Hosts:            hosts,
		CQLVersion:       "3.0.0",
		ProtoVersion:     2,
		Timeout:          600 * time.Millisecond,
		DefaultPort:      9042,
		NumConns:         2,
		NumStreams:       128,
		Consistency:      Quorum,
		ConnPoolType:     NewSimplePool,
		DiscoverHosts:    false,
		MaxPreparedStmts: 1000,
	}
	return cfg
}

// CreateSession initializes the cluster based on this config and returns a
// session object that can be used to interact with the database.
func (cfg *ClusterConfig) CreateSession() (*Session, error) {

	//Check that hosts in the ClusterConfig is not empty
	if len(cfg.Hosts) < 1 {
		return nil, ErrNoHosts
	}
	pool := cfg.ConnPoolType(cfg)

	//Adjust the size of the prepared statements cache to match the latest configuration
	stmtsLRU.mu.Lock()
	if stmtsLRU.lru != nil {
		stmtsLRU.Max(cfg.MaxPreparedStmts)
	} else {
		stmtsLRU.lru = lru.New(cfg.MaxPreparedStmts)
	}
	stmtsLRU.mu.Unlock()

	//See if there are any connections in the pool
	if pool.Size() > 0 {
		s := NewSession(pool, *cfg)
		s.SetConsistency(cfg.Consistency)

		if cfg.DiscoverHosts {
			//Fill out cfg.Hosts
			query := "SELECT peer FROM system.peers"
			peers := s.Query(query).Iter()

			var ip string
			for peers.Scan(&ip) {
				exists := false
				for ii := 0; ii < len(cfg.Hosts); ii++ {
					if cfg.Hosts[ii] == ip {
						exists = true
					}
				}
				if !exists {
					cfg.Hosts = append(cfg.Hosts, ip)
				}
			}

			if err := peers.Close(); err != nil {
				return s, ErrHostQueryFailed
			}
		}

		return s, nil
	}

	pool.Close()
	return nil, ErrNoConnectionsStarted

}

var (
	ErrNoHosts              = errors.New("no hosts provided")
	ErrNoConnectionsStarted = errors.New("no connections were made when creating the session")
	ErrHostQueryFailed      = errors.New("unable to populate Hosts")
)