Merge branch 'master' of ssh://github.com/micro/go-micro

tunnel/default.go

@@ -93,6 +93,9 @@ func (t *tun) getSession(channel, session string) (*session, bool) {
 // delSession deletes a session if it exists
 func (t *tun) delSession(channel, session string) {
 	t.Lock()
+	if s, ok := t.sessions[channel+session]; ok {
+		s.Close()
+	}
 	delete(t.sessions, channel+session)
 	t.Unlock()
 }
@@ -512,10 +515,10 @@ func (t *tun) listen(link *link) {
 			}
 
 			// try get the dialing socket
-			s, exists := t.getSession(channel, sessionId)
+			_, exists := t.getSession(channel, sessionId)
 			if exists {
-				// close and continue
-				s.Close()
+				// delete and continue
+				t.delSession(channel, sessionId)
 				continue
 			}
 			// otherwise its a session mapping of sorts

tunnel/listener.go

@@ -66,6 +66,7 @@ func (t *tunListener) process() {
 			sess, ok := conns[m.session]
 			log.Debugf("Tunnel listener received channel %s session %s exists: %t", m.channel, m.session, ok)
 			if !ok {
+				// we only process open and session types
 				switch m.typ {
 				case "open", "session":
 				default:
@@ -87,7 +88,7 @@ func (t *tunListener) process() {
 					// the link the message was received on
 					link: m.link,
 					// set the connection mode
-					mode: m.mode,
+					mode: t.session.mode,
 					// close chan
 					closed: make(chan bool),
 					// recv called by the acceptor

tunnel/session.go

@@ -413,6 +413,11 @@ func (s *session) Close() error {
 	default:
 		close(s.closed)
 
+		// don't broadcast the close for multicast
+		if s.mode != Unicast {
+			return nil
+		}
+
 		// append to backlog
 		msg := s.newMessage("close")
 		// no error response on close