prevent a channel from blocking all others by using a ring buffer

channel.go

@@ -26,12 +26,12 @@ func randomByte() (byte, error) {
 // For instance, a TCP client endpoint creates a single channel, while a TCP
 // server endpoint creates a channel for each incoming connection.
 type Channel struct {
-	e       Endpoint
-	label   string
-	rwc     io.Closer
-	n       *Node
-	frw     *frame.ReadWriter
-	running bool
+	e           Endpoint
+	label       string
+	rwc         io.Closer
+	n           *Node
+	frw         *frame.ReadWriter
+	running     bool
 	writeBuffer *ringbuffer.RingBuffer
 
 	// in
@@ -69,13 +69,13 @@ func newChannel(n *Node, e Endpoint, label string, rwc io.ReadWriteCloser) (*Cha
 	}
 
 	return &Channel{
-		e:         e,
-		label:     label,
-		rwc:       rwc,
-		n:         n,
-		frw:       frw,
+		e:           e,
+		label:       label,
+		rwc:         rwc,
+		n:           n,
+		frw:         frw,
 		writeBuffer: writeBuffer,
-		terminate: make(chan struct{}),
+		terminate:   make(chan struct{}),
 	}, nil
 }
 
@@ -104,25 +104,21 @@ func (ch *Channel) run() {
 
 	select {
 	case <-readerDone:
-		ch.n.pushEvent(&EventChannelClose{ch})
-		ch.n.closeChannel(ch)
-
-		<-ch.terminate
+		ch.rwc.Close()
 
 		ch.writeBuffer.Close()
 		<-writerDone
 
-		ch.rwc.Close()
-
 	case <-ch.terminate:
-		ch.n.pushEvent(&EventChannelClose{ch})
-
 		ch.writeBuffer.Close()
 		<-writerDone
 
 		ch.rwc.Close()
 		<-readerDone
 	}
+
+	ch.n.pushEvent(&EventChannelClose{ch})
+	ch.n.closeChannel(ch)
 }
 
 func (ch *Channel) runReader(readerDone chan struct{}) {

node.go

@@ -105,11 +105,11 @@ type Node struct {
 	chWriteTo      chan writeToReq
 	chWriteAll     chan interface{}
 	chWriteExcept  chan writeExceptReq
-	terminate    chan struct{}
+	terminate      chan struct{}
 
 	// out
 	chEvent chan Event
-	done   chan struct{}
+	done    chan struct{}
 }
 
 // NewNode allocates a Node. See NodeConf for the options.
@@ -169,13 +169,13 @@ func NewNode(conf NodeConf) (*Node, error) {
 		dialectRW:        dialectRW,
 		channelAccepters: make(map[*channelAccepter]struct{}),
 		channels:         make(map[*Channel]struct{}),
-		chNewChannel:       make(chan *Channel),
-		chCloseChannel:     make(chan *Channel),
-		chWriteTo:          make(chan writeToReq),
-		chWriteAll:         make(chan interface{}),
-		chWriteExcept:      make(chan writeExceptReq),
+		chNewChannel:     make(chan *Channel),
+		chCloseChannel:   make(chan *Channel),
+		chWriteTo:        make(chan writeToReq),
+		chWriteAll:       make(chan interface{}),
+		chWriteExcept:    make(chan writeExceptReq),
 		terminate:        make(chan struct{}),
-		chEvent:           make(chan Event),
+		chEvent:          make(chan Event),
 		done:             make(chan struct{}),
 	}
 
@@ -262,7 +262,6 @@ outer:
 
 		case ch := <-n.chCloseChannel:
 			delete(n.channels, ch)
-			ch.close()
 
 		case req := <-n.chWriteTo:
 			if _, ok := n.channels[req.ch]; !ok {