func streamStatus(sock *websocket.Conn) {
	installed := installedPacks()
	filteredInstalled := []shared.Pack{}
	available := availablePacks()

loopinstalled:
	for _, ip := range installed {
		for ai, ap := range available {
			if ip.Installing && ip.File == ap.File {
				available[ai].Installing = true
				continue loopinstalled
			}
		}
		filteredInstalled = append(filteredInstalled, ip)
	}

	info := StatusInfo{
		Packs: PackInfo{
			Installed: filteredInstalled,
			Available: available,
		},
		Version: global.buildVersion,
	}

	err := sock.WriteJSON(controlResponse{Typ: "StatusResponse", Data: info})
	if err != nil {
		log.Printf("Error while writing status info: %v\n", err)
		return
	}
}

func loadLatestSales(websocketConnection *websocket.Conn) {
	results, err := r.Table("analytics").Filter(M{
		"type": "sale",
	}).OrderBy(
		r.Desc(r.Row.Field("data").Field("date")),
	).Limit(10).Map(func(row r.Term) interface{} {
		return row.Field("data")
	}).Run(rethinkSession)
	if err != nil {
		log.Printf("error loadLatestSales: %q", err)
	}
	defer results.Close()

	var latestSales []interface{}
	err = results.All(&latestSales)
	if err != nil {
		log.Println(err)
		return
	}

	log.Printf("sending latest %d sales\n", len(latestSales))

	socketOutgoingMessage := &SocketOutgoingMessage{
		Function: "loadLatestSales",
		Data: M{
			"type": "sale",
			"data": latestSales,
		},
	}

	err = websocketConnection.WriteJSON(socketOutgoingMessage)
	if err != nil {
		log.Println(err)
	}
}

func jionToChatRoom(uname string, ws *websocket.Conn) {

	// Join chat room.
	Join(uname, ws)
	defer Leave(uname)

	// Message receive loop.
	for {
		_, p, err := ws.ReadMessage()
		if err != nil {
			return
		}
		//=======deal with chat====
		// js, err := simplejson.NewJson(p)
		// if err != nil {
		// 	panic(err.Error())
		// }
		chat := new(models.ChatBean)
		json.Unmarshal(p, chat)

		//=====================
		if chat.Type == 1 || len(chat.To) > 0 {
			publish <- newEvent(models.EVENT_CHAT, uname, string(p))
		} else {
			publish <- newEvent(models.EVENT_MESSAGE, uname, string(p))
		}

	}
}

// NOTE :: When a guild voice server changes how do we shut this down
// properly, so a new connection can be setup without fuss?
//
// wsHeartbeat sends regular heartbeats to voice Discord so it knows the client
// is still connected.  If you do not send these heartbeats Discord will
// disconnect the websocket connection after a few seconds.
func (v *VoiceConnection) wsHeartbeat(wsConn *websocket.Conn, close <-chan struct{}, i time.Duration) {

	if close == nil || wsConn == nil {
		return
	}

	var err error
	ticker := time.NewTicker(i * time.Millisecond)
	for {
		v.log(LogDebug, "sending heartbeat packet")
		v.wsMutex.Lock()
		err = wsConn.WriteJSON(voiceHeartbeatOp{3, int(time.Now().Unix())})
		v.wsMutex.Unlock()
		if err != nil {
			v.log(LogError, "error sending heartbeat to voice endpoint %s, %s", v.endpoint, err)
			return
		}

		select {
		case <-ticker.C:
			// continue loop and send heartbeat
		case <-close:
			return
		}
	}
}

func newWSConn(conn *websocket.Conn, pendingWriteNum int) *WSConn {
	wsConn := new(WSConn)
	wsConn.conn = conn
	wsConn.writeChan = make(chan []byte, pendingWriteNum)

	go func() {
		for b := range wsConn.writeChan {
			if b == nil {
				break
			}

			err := conn.WriteMessage(websocket.BinaryMessage, b)
			if err != nil {
				break
			}
		}

		conn.Close()
		wsConn.Lock()
		wsConn.closeFlag = true
		wsConn.Unlock()
	}()

	return wsConn
}

func connectToAgent(timeout int, onReady func(*websocket.Conn)) (*websocket.Conn, error) {
	var ws *websocket.Conn
	var err error

	t := 0
	for {
		if t > timeout {
			return nil, errors.New("Connection to Agent : timeout reached")
		}

		ws, err = newClient()
		if err == nil {
			break
		}
		time.Sleep(1 * time.Second)
		t++
	}

	ready := false
	h := func(message string) error {
		err := ws.WriteControl(websocket.PongMessage, []byte(message), time.Now().Add(time.Second))
		if err != nil {
			return err
		}
		if !ready {
			ready = true
			onReady(ws)
		}
		return nil
	}
	ws.SetPingHandler(h)

	return ws, nil
}

func createPlayer(conn *websocket.Conn) {
	message, _ := json.Marshal(&Message{Title: "createPlayer", Data: connections.m[conn]})

	if err := conn.WriteMessage(websocket.TextMessage, message); err != nil {
		closeConnection(conn)
	}
}

func (w *WebsocketController) listenForIncomingData(connection *websocket.Conn, client *WebsocketClient, disconnect chan bool) {
	for {
		mt, message, err := connection.ReadMessage()

		if mt == websocket.CloseMessage {
			disconnect <- true
		}
		if _, ok := err.(*websocket.CloseError); ok {
			disconnect <- true
		}

		if err != nil {
			w.logger.Println("warning: read message error:", err)
			continue
		}
		if mt != websocket.TextMessage && mt != websocket.CloseMessage {
			w.logger.Println("warning: recieved unsupported message: ", mt, message)
			continue
		}

		if mt == websocket.TextMessage {
			w.Received() <- ReceivedData{client, message}
		}
	}
}

func handleWs(s Service, ws *websocket.Conn, subs map[uint32]*wsSub, slock *sync.Mutex) error {
	ch := make(chan []byte)
	defer close(ch)

	go func() {
		for {
			v, ok := <-ch
			if !ok {
				break
			}

			if err := ws.WriteMessage(websocket.BinaryMessage, v); err != nil {
				fmt.Printf("ws.WriteMessage: %#v\n", err)
				break
			}
		}
	}()

	for {
		mt, msg, err := ws.ReadMessage()
		if err != nil {
			fmt.Printf("ws.ReadMessage: %#v\n", err)
			break
		}

		if mt != websocket.BinaryMessage {
			continue
		}

		if err := iter(ch, msg, s, subs, slock); err != nil {
			return err
		}
	}
	return nil
}

func (w *WebsocketServer) streamWebsocket(websocketSink *websocket.WebsocketSink, websocketConnection *gorilla.Conn, register func(sinks.Sink) bool, unregister func(sinks.Sink)) {
	register(websocketSink)
	defer unregister(websocketSink)

	go websocketConnection.ReadMessage()
	server.NewKeepAlive(websocketConnection, w.keepAliveInterval).Run()
}

// writer runs in a goroutine for each connected WS client. It emits all message returned by the observer.
func writer(ws *websocket.Conn, a *ApiHandlers) {
	pingTicker := time.NewTicker(pingPeriod)
	s := a.getDBSubscriber()
	jww.INFO.Println("Opened WebSocket connection.")
	defer func(is *subscriber) {
		jww.INFO.Println("Closing WebSocket connection.")
		is.quitChan <- true
		pingTicker.Stop()
		ws.Close()
	}(s)

	for {
		select {
		case msg := <-s.bufChan:
			ws.SetWriteDeadline(time.Now().Add(writeWait))
			if err := ws.WriteJSON(msg); err != nil {
				return
			}
		case <-pingTicker.C:
			ws.SetWriteDeadline(time.Now().Add(writeWait))
			if err := ws.WriteMessage(websocket.PingMessage, []byte{}); err != nil {
				return
			}
		}
	}
}

func (app *App) registerSocket(ws *websocket.Conn, hub *Hub, toHub *Hub) {
	defer hub.RemoveMember(ws)
	hub.AddMember(ws)

	app.KeepAlive()

	from := toHub.FromLabel
	to := toHub.ToLabel
	appname := app.Name

	for {
		mtype, p, err := ws.ReadMessage()
		if err != nil {
			log.Printf("%s: Connection (%s): %s", app.Name, from, err)
			break
		}

		msg := &AppMessage{Data: p,
			Mtype: mtype,
			Id:    int(atomic.AddInt32(&msgcnt, 1)),
			To:    toHub}
		log.Printf("%s: QUEUE [%d] %s -> %s (of %d) [%s]", appname, msg.Id, from, to, len(hub.Members), string(msg.Data))
		app.Channel <- msg
	}
}

func PushMsg(ws *websocket.Conn, WSWriter chan []uint8) {

	defer ws.Close()

	for {
		msg := <-WSWriter

		if msg == nil {
			continue
		}

		// write message to client
		if err := ws.WriteMessage(websocket.TextMessage, []byte(msg)); err != nil {
			log.Println("[ErrorInfo]", err.Error())
			break
		}

		if string(msg) == "bye" {
			break
		}
	}

	endChan <- true

}

func streamResults(sock *websocket.Conn, params searchParams, limit int) {
	start := time.Now()
	count := 0
	results := make(chan searchResult)
	control := make(chan struct{}, 1)

	go find(results, control, params)
	for {
		res, ok := <-results
		if !ok {
			log.Printf("Finished request in %v\n", time.Since(start))
			return
		}

		count++
		log.Printf("Found result #%v after %v\n", count, time.Since(start))

		err := sock.WriteJSON(res)
		if err != nil {
			log.Printf("Error while writing result: %v\n", err)
			control <- struct{}{}
			return
		}

		if count >= limit {
			log.Printf("Finished request after hitting limit in %v\n", time.Since(start))
			control <- struct{}{}
			return
		}
	}
}

func wsSend(con *websocket.Conn, msgSend chan WSMsg, done, readDone, stopWS chan int) {
	defer fmt.Println("wsSend: goroutine stopped")
	seq := 1
	for {
		select {
		case <-stopWS:
			stopSend(con, done)
			return
		case <-readDone:
			stopSend(con, done)
			return
		case nextMsg, ok := <-msgSend:
			if ok {
				//send the message on the channel to the connection
				nextMsg.Seq = seq
				seq++
				fmt.Println("wsSend: sending msg", nextMsg.Type)
				j, _ := json.Marshal(nextMsg)
				fmt.Printf("wsSend: msg sent: `%s`\n", j)
				if err := con.WriteJSON(&nextMsg); err != nil {
					fmt.Println("wsSend:", err)
				}
			}
		}
	}
}

// Handle open Subscriber.
func onOpen(conn *websocket.Conn, w http.ResponseWriter, r *http.Request, sessionID string, app *app) websocketError {
	params := r.URL.Query()
	p := params.Get("protocol")

	protocol, err := strconv.Atoi(p)

	if err != nil {
		return newInvalidVersionStringFormatError()
	}

	switch {
	case strings.TrimSpace(p) == "":
		return newNoProtocolVersionSuppliedError()
	case protocol != SUPPORTED_PROTOCOL_VERSION:
		return newUnsupportedProtocolVersionError()
	case app.ApplicationDisabled:
		return newApplicationDisabledError()
	case r.TLS != nil:
		if app.OnlySSL {
			return newApplicationOnlyAccepsSSLError()
		}
	}

	// Create the new Subscriber
	connection := newConnection(sessionID, conn)
	app.Connect(connection)

	// Everything went fine. Huhu.
	if err := conn.WriteJSON(newConnectionEstablishedEvent(connection.SocketID)); err != nil {
		return newGenericReconnectImmediatelyError()
	}

	return nil
}

func (c Discord) WSInit(con *websocket.Conn, msgChan chan WSMsg) {
	//send init on wire
	p := Properties{
		OS:              "DiscordBot",
		Browser:         "discord.go",
		Device:          "console",
		Referrer:        "",
		ReferringDomain: "",
	}
	msgData := INIT{
		Token:      c.Token,
		Version:    3, // hard-coded so changes will only happen when coded in
		Properties: &p,
	}
	msg := WSMsg{
		Op:   2,
		Data: msgData,
	}
	j, _ := json.Marshal(msg)
	fmt.Printf("msgSentInit: `%s`\n", j)
	/* if err := con.WriteMessage(1, j); err != nil {
		fmt.Println("wsInit:",err)
	} */
	err := con.WriteJSON(msg)
	if err != nil {
		fmt.Println("wsInit:", err)
	}
}

func pipe_to_net(netConn net.Conn, wsConn *websocket.Conn, protocolType int, stopChan chan int) {
	for {
		select {
		case <-stopChan:
			return
		default:
			{
				_, p, err := wsConn.ReadMessage()
				if err != nil {
					stopChan <- 1
					return
				}
				if protocolType == websocket.BinaryMessage {
					_, err = netConn.Write(p)
					if err != nil {
						stopChan <- 1
						return
					}
				} else if protocolType == websocket.TextMessage {
					data := make([]byte, base64.StdEncoding.DecodedLen(len(p)))
					n, err := base64.StdEncoding.Decode(data, p)
					if err != nil {
						stopChan <- 1
						return
					}
					_, err = netConn.Write(data[:n])
					if err != nil {
						stopChan <- 1
						return
					}
				}
			}
		}
	}
}

func WebsocketSendStream(conn *websocket.Conn, r io.Reader) chan bool {
	ch := make(chan bool)

	if r == nil {
		close(ch)
		return ch
	}

	go func(conn *websocket.Conn, r io.Reader) {
		in := ReaderToChannel(r)
		for {
			buf, ok := <-in
			if !ok {
				break
			}

			w, err := conn.NextWriter(websocket.BinaryMessage)
			if err != nil {
				Debugf("Got error getting next writer %s", err)
				break
			}

			_, err = w.Write(buf)
			w.Close()
			if err != nil {
				Debugf("Got err writing %s", err)
				break
			}
		}
		conn.WriteMessage(websocket.TextMessage, []byte{})
		ch <- true
	}(conn, r)

	return ch
}

func pipe_to_ws(netConn net.Conn, wsConn *websocket.Conn, protocolType int, stopChan chan int) {
	for {
		select {
		case <-stopChan:
			return
		default:
			{
				data := make([]byte, BUFFER_SIZE)
				n, err := netConn.Read(data)
				if err != nil {
					stopChan <- 1
					return
				}

				if protocolType == websocket.BinaryMessage {
					err = wsConn.WriteMessage(protocolType, data[:n])
					if err != nil {
						stopChan <- 1
						return
					}
				} else if protocolType == websocket.TextMessage {
					encodedData := []byte(base64.StdEncoding.EncodeToString(data[:n]))
					err = wsConn.WriteMessage(protocolType, encodedData)
					if err != nil {
						stopChan <- 1
						return
					}
				}
			}
		}
	}
}