bugfix: flaky queue/Example_connectionPool

We need to wait a ready state for a queue. We also need to wait for
a success queue configuration on all instances before start a work.

Closes #278

Author: oleg-jukovec

queue/example_connection_pool_test.go

@@ -18,12 +18,12 @@ type QueueConnectionHandler struct {
 	name string
 	cfg  queue.Cfg
 
-	uuid          uuid.UUID
-	registered    bool
-	err           error
-	mutex         sync.Mutex
-	masterUpdated chan struct{}
-	masterCnt     int32
+	uuid       uuid.UUID
+	registered bool
+	err        error
+	mutex      sync.Mutex
+	updated    chan struct{}
+	masterCnt  int32
 }
 
 // QueueConnectionHandler implements the ConnectionHandler interface.
@@ -32,9 +32,9 @@ var _ connection_pool.ConnectionHandler = &QueueConnectionHandler{}
 // NewQueueConnectionHandler creates a QueueConnectionHandler object.
 func NewQueueConnectionHandler(name string, cfg queue.Cfg) *QueueConnectionHandler {
 	return &QueueConnectionHandler{
-		name:          name,
-		cfg:           cfg,
-		masterUpdated: make(chan struct{}, 10),
+		name:    name,
+		cfg:     cfg,
+		updated: make(chan struct{}, 10),
 	}
 }
 
@@ -53,14 +53,25 @@ func (h *QueueConnectionHandler) Discovered(conn *tarantool.Connection,
 	}
 
 	master := role == connection_pool.MasterRole
-	if master {
-		defer func() {
-			h.masterUpdated <- struct{}{}
-		}()
+
+	q := queue.New(conn, h.name)
+
+	// Check is queue ready to work.
+	if state, err := q.State(); err != nil {
+		h.updated <- struct{}{}
+		h.err = err
+		return err
+	} else if master && state != queue.RunningState {
+		return fmt.Errorf("queue state is not RUNNING: %d", state)
+	} else if !master && state != queue.InitState && state != queue.WaitingState {
+		return fmt.Errorf("queue state is not INIT and not WAITING: %d", state)
 	}
 
+	defer func() {
+		h.updated <- struct{}{}
+	}()
+
 	// Set up a queue module configuration for an instance.
-	q := queue.New(conn, h.name)
 	opts := queue.CfgOpts{InReplicaset: true, Ttr: 60 * time.Second}
 
 	if h.err = q.Cfg(opts); h.err != nil {
@@ -106,7 +117,7 @@ func (h *QueueConnectionHandler) Deactivated(conn *tarantool.Connection,
 
 // Closes closes a QueueConnectionHandler object.
 func (h *QueueConnectionHandler) Close() {
-	close(h.masterUpdated)
+	close(h.updated)
 }
 
 // Example demonstrates how to use the queue package with the connection_pool
@@ -155,8 +166,10 @@ func Example_connectionPool() {
 	}
 	defer connPool.Close()
 
-	// Wait for a master instance identification in the queue.
-	<-h.masterUpdated
+	// Wait for a queue initialization and master instance identification in
+	// the queue.
+	<-h.updated
+	<-h.updated
 	if h.err != nil {
 		fmt.Printf("Unable to identify in the pool: %s", h.err)
 		return
@@ -183,8 +196,10 @@ func Example_connectionPool() {
 		return
 	}
 
-	// Wait for a new master instance re-identification.
-	<-h.masterUpdated
+	// Wait for a replica instance connection and a new master instance
+	// re-identification.
+	<-h.updated
+	<-h.updated
 	h.mutex.Lock()
 	err = h.err
 	h.mutex.Unlock()