Merge pull request #862 from milosgajdos83/tunnel-cleanup

Cleanup of tunnel.Dial(). Clean up network channel processors

network/default.go

@@ -296,7 +296,7 @@ func (n *network) acceptNetConn(l tunnel.Listener, recv chan *transport.Message)
 }
 
 // processNetChan processes messages received on NetworkChannel
-func (n *network) processNetChan(client transport.Client, listener tunnel.Listener) {
+func (n *network) processNetChan(listener tunnel.Listener) {
 	// receive network message queue
 	recv := make(chan *transport.Message, 128)
 
@@ -629,7 +629,7 @@ func (n *network) setRouteMetric(route *router.Route) {
 }
 
 // processCtrlChan processes messages received on ControlChannel
-func (n *network) processCtrlChan(client transport.Client, listener tunnel.Listener) {
+func (n *network) processCtrlChan(listener tunnel.Listener) {
 	// receive control message queue
 	recv := make(chan *transport.Message, 128)
 
@@ -739,7 +739,7 @@ func (n *network) processCtrlChan(client transport.Client, listener tunnel.Liste
 }
 
 // advertise advertises routes to the network
-func (n *network) advertise(client transport.Client, advertChan <-chan *router.Advert) {
+func (n *network) advertise(advertChan <-chan *router.Advert) {
 	hasher := fnv.New64()
 	for {
 		select {
@@ -910,11 +910,11 @@ func (n *network) Connect() error {
 	// prune stale nodes
 	go n.prune()
 	// listen to network messages
-	go n.processNetChan(netClient, netListener)
+	go n.processNetChan(netListener)
 	// advertise service routes
-	go n.advertise(ctrlClient, advertChan)
+	go n.advertise(advertChan)
 	// accept and process routes
-	go n.processCtrlChan(ctrlClient, ctrlListener)
+	go n.processCtrlChan(ctrlListener)
 
 	n.Lock()
 	n.connected = true

tunnel/default.go

@@ -982,28 +982,32 @@ func (t *tun) Dial(channel string, opts ...DialOption) (Session, error) {
 	var links []string
 
 	// non multicast so we need to find the link
-	t.RLock()
-	for _, link := range t.links {
-		// use the link specified it its available
-		if id := options.Link; len(id) > 0 && link.id != id {
-			continue
-		}
+	if id := options.Link; id != "" {
+		t.RLock()
+		for _, link := range t.links {
+			// use the link specified it its available
+			if link.id != id {
+				continue
+			}
 
-		link.RLock()
-		_, ok := link.channels[channel]
-		link.RUnlock()
+			link.RLock()
+			_, ok := link.channels[channel]
+			link.RUnlock()
 
-		// we have at least one channel mapping
-		if ok {
-			c.discovered = true
-			links = append(links, link.id)
+			// we have at least one channel mapping
+			if ok {
+				c.discovered = true
+				links = append(links, link.id)
+			}
+		}
+		t.RUnlock()
+		// link not found
+		if len(links) == 0 {
+			// delete session and return error
+			t.delSession(c.channel, c.session)
+			return nil, ErrLinkNotFound
 		}
-	}
-	t.RUnlock()
 
-	// link not found
-	if len(links) == 0 && len(options.Link) > 0 {
-		return nil, ErrLinkNotFound
 	}
 
 	// discovered so set the link if not multicast
@@ -1028,9 +1032,11 @@ func (t *tun) Dial(channel string, opts ...DialOption) (Session, error) {
 
 		select {
 		case <-time.After(after()):
+			t.delSession(c.channel, c.session)
 			return nil, ErrDialTimeout
 		case err := <-c.errChan:
 			if err != nil {
+				t.delSession(c.channel, c.session)
 				return nil, err
 			}
 		}
@@ -1041,7 +1047,7 @@ func (t *tun) Dial(channel string, opts ...DialOption) (Session, error) {
 		dialTimeout := after()
 
 		// set a shorter delay for multicast
-		if c.mode > Unicast {
+		if c.mode != Unicast {
 			// shorten this
 			dialTimeout = time.Millisecond * 500
 		}
@@ -1057,14 +1063,15 @@ func (t *tun) Dial(channel string, opts ...DialOption) (Session, error) {
 		}
 
 		// if its multicast just go ahead because this is best effort
-		if c.mode > Unicast {
+		if c.mode != Unicast {
 			c.discovered = true
 			c.accepted = true
 			return c, nil
 		}
 
 		// otherwise return an error
 		if err != nil {
+			t.delSession(c.channel, c.session)
 			return nil, err
 		}