implement ping

synced-pool.go

@@ -358,6 +358,19 @@ func (p *SyncedPool) addRemoteHost(address string, remote *RemoteHost) error {
 	return nil
 }
 
+func (h *RemoteHost) Ping() error {
+	_, err := h.Call(Ping, Pong, nil)
+
+	if err != nil {
+		h.MissedPings++
+		log.Printf("Error pinging remote %s: %v\n, missed pings: %d", h.Host, err, h.MissedPings)
+
+	} else {
+		h.MissedPings = 0
+	}
+	return err
+}
+
 func (p *SyncedPool) AddRemote(host string) error {
 	if host == "" || p.IsKnown(host) {
 		return nil
@@ -375,11 +388,28 @@ func (p *SyncedPool) AddRemote(host string) error {
 
 	//pool := NewRemoteGrainPool(host)
 	remote := RemoteHost{
-		Client: client,
+		Client:      client,
-		//		Pool:   pool,
+		MissedPings: 0,
-		Host: host,
+		Host:        host,
 	}
+	go func() {
+		for range time.Tick(time.Second * 2) {
+			var err error
 
+			err = remote.Ping()
+			if err != nil {
+				for err != nil {
+					time.Sleep(time.Millisecond * 200)
+					err = remote.Ping()
+					if remote.MissedPings > 3 {
+						log.Printf("Error pinging remote %s: %v\n, missed pings: %d", host, err, remote.MissedPings)
+						p.RemoveHost(&remote)
+						return
+					}
+				}
+			}
+		}
+	}()
 	go func() {
 		for range client.Errors {
 			if client.ErrorCount > 3 {

tcp-client.go

@@ -48,10 +48,6 @@ func (m *TCPClient) Connect() error {
 	if m.Conn == nil {
 		connection, err := net.Dial("tcp", m.address)
 		if err != nil {
-
-			m.Errors <- err
-			m.ErrorCount++
-
 			return err
 		}
 		m.ErrorCount = 0
@@ -60,6 +56,16 @@ func (m *TCPClient) Connect() error {
 	return nil
 }
 
+func (m *TCPClient) HandleConnectionError(err error) error {
+	if err != nil {
+
+		m.Errors <- err
+		m.ErrorCount++
+
+	}
+	return err
+}
+
 func (m *TCPClient) Close() {
 	m.Conn.Close()
 }
@@ -75,10 +81,10 @@ func (m *TCPClient) SendPacket(messageType uint16, data []byte) error {
 		DataLength:  uint16(len(data)),
 	})
 	if err != nil {
-		return err
+		return m.HandleConnectionError(err)
 	}
 	_, err = m.Conn.Write(data)
-	return err
+	return m.HandleConnectionError(err)
 }
 
 // func (m *TCPClient) SendPacketFn(messageType uint16, datafn func(w io.Writer) error) error {