Add new API for non blocking Session creation

For some use-cases it is helpful to be able to start the
initialization of the Session, but not neccesarly wait for
it to finish. To accomodate this need this commit adds new
non blocking API.

If the user tries to query Session that isn't initialized
yet, the new ErrSessionNotReady will be returned.

cluster.go

@@ -340,6 +340,10 @@ func (cfg *ClusterConfig) CreateSession() (*Session, error) {
 	return NewSession(*cfg)
 }
 
+func (cfg *ClusterConfig) CreateSessionNonBlocking() (*Session, error) {
+	return NewSessionNonBlocking(*cfg)
+}
+
 // translateAddressPort is a helper method that will use the given AddressTranslator
 // if defined, to translate the given address and port into a possibly new address
 // and port, If no AddressTranslator or if an error occurs, the given address and

scylla_shard_aware_port_common_test.go

@@ -46,6 +46,11 @@ func testShardAwarePortNoReconnections(t *testing.T, makeCluster makeClusterTest
 			}
 			defer sess.Close()
 
+			if err = sess.WaitUntilReady(); err != nil {
+				cancel()
+				return
+			}
+
 			if err := waitUntilPoolsStopFilling(ctx, sess, 10*time.Second); err != nil {
 				cancel()
 				return
@@ -118,6 +123,10 @@ func testShardAwarePortMaliciousNAT(t *testing.T, makeCluster makeClusterTestFun
 	}
 	defer sess.Close()
 
+	if err = sess.WaitUntilReady(); err != nil {
+		t.Fatalf("an error occurred while initializing a session: %s", err)
+	}
+
 	// In this situation we are guaranteed that the connection will miss one
 	// shard at this point. The first connection receives a random shard,
 	// then we establish N-1 connections, targeting remaining shards.
@@ -150,6 +159,10 @@ func testShardAwarePortUnreachable(t *testing.T, makeCluster makeClusterTestFunc
 	}
 	defer sess.Close()
 
+	if err = sess.WaitUntilReady(); err != nil {
+		t.Fatalf("an error occurred while initializing a session: %s", err)
+	}
+
 	// In this situation, the connecting to the shard-aware port will fail,
 	// but connections to the non-shard-aware port will succeed. This test
 	// checks that we detect that the shard-aware-port is unreachable and
@@ -187,6 +200,10 @@ func testShardAwarePortUnusedIfNotEnabled(t *testing.T, makeCluster makeClusterT
 	}
 	defer sess.Close()
 
+	if err = sess.WaitUntilReady(); err != nil {
+		t.Fatalf("an error occurred while initializing a session: %s", err)
+	}
+
 	if err := waitUntilPoolsStopFilling(context.Background(), sess, 10*time.Second); err != nil {
 		t.Fatal(err)
 	}

session.go

@@ -79,6 +79,8 @@ type Session struct {
 	// isInitialized is true once Session.init succeeds.
 	// you can use initialized() to read the value.
 	isInitialized bool
+	initErr       error
+	readyCh       chan error
 
 	logger StdLogger
 
@@ -114,8 +116,7 @@ func addrsToHosts(addrs []string, defaultPort int, logger StdLogger) ([]*HostInf
 	return hosts, nil
 }
 
-// NewSession wraps an existing Node.
-func NewSession(cfg ClusterConfig) (*Session, error) {
+func newSessionCommon(cfg ClusterConfig) (*Session, error) {
 	if err := cfg.Validate(); err != nil {
 		return nil, fmt.Errorf("gocql: unable to create session: cluster config validation failed: %v", err)
 	}
@@ -132,6 +133,7 @@ func NewSession(cfg ClusterConfig) (*Session, error) {
 		ctx:             ctx,
 		cancel:          cancel,
 		logger:          cfg.logger(),
+		readyCh:         make(chan error, 1),
 	}
 
 	// Close created resources on error otherwise they'll leak
@@ -181,6 +183,16 @@ func NewSession(cfg ClusterConfig) (*Session, error) {
 	}
 	s.connCfg = connCfg
 
+	return s, nil
+}
+
+// NewSession wraps an existing Node.
+func NewSession(cfg ClusterConfig) (*Session, error) {
+	s, err := newSessionCommon(cfg)
+	if err != nil {
+		return nil, err
+	}
+
 	if err = s.init(); err != nil {
 		if err == ErrNoConnectionsStarted {
 			//This error used to be generated inside NewSession & returned directly
@@ -192,6 +204,29 @@ func NewSession(cfg ClusterConfig) (*Session, error) {
 		}
 	}
 
+	s.readyCh <- nil
+	close(s.readyCh)
+
+	return s, nil
+}
+
+func NewSessionNonBlocking(cfg ClusterConfig) (*Session, error) {
+	s, err := newSessionCommon(cfg)
+	if err != nil {
+		return nil, err
+	}
+
+	go func() {
+		if initErr := s.init(); initErr != nil {
+			s.sessionStateMu.Lock()
+			s.initErr = fmt.Errorf("gocql: unable to create session: %v", initErr)
+			s.sessionStateMu.Unlock()
+		}
+
+		s.readyCh <- s.initErr
+		close(s.readyCh)
+	}()
+
 	return s, nil
 }
 
@@ -404,6 +439,9 @@ func (s *Session) AwaitSchemaAgreement(ctx context.Context) error {
 	if s.cfg.disableControlConn {
 		return errNoControl
 	}
+	if err := s.Ready(); err != nil {
+		return err
+	}
 	ch := s.control.getConn()
 	return (&Iter{err: ch.conn.awaitSchemaAgreement(ctx)}).err
 }
@@ -570,11 +608,32 @@ func (s *Session) initialized() bool {
 	return initialized
 }
 
+func (s *Session) Ready() error {
+	s.sessionStateMu.RLock()
+	err := ErrSessionNotReady
+	if s.isInitialized || s.initErr != nil {
+		err = s.initErr
+	}
+	s.sessionStateMu.RUnlock()
+	return err
+}
+
+func (s *Session) WaitUntilReady() error {
+	err, ok := <-s.readyCh
+	if !ok {
+		return nil
+	}
+	return err
+}
+
 func (s *Session) executeQuery(qry *Query) (it *Iter) {
 	// fail fast
 	if s.Closed() {
 		return &Iter{err: ErrSessionClosed}
 	}
+	if err := s.Ready(); err != nil {
+		return &Iter{err: err}
+	}
 
 	iter, err := s.executor.executeQuery(qry)
 	if err != nil {
@@ -599,6 +658,8 @@ func (s *Session) KeyspaceMetadata(keyspace string) (*KeyspaceMetadata, error) {
 	// fail fast
 	if s.Closed() {
 		return nil, ErrSessionClosed
+	} else if err := s.Ready(); err != nil {
+		return nil, err
 	} else if keyspace == "" {
 		return nil, ErrNoKeyspace
 	}
@@ -611,6 +672,8 @@ func (s *Session) TabletsMetadata() (TabletInfoList, error) {
 	// fail fast
 	if s.Closed() {
 		return nil, ErrSessionClosed
+	} else if err := s.Ready(); err != nil {
+		return nil, err
 	} else if !s.tabletsRoutingV1 {
 		return nil, ErrTabletsNotUsed
 	}
@@ -798,6 +861,9 @@ func (s *Session) executeBatch(batch *Batch) *Iter {
 	if s.Closed() {
 		return &Iter{err: ErrSessionClosed}
 	}
+	if err := s.Ready(); err != nil {
+		return &Iter{err: err}
+	}
 
 	// Prevent the execution of the batch if greater than the limit
 	// Currently batches have a limit of 65536 queries.
@@ -2364,6 +2430,7 @@ var (
 	ErrKeyspaceDoesNotExist = errors.New("keyspace does not exist")
 	ErrNoMetadata           = errors.New("no metadata available")
 	ErrTabletsNotUsed       = errors.New("tablets not used")
+	ErrSessionNotReady      = errors.New("session is not ready yet")
 )
 
 type ErrProtocol struct{ error }