Merge pull request #785 from micro/link

Keep track of errors and delete beyond error count > 3

tunnel/default.go

@@ -200,6 +200,35 @@ func (t *tun) monitor() {
 		case <-t.closed:
 			return
 		case <-reconnect.C:
+			t.RLock()
+
+			var delLinks []string
+			// check the link status and purge dead links
+			for node, link := range t.links {
+				// check link status
+				switch link.Status() {
+				case "closed":
+					delLinks = append(delLinks, node)
+				case "error":
+					delLinks = append(delLinks, node)
+				}
+			}
+
+			t.RUnlock()
+
+			// delete the dead links
+			if len(delLinks) > 0 {
+				t.Lock()
+				for _, node := range delLinks {
+					log.Debugf("Tunnel deleting dead link for %s", node)
+					link := t.links[node]
+					link.Close()
+					delete(t.links, node)
+				}
+				t.Unlock()
+			}
+
+			// check current link status
 			var connect []string
 
 			// build list of unknown nodes to connect to

tunnel/link.go

@@ -31,6 +31,9 @@ type link struct {
 	lastKeepAlive time.Time
 	// channels keeps a mapping of channels and last seen
 	channels map[string]time.Time
+
+	// keep an error count on the link
+	errCount int
 }
 
 func newLink(s transport.Socket) *link {
@@ -101,11 +104,33 @@ func (l *link) Close() error {
 	return nil
 }
 
+func (l *link) Send(m *transport.Message) error {
+	err := l.Socket.Send(m)
+
+	l.Lock()
+	defer l.Unlock()
+
+	// if theres no error reset the counter
+	if err == nil {
+		l.errCount = 0
+	}
+
+	// otherwise increment the counter
+	l.errCount++
+
+	return err
+}
+
 func (l *link) Status() string {
 	select {
 	case <-l.closed:
 		return "closed"
 	default:
+		l.RLock()
+		defer l.RUnlock()
+		if l.errCount > 3 {
+			return "error"
+		}
 		return "connected"
 	}
 }