func (w *ws) listenWs() {
	for {
		if w.ws == nil {
			return
		}
		select {
		case <-w.kill:  
			return
		default:
			_, msg, err := w.ws.ReadMessage()
			if err != nil {
				if cl, ok := err.(*websocket.CloseError); ok {
					w.log.Errorf("close error code: %d", cl.Code)
				}
				if _, ok := err.(*net.OpError); ok {
					return
				}
				w.stop(err)
				return
			}
			w.log.Debugf("srv->ws: %s", string(msg))
			w.lock.RLock()
			if w.downstream == nil {
				w.lock.RUnlock()
				return
			}
			w.downstream <- msg
			w.lock.RUnlock()
		}
	}
}
listen on ws & fwd to listen()
package websocket
import (
	"context"
	"crypto/tls"
	"encoding/json"
	"fmt"
	"github.com/op/go-logging"
	"net"
	"net/http"
	"sync"
	"time"
	"github.com/gorilla/websocket"
)
const WS_WRITE_CAPACITY = 5000
const WS_READ_CAPACITY = 10
const KEEP_ALIVE_TIMEOUT = 10
func newWs(baseURL string, logTransport bool, log *logging.Logger) *ws {
	return &ws{
		BaseURL:      baseURL,
		downstream:   make(chan []byte, WS_READ_CAPACITY),
		quit:         make(chan error),
		kill:         make(chan interface{}),
		logTransport: logTransport,
		log:          log,
		lock:         &sync.RWMutex{},
		createTime:   time.Now(),
		writeChan:    make(chan []byte, WS_WRITE_CAPACITY),
	}
}
type ws struct {
	ws            *websocket.Conn
	lock          *sync.RWMutex
	BaseURL       string
	TLSSkipVerify bool
	downstream    chan []byte
	logTransport  bool
	log           *logging.Logger
	createTime    time.Time
	writeChan     chan []byte
	kill chan interface{}  
	quit chan error    	   
}
func (w *ws) Connect() error {
	if w.ws != nil {
		return nil  
	}
	var d = websocket.Dialer{
		Subprotocols:    []string{"p1", "p2"},
		ReadBufferSize:  1024,
		WriteBufferSize: 1024,
		Proxy:           http.ProxyFromEnvironment,
		HandshakeTimeout: time.Second * 10,
	}
	d.TLSClientConfig = &tls.Config{InsecureSkipVerify: w.TLSSkipVerify}
	w.log.Infof("connecting ws to %s", w.BaseURL)
	ws, resp, err := d.Dial(w.BaseURL, nil)
	if err != nil {
		if err == websocket.ErrBadHandshake {
			w.log.Errorf("bad handshake: status code %d", resp.StatusCode)
		}
		return err
	}
	w.ws = ws
	go w.listenWriteChannel()
	go w.listenWs()
	go w.keepAlivePinger()
	return nil
}
func (w *ws) keepAlivePinger() {
	for {
		pingTimer := time.After(time.Second * KEEP_ALIVE_TIMEOUT)
		select {
		case <-w.kill:
			return
		case <-pingTimer:
			w.writeChan <- []byte("ping")
		}
	}
}
func (w *ws) Send(ctx context.Context, msg interface{}) error {
	bs, err := json.Marshal(msg)
	if err != nil {
		return err
	}
	select {
	case <- ctx.Done():
		return ctx.Err()
	case <- w.kill:  
		return fmt.Errorf("websocket connection closed")
	default:
	}
	w.log.Debug("ws->srv: %s", string(bs))
	w.writeChan <- bs
	return nil
}
func (w *ws) Done() <-chan error {
	return w.quit
}
func (w *ws) listenWriteChannel() {
	for {
		select {
		case <-w.kill:  
			return
		case message := <- w.writeChan:
			wsWriter, err := w.ws.NextWriter(websocket.TextMessage)
			if err != nil {
				w.log.Error("Unable to provision ws connection writer: ", err)
				w.stop(err)
				return
			}
			_, err = wsWriter.Write(message)
			if err != nil {
				w.log.Error("Unable to write to ws: ", err)
				w.stop(err)
				return
			}
			if err := wsWriter.Close(); err != nil {
				w.log.Error("Unable to close ws connection writer: ", err)
				w.stop(err)
				return
			}
		}
	}
}