heartbeat handling works now

1 files changed, 56 insertions, 12 deletions

diff --git a/src/rhctl/playout_server.go b/src/rhctl/playout_server.go
index 506279b..5666586 100644
--- a/src/rhctl/playout_server.go
+++ b/src/rhctl/playout_server.go
@@ -24,26 +24,41 @@ package main
 import (
 	"fmt"
 	"strings"
+	"time"
 )
 
 type ServerHealth bool
 
 const (
-	Dead  ServerHealth = false
-	Alive              = true
+	ServerDead  ServerHealth = false
+	ServerAlive ServerHealth = true
 )
 
+func (s ServerHealth) String() string {
+	switch s {
+	case ServerDead:
+		return "dead"
+	case ServerAlive:
+		return "alive"
+	}
+	return "unknown"
+}
+
 type ServerStatus struct {
-	health  ServerHealth
-	channel string
+	Health  ServerHealth
+	Channel string
 }
 
 type PlayoutServer struct {
-	name      string
-	control   *SerialPort
-	heartbeat *SerialPort
-	status    ServerStatus
-	Updates   chan ServerStatus
+	name        string
+	control     *SerialPort
+	heartbeat   *SerialPort
+	hbtimeout   time.Duration
+	hbtimer     *time.Timer
+	hbthreshold uint
+	hbcnt       uint
+	status      ServerStatus
+	Updates     chan ServerStatus
 }
 
 func (srv *PlayoutServer) handleControl(data string) {
@@ -54,7 +69,7 @@ func (srv *PlayoutServer) handleControl(data string) {
 			rhl.Printf("Server(%s) sent empty channel name", srv.name)
 			return
 		}
-		srv.status.channel = data[8:]
+		srv.status.Channel = data[8:]
 		srv.Updates <- srv.status
 		return
 	}
@@ -63,12 +78,31 @@ func (srv *PlayoutServer) handleControl(data string) {
 
 func (srv *PlayoutServer) handleHeartbeat(data string) {
 	rhdl.Printf("Server(%s): got heartbeat message: %q", srv.name, data)
+	srv.hbtimer.Reset(srv.hbtimeout)
+	old := srv.status.Health
+	srv.hbcnt++
+	if srv.hbcnt < srv.hbthreshold {
+		return
+	}
+	srv.status.Health = ServerAlive
+	if old != srv.status.Health {
+		srv.Updates <- srv.status
+	}
+}
+
+func (srv *PlayoutServer) handleHBTimeout() {
+	rhl.Printf("Server(%s): heartbeat timed-out", srv.name)
+	srv.hbcnt = 0
+	srv.status.Health = ServerDead
+	srv.Updates <- srv.status
 }
 
 func (srv *PlayoutServer) Run() {
 	stop := make(chan bool)
 	srv.control.Run(stop)
 	srv.heartbeat.Run(stop)
+	srv.hbtimer = time.NewTimer(srv.hbtimeout)
+	srv.hbcnt = 0
 
 	rhdl.Printf("Server(%s) handler running...", srv.name)
 	for {
@@ -79,6 +113,8 @@ func (srv *PlayoutServer) Run() {
 			srv.handleControl(data)
 		case data := <-srv.heartbeat.rx:
 			srv.handleHeartbeat(data)
+		case <-srv.hbtimer.C:
+			srv.handleHBTimeout()
 		}
 	}
 }
@@ -86,8 +122,16 @@ func (srv *PlayoutServer) Run() {
 func ServerInit(name string, conf *Config) (srv *PlayoutServer, err error) {
 	srv = &PlayoutServer{}
 	srv.name = name
-	srv.status.health = Dead
-	srv.status.channel = ""
+	srv.hbtimeout = 3 * time.Second
+	if conf.Servers[name].HeartbeatTimeout.Duration > time.Duration(0) {
+		srv.hbtimeout = conf.Servers[name].HeartbeatTimeout.Duration
+	}
+	srv.hbthreshold = 10
+	if conf.Servers[name].HeartbeatThreshold > 0 {
+		srv.hbthreshold = conf.Servers[name].HeartbeatThreshold
+	}
+	srv.status.Health = ServerDead
+	srv.status.Channel = ""
 	srv.Updates = make(chan ServerStatus, 8)
 
 	if srv.control, err = SerialOpen(conf.Servers[name].ControlDevice, conf.Servers[name].ControlBaudrate, "\r\n"); err != nil {