func (c *LiveConnection) On(typ ResponseType, cb LiveListener) {
	if typ == WildcardResponse {
		c.OnError(cb)
		c.OnInvalidRequest(cb)
		c.OnRecordMessage(cb)
		c.OnHeartbeat(cb)
		c.OnSuccess(cb)
		c.OnStats(cb)
		c.OnEnd(cb)
		return
	}
	c.mu.Lock()
	c.listeners[typ] = append(c.listeners[typ], cb)
	c.mu.Unlock()
}
On adds a listener, a websocket message subscriber based on the given "typ" `ResponseType`.
Use the `WildcardResponse` to subscribe to all message types.
package websocket
import (
	"crypto/tls"
	"encoding/json"
	"fmt"
	"net"
	"net/http"
	"os"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"github.com/gorilla/websocket"
	"github.com/kataras/golog"
	conf "github.com/lensesio/lenses-go/pkg/configs"
)
type ResponseType string
const (
	WildcardResponse ResponseType = "*"
	ErrorResponse ResponseType = "ERROR"
	InvalidRequestResponse ResponseType = "INVALIDREQUEST"
	RecordMessageResponse ResponseType = "RECORD"
	HeartbeatResponse ResponseType = "HEARTBEAT"
	SuccessResponse ResponseType = "SUCCESS"
	StatsResponse ResponseType = "STATS"
	EndResponse ResponseType = "END"
)
type (
	MetaData struct {
		Timestamp int `json:"timestamp"`
		KeySize   int `json:"__keysize"`
		ValueSize int `json:"__valuesize"`
		Partition int `json:"partition"`
		Offset    int `json:"offset"`
	}
	Data struct {
		Key      json.RawMessage `json:"key"`
		Value    json.RawMessage `json:"value"`
		Metadata MetaData        `json:"metadata"`
		RowNum   int             `json:"rownum"`
	}
	LiveResponse struct {
		Type ResponseType `json:"type"`
		Data Data `json:"data"`
	}
)
type (
	Message struct {
		Token string `json:"token"`
		SQL   string `json:"sql"`
		Live  bool   `json:"live"`
		Stats int    `json:"stats"`
	}
	LiveConfiguration struct {
		Host    string `json:"host"`
		Debug   bool   `json:"debug"`
		Message Message
		HandshakeTimeout time.Duration
		ReadBufferSize, WriteBufferSize int
		TLSClientConfig *tls.Config
	}
	LiveConnection struct {
		conn   *websocket.Conn
		config LiveConfiguration
		receiveStop chan struct{}
		closed      uint32
		authToken string  
		endpoint  string  
		listeners map[ResponseType][]LiveListener
		mu        sync.RWMutex
		errors chan error  
	}
)
func OpenLiveConnection(config LiveConfiguration) (*LiveConnection, error) {
	if config.Debug {
		golog.SetLevel("debug")
	}
	if config.HandshakeTimeout == 0 {
		config.HandshakeTimeout = 45 * time.Second
	}
	config.Host = strings.Replace(config.Host, "https://", "wss://", 1)
	config.Host = strings.Replace(config.Host, "http://", "ws://", 1)
	endpoint := fmt.Sprintf("%s/api/ws/v2/sql/execute", config.Host)
	if conf.Manager.Config.GetCurrent().Insecure == true {
		config.TLSClientConfig = &tls.Config{InsecureSkipVerify: true}
	}
	c := &LiveConnection{
		config:      config,
		endpoint:    endpoint,
		receiveStop: make(chan struct{}),
		listeners:   make(map[ResponseType][]LiveListener),
		errors:      make(chan error),
	}
	return c, c.start()
}
func (c *LiveConnection) start() error {
	dialer := websocket.Dialer{
		Proxy:            http.ProxyFromEnvironment,
		HandshakeTimeout: c.config.HandshakeTimeout,
		ReadBufferSize:   c.config.ReadBufferSize,
		WriteBufferSize:  c.config.WriteBufferSize,
		TLSClientConfig:  c.config.TLSClientConfig,
	}
	conn, _, err := dialer.Dial(c.endpoint, nil)
	if err != nil {
		err = fmt.Errorf("connect failure for [%s]: %v", c.config.Host, err)
		golog.Debug(err)
		return err
	}
	err = conn.WriteJSON(c.config.Message)
	if err != nil {
		golog.Debug(err)
		return err
	}
	c.conn = conn
	go c.readLoop()
	return nil
}
func (c *LiveConnection) Wait(interruptSignal <-chan os.Signal) error {
	select {
	case <-interruptSignal:
		return c.Close()
	}
}
func (c *LiveConnection) Err() <-chan error {
	return c.errors
}
func (c *LiveConnection) sendErr(err error) {
	golog.Debug(err)
	c.errors <- err
}
func (c *LiveConnection) readLoop() {
	defer c.Close()  
	for {
		select {
		case <-c.receiveStop:
			golog.Debugf("stop receiving by signal")
			return
		default:
			resp := LiveResponse{}
			if err := c.conn.ReadJSON(&resp); err != nil {
				if _, is := err.(*net.OpError); is {
					c.sendErr(err)
					continue
				}
				c.sendErr(fmt.Errorf("live: read json: [%v]", err))
				continue
			}
			golog.Debugf("read: [%#+v]", resp)
			c.mu.RLock()
			callbacks, ok := c.listeners[resp.Type]
			c.mu.RUnlock()
			if ok {
				for _, cb := range callbacks {
					if err := cb(resp); err != nil {
						c.sendErr(err)  
					}
				}
			}
		}
	}
}
type LiveListener func(LiveResponse) error