Tests & vartious fixes for consul pipe router.

Fix a few bugs in the consul pipe router:
- Don't share a pointer
- Write nil to pipe when closing a bridge connection to ensure the connection shutdown.
- Ensure we shutdown bridge connections correctly

common/xfer/pipes.go

@@ -22,6 +22,7 @@ type pipe struct {
 	mtx             sync.Mutex
 	wg              sync.WaitGroup
 	port, starboard io.ReadWriter
+	closers         []io.Closer
 	quit            chan struct{}
 	closed          bool
 	onClose         func()
@@ -53,6 +54,9 @@ func NewPipe() Pipe {
 		}{
 			r2, w1,
 		},
+		closers: []io.Closer{
+			r1, r2, w1, w2,
+		},
 		quit: make(chan struct{}),
 	}
 }
@@ -67,6 +71,9 @@ func (p *pipe) Close() error {
 	if !p.closed {
 		p.closed = true
 		close(p.quit)
+		for _, c := range p.closers {
+			c.Close()
+		}
 		onClose = p.onClose
 	}
 	p.mtx.Unlock()
@@ -116,6 +123,10 @@ func (p *pipe) CopyToWebsocket(end io.ReadWriter, conn Websocket) error {
 				return
 			}
 
+			if p.Closed() {
+				return
+			}
+
 			if _, err := end.Write(buf); err != nil {
 				errors <- err
 				return
@@ -133,6 +144,10 @@ func (p *pipe) CopyToWebsocket(end io.ReadWriter, conn Websocket) error {
 				return
 			}
 
+			if p.Closed() {
+				return
+			}
+
 			if err := conn.WriteMessage(websocket.BinaryMessage, buf[:n]); err != nil {
 				errors <- err
 				return

common/xfer/websocket.go

@@ -90,7 +90,7 @@ func (p *pingingWebsocket) ping() {
 	defer p.writeLock.Unlock()
 	if err := p.conn.WriteControl(websocket.PingMessage, nil, mtime.Now().Add(writeWait)); err != nil {
 		log.Errorf("websocket ping error: %v", err)
-		p.Close()
+		p.conn.Close()
 		return
 	}
 	p.pinger.Reset(pingPeriod)
@@ -158,6 +158,8 @@ func (p *pingingWebsocket) ReadJSON(v interface{}) error {
 
 // Close closes the connection
 func (p *pingingWebsocket) Close() error {
+	p.writeLock.Lock()
+	defer p.writeLock.Unlock()
 	p.pinger.Stop()
 	return p.conn.Close()
 }