controller error handling

2 files changed, 35 insertions, 19 deletions

diff --git a/cmd/dolmetschctl/dolmetschctl.go b/cmd/dolmetschctl/dolmetschctl.go
index 86fde95..62b176d 100644
--- a/cmd/dolmetschctl/dolmetschctl.go
+++ b/cmd/dolmetschctl/dolmetschctl.go
@@ -70,7 +70,6 @@ func wsReader(ws *websocket.Conn, ch chan<- types.WebSocketResponseFull) {
 			io.Copy(ioutil.Discard, r) // consume all the data
 		}
 	}
-	return
 }
 
 func handleControllerEvent(ctrl *controller.Controller, ws *websocket.Conn, ev controller.Event) error {
@@ -120,7 +119,7 @@ func handleWebSocketMessage(ctrl *controller.Controller, ws *websocket.Conn, msg
 	return nil
 }
 
-func run(ctrl *controller.Controller, ws *websocket.Conn) error {
+func run(ctrl *controller.Controller, ws *websocket.Conn) (error, bool) {
 	ctrlCh := make(chan controller.Event, 100)
 	unsub := ctrl.Subscribe(ctrlCh)
 	defer close(unsub)
@@ -129,7 +128,7 @@ func run(ctrl *controller.Controller, ws *websocket.Conn) error {
 	go wsReader(ws, wsCh)
 
 	if err := ws.WriteJSON(types.WebSocketRequest{Command: "subscribe"}); err != nil {
-		return err
+		return err, true
 	}
 	log.Printf("subscribed to state changes!")
 
@@ -137,21 +136,20 @@ func run(ctrl *controller.Controller, ws *websocket.Conn) error {
 		select {
 		case ev, ok := <-ctrlCh:
 			if !ok {
-				return fmt.Errorf("controller channel was closed.")
+				return fmt.Errorf("controller channel was closed."), false
 			}
 			if err := handleControllerEvent(ctrl, ws, ev); err != nil {
-				return err
+				return err, true
 			}
 		case msg, ok := <-wsCh:
 			if !ok {
-				return fmt.Errorf("websocket channel was closed.")
+				return fmt.Errorf("websocket channel was closed."), true
 			}
 			if err := handleWebSocketMessage(ctrl, ws, msg); err != nil {
-				return err
+				return err, true
 			}
 		}
 	}
-	return nil
 }
 
 func main() {
@@ -176,8 +174,13 @@ func main() {
 			log.Printf("Error connecting to daemon: %v", err)
 		} else {
 			log.Printf("successfully conncted!")
-			err := run(c, ws)
-			log.Printf("run() returned: %v", err)
+			err, restart := run(c, ws)
+			if restart {
+				log.Printf("run() returned temporary error: %v", err)
+			} else {
+				log.Printf("run() returned fatal error: %v", err)
+				break
+			}
 		}
 		log.Printf("will retry in one second...")
 		time.Sleep(time.Second)
diff --git a/pkg/controller/controller.go b/pkg/controller/controller.go
index fb5f5ca..2aa8023 100644
--- a/pkg/controller/controller.go
+++ b/pkg/controller/controller.go
@@ -26,7 +26,7 @@ import (
 	"container/list"
 	"errors"
 	"fmt"
-	"log"
+	//	"log"
 	"strings"
 	"sync"
 
@@ -125,15 +125,14 @@ func (c *Controller) publishEvent(ev Event) {
 
 		select {
 		case <-sub.unsubscribe:
-			log.Printf("controller: removing subscriber '%v', because it has unsubscribed", sub.publish)
+			// log.Printf("controller: removing subscriber '%v', because it has unsubscribed", sub.publish)
 			close(sub.publish)
 			c.subscribers.Remove(entry)
 		default:
 			select {
 			case sub.publish <- ev:
 			default:
-				// subscriber is not responding...
-				log.Printf("controller: removing subscriber '%v', because it is not responding", sub.publish)
+				// log.Printf("controller: removing subscriber '%v', because it is not responding", sub.publish)
 				close(sub.publish)
 				c.subscribers.Remove(entry)
 			}
@@ -163,8 +162,11 @@ func (c *Controller) Init() error {
 
 	go func() {
 		for {
-			// TODO: handle Errors (reopen the device!)
-			c.handleMidiPacket(<-ch)
+			p := <-ch
+			if p.Err != nil {
+				c.Shutdown()
+			}
+			c.handleMidiPacket(p)
 		}
 	}()
 	return nil
@@ -173,9 +175,20 @@ func (c *Controller) Init() error {
 func (c *Controller) Shutdown() error {
 	if c.Dev != nil {
 		c.Dev.Close()
+		c.Dev = nil
+	}
+
+	c.subscribersLock.Lock()
+	defer c.subscribersLock.Unlock()
+
+	for entry := c.subscribers.Front(); entry != nil; entry = entry.Next() {
+		sub, ok := (entry.Value).(subscriber)
+		if !ok {
+			panic(fmt.Sprintf("controller: subscriber list element value has wrong type: %T", entry.Value))
+		}
+		close(sub.publish)
 	}
-	// TODO: also close all subscribed channels
-	//       terminate go-routine started by Init()
+	c.subscribers.Init()
 	return nil
 }
 
@@ -218,7 +231,7 @@ func (c *Controller) Subscribe(out chan<- Event) chan<- struct{} {
 	c.subscribersLock.Lock()
 	defer c.subscribersLock.Unlock()
 
-	log.Printf("controller: subscribing '%v' to events", out)
+	// log.Printf("controller: subscribing '%v' to events", out)
 	unsubscribe := make(chan struct{})
 	c.subscribers.PushBack(subscriber{publish: out, unsubscribe: unsubscribe})
 	return unsubscribe