Merge pull request #596 from Zariel/set-host-up

update host state after updating the pool

events.go

@@ -186,12 +186,13 @@ func (s *Session) handleNewNode(host net.IP, port int, waitForBinary bool) {
 	}
 
 	// should this handle token moving?
-	if existing, ok := s.ring.addHostIfMissing(hostInfo); !ok {
+	if existing, ok := s.ring.addHostIfMissing(hostInfo); ok {
 		existing.update(hostInfo)
 		hostInfo = existing
 	}
 
 	s.pool.addHost(hostInfo)
+	hostInfo.setState(NodeUp)
 
 	if s.control != nil {
 		s.hostSource.refreshRing()
@@ -211,6 +212,7 @@ func (s *Session) handleRemovedNode(ip net.IP, port int) {
 		return
 	}
 
+	host.setState(NodeDown)
 	s.pool.removeHost(addr)
 	s.ring.removeHost(addr)
 
@@ -238,8 +240,8 @@ func (s *Session) handleNodeUp(ip net.IP, port int, waitForBinary bool) {
 			time.Sleep(t)
 		}
 
-		host.setState(NodeUp)
 		s.pool.hostUp(host)
+		host.setState(NodeUp)
 		return
 	}
 
@@ -249,9 +251,7 @@ func (s *Session) handleNodeUp(ip net.IP, port int, waitForBinary bool) {
 func (s *Session) handleNodeDown(ip net.IP, port int) {
 	addr := ip.String()
 	host := s.ring.getHost(addr)
-	if host != nil {
-		host.setState(NodeDown)
-	} else {
+	if host == nil {
 		host = &HostInfo{peer: addr}
 	}
 
@@ -259,5 +259,6 @@ func (s *Session) handleNodeDown(ip net.IP, port int) {
 		return
 	}
 
+	host.setState(NodeDown)
 	s.pool.hostDown(addr)
 }

events_ccm_test.go

@@ -73,6 +73,13 @@ func TestEventNodeDownControl(t *testing.T) {
 		t.Fatal("node not removed after remove event")
 	}
 	session.pool.mu.RUnlock()
+
+	host := session.ring.getHost(node.Addr)
+	if host == nil {
+		t.Fatal("node not in metadata ring")
+	} else if host.IsUp() {
+		t.Fatalf("not not marked as down after event in metadata: %v", host)
+	}
 }
 
 func TestEventNodeDown(t *testing.T) {
@@ -107,6 +114,13 @@ func TestEventNodeDown(t *testing.T) {
 	if _, ok := poolHosts[node.Addr]; ok {
 		t.Fatal("node not removed after remove event")
 	}
+
+	host := session.ring.getHost(node.Addr)
+	if host == nil {
+		t.Fatal("node not in metadata ring")
+	} else if host.IsUp() {
+		t.Fatalf("not not marked as down after event in metadata: %v", host)
+	}
 }
 
 func TestEventNodeUp(t *testing.T) {
@@ -122,16 +136,14 @@ func TestEventNodeUp(t *testing.T) {
 
 	session := createSession(t)
 	defer session.Close()
-	poolHosts := session.pool.hostConnPools
 
 	const targetNode = "node2"
+	node := status[targetNode]
 
-	session.pool.mu.RLock()
-	_, ok := poolHosts[status[targetNode].Addr]
-	session.pool.mu.RUnlock()
+	_, ok := session.pool.getPool(node.Addr)
 	if !ok {
 		session.pool.mu.RLock()
-		t.Errorf("target pool not in connection pool: addr=%q pools=%v", status[targetNode].Addr, poolHosts)
+		t.Errorf("target pool not in connection pool: addr=%q pools=%v", status[targetNode].Addr, session.pool.hostConnPools)
 		session.pool.mu.RUnlock()
 		t.FailNow()
 	}
@@ -142,30 +154,29 @@ func TestEventNodeUp(t *testing.T) {
 
 	time.Sleep(5 * time.Second)
 
-	session.pool.mu.RLock()
-	log.Printf("poolhosts=%+v\n", poolHosts)
-	node := status[targetNode]
-
-	if _, ok := poolHosts[node.Addr]; ok {
-		session.pool.mu.RUnlock()
+	_, ok = session.pool.getPool(node.Addr)
+	if ok {
 		t.Fatal("node not removed after remove event")
 	}
-	session.pool.mu.RUnlock()
 
 	if err := ccm.NodeUp(targetNode); err != nil {
 		t.Fatal(err)
 	}
 
 	// cassandra < 2.2 needs 10 seconds to start up the binary service
-	time.Sleep(10 * time.Second)
+	time.Sleep(15 * time.Second)
 
-	session.pool.mu.RLock()
-	log.Printf("poolhosts=%+v\n", poolHosts)
-	if _, ok := poolHosts[node.Addr]; !ok {
-		session.pool.mu.RUnlock()
+	_, ok = session.pool.getPool(node.Addr)
+	if !ok {
 		t.Fatal("node not added after node added event")
 	}
-	session.pool.mu.RUnlock()
+
+	host := session.ring.getHost(node.Addr)
+	if host == nil {
+		t.Fatal("node not in metadata ring")
+	} else if !host.IsUp() {
+		t.Fatalf("not not marked as UP after event in metadata: addr=%q host=%p: %v", node.Addr, host, host)
+	}
 }
 
 func TestEventFilter(t *testing.T) {

ring_test.go

@@ -0,0 +1,35 @@
+package gocql
+
+import "testing"
+
+func TestRing_AddHostIfMissing_Missing(t *testing.T) {
+	ring := &ring{}
+
+	host := &HostInfo{peer: "test1"}
+	h1, ok := ring.addHostIfMissing(host)
+	if ok {
+		t.Fatal("host was reported as already existing")
+	} else if !h1.Equal(host) {
+		t.Fatalf("hosts not equal that are returned %v != %v", h1, host)
+	} else if h1 != host {
+		t.Fatalf("returned host same pointer: %p != %p", h1, host)
+	}
+}
+
+func TestRing_AddHostIfMissing_Existing(t *testing.T) {
+	ring := &ring{}
+
+	host := &HostInfo{peer: "test1"}
+	ring.addHostIfMissing(host)
+
+	h2 := &HostInfo{peer: "test1"}
+
+	h1, ok := ring.addHostIfMissing(h2)
+	if !ok {
+		t.Fatal("host was not reported as already existing")
+	} else if !h1.Equal(host) {
+		t.Fatalf("hosts not equal that are returned %v != %v", h1, host)
+	} else if h1 != host {
+		t.Fatalf("returned host same pointer: %p != %p", h1, host)
+	}
+}