// readPump pumps messages from the websocket connection to the hub.
func (c *Conn) readPump() {
	defer func() {
		hub.unregister <- c
		c.ws.Close()
	}()
	c.ws.SetReadLimit(maxMessageSize)
	c.ws.SetReadDeadline(time.Now().Add(pongWait))
	c.ws.SetPongHandler(func(string) error { c.ws.SetReadDeadline(time.Now().Add(pongWait)); return nil })
	for {
		_, message, err := c.ws.ReadMessage()
		if err != nil {
			if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway) {
				log.Printf("error: %v", err)
			}
			break
		}
		message = bytes.TrimSpace(bytes.Replace(message, newline, space, -1))

		msg := &Message{}
		err = json.Unmarshal(message, msg)
		bo := strings.Split(string(message), ",")
		target := hub.connections[bo[0]]
		if target != nil {
			target.send <- []byte(bo[1])
		}

		if err != nil {
			// byte message
			continue
		}
		//service.SendMessage(msg, string(message))
		//		hub.broadcast <- message
	}
}

// writePump pumps messages from the toplevel to the websocket connection.
func (c *Client) writePump() {
	defer func() {
		c.conn.Close()
		c.top.terminate()
	}()

	buf := make([]byte, 4*1024)
	for {
		n, err := c.top.stdout.Read(buf)
		//		log.Printf("Read %s", buf[:n])
		if n > 0 {
			err := c.conn.WriteMessage(websocket.TextMessage, buf[:n])
			if err != nil {
				if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway,
					websocket.CloseNoStatusReceived) {
					log.Printf("Writepump websocket error: %s", err)
				}

				break
			}
		} else {
			if err != nil {
				if err != io.EOF {
					log.Printf("Writepump Unexpected error while reading STDOUT from process: %s", err)
				} else {
					log.Printf("Writepump Process STDOUT closed: %s", err)
				}
				break
			}
		}
	}
}

func (h *Hijacker) handleOutput(conn *websocket.Conn, pio ProcessIO) int {
	var exitStatus int
	for {
		var output atc.HijackOutput
		err := conn.ReadJSON(&output)
		if err != nil {
			if !websocket.IsCloseError(err) && !websocket.IsUnexpectedCloseError(err) {
				fmt.Println(err)
			}
			break
		}

		if output.ExitStatus != nil {
			exitStatus = *output.ExitStatus
		} else if len(output.Error) > 0 {
			fmt.Fprintf(os.Stderr, "%s\n", ansi.Color(output.Error, "red+b"))
			exitStatus = 255
		} else if len(output.Stdout) > 0 {
			pio.Out.Write(output.Stdout)
		} else if len(output.Stderr) > 0 {
			pio.Err.Write(output.Stderr)
		}
	}

	return exitStatus
}

// readPump pumps messages from the websocket connection to the hub.
func (c *connection) readPump() {
	defer func() {
		h.unregister <- c
		c.ws.Close()
	}()

	c.ws.SetReadLimit(maxMessageSize)
	c.ws.SetReadDeadline(time.Now().Add(pongWait))
	c.ws.SetPongHandler(func(string) error {
		c.ws.SetReadDeadline(time.Now().Add(pongWait))
		return nil
	})

	for {
		_, message, err := c.ws.ReadMessage()
		fmt.Println(string(message))

		if err != nil {
			if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway) {
				log.Printf("error: %v", err)
			}
			break
		}
	}
}

func handleConnectionError(connection *Connection, err error) {
	if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseNormalClosure) {
		log.Printf("Unexpected error from connection: %q", err)
	}

	hub.Unregister(connection)
}

//Callback function that is called for new connection.
func connectionHandler(w http.ResponseWriter, r *http.Request, server *WServer) {
	conn, err := gorillaUpgrader.Upgrade(w, r, nil)
	if err != nil {
		log.Println("connectionHandler: gorillaUpgrader: ", err)
		return
	}
	defer func() {
		//dev
		log.Println("Connection close: ", conn.RemoteAddr())
		conn.Close()
	}()

	//dev
	log.Println("New connection: ", conn.RemoteAddr())

	client := Client{server.OnMessage, conn}
	defer server.OnClose(&client)
	for {
		messageType, r, err := conn.NextReader()
		if err != nil {
			if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway) {
				log.Println("Connection handler: ", err)
			}
			return
		}
		new_msg := &Message{&r, messageType}
		client.onmessage(new_msg, &client)
	}
}

// readPump pumps messages from the websocket connection to the hub.
func (c *Client) readPump() {
	defer func() {
		c.hub.unregister <- c
		c.conn.Close()
	}()
	c.conn.SetReadLimit(maxMessageSize)
	c.conn.SetReadDeadline(time.Now().Add(pongWait))
	c.conn.SetPongHandler(func(string) error { c.conn.SetReadDeadline(time.Now().Add(pongWait)); return nil })
	for {
		_, message, err := c.conn.ReadMessage()
		if err != nil {
			if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway) {
				log.Printf("error: %v", err)
			} else {
				log.Printf("Client disconnected")
			}
			break
		}
		if c.hub.read {
			c.hub.ReadBuffer.Lock()
			c.hub.ReadBuffer.Messages = append(c.hub.ReadBuffer.Messages, &message)
			c.hub.ReadBuffer.Unlock()
		}
	}
}

// readPump pumps messages from the websocket connection to the hub.
func (u *User) readPump() {
	defer func() {
		u.g.unregister <- u
		u.ws.Close()
	}()

	u.ws.SetReadLimit(maxMessageSize)
	u.ws.SetReadDeadline(time.Now().Add(pongWait))
	u.ws.SetPongHandler(func(string) error {
		u.ws.SetReadDeadline(time.Now().Add(pongWait))
		return nil
	})

	for {
		_, message, err := u.ws.ReadMessage()
		if err != nil {
			if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway) {
				// TODO: do something??
			}

			u.Log.Error(err.Error())

			return
		}

		u.Log.WithField("content", string(message)).Debug("new message")

		c := NewCommand(u, string(message))

		u.g.commands <- c
	}
}

func (this *subServer) wsReadPump(clientGone chan struct{}, ws *websocket.Conn) {
	ws.SetReadLimit(this.wsReadLimit)
	ws.SetReadDeadline(time.Now().Add(this.wsPongWait))
	ws.SetPongHandler(func(string) error {
		ws.SetReadDeadline(time.Now().Add(this.wsPongWait))
		return nil
	})

	// if kateway shutdown while there are open ws conns, the shutdown will
	// wait 1m: this.subServer.wsPongWait
	for {
		_, message, err := ws.ReadMessage()
		if err != nil {
			if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway) {
				log.Warn("%s: %v", ws.RemoteAddr(), err)
			} else {
				log.Debug("%s: %v", ws.RemoteAddr(), err)
			}

			close(clientGone)
			break
		}

		log.Debug("ws[%s] read: %s", ws.RemoteAddr(), string(message))
	}
}

// The websocket.IsUnexpectedCloseError function is useful for identifying
// application and protocol errors.
//
// This server application works with a client application running in the
// browser. The client application does not explicitly close the websocket. The
// only expected close message from the client has the code
// websocket.CloseGoingAway. All other other close messages are likely the
// result of an application or protocol error and are logged to aid debugging.
func ExampleIsUnexpectedCloseError(err error, c *websocket.Conn, req *http.Request) {
	for {
		messageType, p, err := c.ReadMessage()
		if err != nil {
			if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway) {
				log.Printf("error: %v, user-agent: %v", err, req.Header.Get("User-Agent"))
			}
			return
		}
		processMesage(messageType, p)
	}
}

// readPump pumps messages from the websocket connection to the hub.
func (s *subscription) readPump() {
	defer func() {
		Hub.unregister <- s
		s.conn.ws.Close()
	}()
	s.conn.ws.SetReadLimit(maxMessageSize)
	s.conn.ws.SetReadDeadline(time.Now().Add(pongWait))
	s.conn.ws.SetPongHandler(func(string) error {
		log.Println("[DEBUG] In pong handler", s.conn.username)
		s.conn.ws.SetReadDeadline(time.Now().Add(pongWait))
		return nil
	})
	for {
		_, message, err := s.conn.ws.ReadMessage()
		if err != nil {

			if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway) {
				log.Printf("error: %v", err)
			}
			log.Printf("error: %v", err)
			break
		}
		// Parse as JSON.
		var m Message
		err = json.Unmarshal(message, &m)
		if err != nil {
			log.Println("error unmarshaling: ", err)
			break
		}
		// Save raw data for this message for further processing, but append
		// username.
		// XXX: we'll end up unmarshalling twice. We should re-think this later.
		m.realm = s.realm
		m.From = s.conn.username
		// Remarshal to m.rawdata
		rawdata, err := json.Marshal(m)
		if err != nil {
			log.Println("Error re-marshalling: ", err)
			break
		}
		m.rawdata = rawdata
		// Pass it on to the external handler.
		Hub.handler.HandleMessage(m)
		// For right now only PrivateMT should not be broadcast.
		if m.Mtype != PrivateMT {
			Hub.broadcast <- m
		}

	}
}

// readPump pumps messages from the websocket connection to the toplevel.
func (c *Client) readPump() {
	defer func() {
		c.reg.unregister <- c
		c.conn.Close()
		c.top.terminate()
	}()

	for {
		_, message, err := c.conn.ReadMessage()
		if err != nil {
			if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway,
				websocket.CloseNoStatusReceived) {
				log.Printf("readPump error: %v", err)
			}
			break
		}
		c.top.stdin.Write(message)
	}
}

// readPump pumps messages from the websocket connection to the hub.
func (c *Conn) readPump() {
	defer func() {
		hub.unregister <- c
		c.ws.Close()
	}()
	c.ws.SetReadLimit(maxMessageSize)
	c.ws.SetReadDeadline(time.Now().Add(pongWait))
	c.ws.SetPongHandler(func(string) error { c.ws.SetReadDeadline(time.Now().Add(pongWait)); return nil })
	for {
		_, message, err := c.ws.ReadMessage()
		if err != nil {
			if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway) {
				log.Printf("error: %v", err)
			}
			break
		}
		message = bytes.TrimSpace(bytes.Replace(message, newline, space, -1))
		hub.broadcast <- message
	}
}

func (c *Client) read() {
	hub := GetHub()
	defer func() {
		hub.unregister <- c
		c.con.Close()
	}()
	c.con.SetReadLimit(maxMessageSize)
	c.con.SetReadDeadline(time.Now().Add(pongWait))
	c.con.SetPongHandler(func(string) error { c.con.SetReadDeadline(time.Now().Add(pongWait)); return nil })
	for {
		message := &SendMessage{}
		err := c.con.ReadJSON(message)
		if err != nil {
			if soc.IsUnexpectedCloseError(err, soc.CloseGoingAway) {
				log.Printf("error: %v", err)
			}
			break
		}
		message.Message = string(bytes.TrimSpace(bytes.Replace([]byte(message.Message), newline, space, -1)))
		hub.broadcast <- message
	}
}

// readMessagesLoop reads messages from the websocket connection.
func (c *wsConnection) readMessagesLoop(mode int) {
	// When disconnect
	defer func() {
		close(c.done)
		close(c.outboxMessages)
		close(c.inboxMessages)
	}()

	for {
		// Reset the read deadline.
		c.ws.SetReadDeadline(time.Now().Add(c.sc.pingTimeout))

		messageType, message, err := c.ws.ReadMessage()

		if err != nil {
			isUnexpectedError := websocket.IsUnexpectedCloseError(err, websocket.CloseNormalClosure,
				websocket.CloseGoingAway, websocket.CloseNoStatusReceived)

			if err != io.EOF && isUnexpectedError {
				log.Error("failed to read data from websocket: %v", err)
			}

			return
		}

		switch messageType {
		case mode:
			log.Debug("received message with a payload")
			c.inboxMessages <- message
		case websocket.CloseMessage:
			log.Debug("got close message")
			return
		default:
			log.Debug("got wrong message, message dropped")
		}
	}
}

// readPump pumps messages from the websocket connection to the hub.
func (c *Connection) readPump() {
	defer func() {
		c.tb.disconnected(c)
		c.ws.Close()
	}()
	c.ws.SetReadLimit(maxMessageSize)
	c.ws.SetReadDeadline(time.Now().Add(pongWait))
	c.ws.SetPongHandler(func(d string) error {
		c.ws.SetReadDeadline(time.Now().Add(pongWait))
		return nil
	})
	for {
		var event Event
		err := c.ws.ReadJSON(&event)

		if err != nil {
			if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway) {
				log.Printf("error: %v", err)
			}
			break
		}

		log.Printf("Received event %v", event)

		switch event.Type {
		case "subscribe":
			c.Subscribed(event.Channel)
		case "message":
			for _, ch := range c.tb.Channels(event.Channel) {
				ch.Received(event)
			}
		default:
			log.Printf("unknown event command %s", event.Type)
		}
	}
}

func (c *Conn) readPump() {
	defer func() {
		c.ws.Close()
	}()

	c.ws.SetReadLimit(maxMessageSize)
	c.ws.SetReadDeadline(time.Now().Add(pongWait))
	c.ws.SetPongHandler(func(string) error { c.ws.SetReadDeadline(time.Now().Add(pongWait)); return nil })
	for {
		_, message, err := c.ws.ReadMessage()
		if err != nil {
			if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway) {
				Debug(fmt.Sprintf("ERROR: %v", err))
			}
			break
		}
		message = bytes.TrimSpace(bytes.Replace(message, newline, space, -1))
		Debug(string(message))

		if c.readCallback != nil {
			c.readCallback(c, string(message))
		}
	}
}

func (i *Informer) watch() {
	const (
		// Time allowed to write a message to the peer.
		writeWait = 10 * time.Second
		// Time allowed to read the next pong message from the peer.
		pongWait = 10 * time.Second
		// Send pings to peer with this period. Must be less than pongWait.
		pingPeriod = (pongWait * 9) / 10
	)

	// write writes a message with the given message type and payload.
	writeFn := func(ws *websocket.Conn, mt int, payload []byte) error {
		ws.SetWriteDeadline(time.Now().Add(writeWait))
		return ws.WriteMessage(mt, payload)
	}

	for {
		ws, resp, err := i.wsDialer.Dial(i.wsURL, i.wsHeader)
		if err != nil {
			if err == websocket.ErrBadHandshake {
				err = fmt.Errorf("handshake failed with status %d", resp.StatusCode)
			}
			i.notifyError(err)
			continue
		}

		// TODO: Look which is the max resource limit in kubernetes (the json serialized one)
		//ws.SetReadLimit(maxResourceSize)
		ws.SetReadDeadline(time.Now().Add(pongWait))
		ws.SetPongHandler(func(string) error {
			ws.SetReadDeadline(time.Now().Add(pongWait))
			return nil
		})

		// this routine pumps messages from the hub to the websocket connection.
		go func() {
			ticker := time.NewTicker(pingPeriod)
			defer func() {
				ticker.Stop()
				ws.Close()
			}()
			for {
				select {
				case <-ticker.C:
					if err := writeFn(ws, websocket.PingMessage, []byte{}); err != nil {
						return
					}
				}
			}
		}()

	L:
		for {
			select {
			case <-i.stopChan:
				break
			default:
				v := i.rc.item()
				we := &kapi.WatchEvent{Object: v}
				if err := ws.ReadJSON(&we); err != nil {
					if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway) {
						i.notifyError(err)
					}
					break L
				} else {
					// notify watch event
					i.notify(we)
				}
			}
		}
	}
}

func (connection *Connection) ReadPump() {
	defer func() {
		HubInstance.unregister <- connection
		connection.ws.Close()
	}()
	connection.ws.SetReadLimit(consts.MaxMessageSize)
	connection.ws.SetReadDeadline(time.Now().Add(consts.PongWait))
	connection.ws.SetPongHandler(func(string) error {
		connection.ws.SetReadDeadline(time.Now().Add(consts.PongWait))
		return nil
	})
	for {
		_, message, err := connection.ws.ReadMessage()
		if err != nil {
			if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway) {
				log.Printf("error: %v", err)
			}
			break
		}
		messageRequest := &MessageRequest{}
		if unmarshalErr := json.Unmarshal(message, messageRequest); unmarshalErr != nil {
			// request type error.
			HubInstance.typeerr <- connection
			continue
		}
		connection.request = messageRequest
		switch messageRequest.RequestType {
		case consts.CpuMetric:
			// cpu metric request.
			HubInstance.cpumetric <- connection
		case consts.MemoryMetric:
			// memory metric request.
			HubInstance.memorymetric <- connection
		case consts.DiskMetric:
			// disk metric request.
			HubInstance.diskmetric <- connection
		case consts.DiskReadMetric:
			// disk read metric request.
			HubInstance.diskreadmetric <- connection
		case consts.NetworkReceiveMetric:
			// network receive metric request.
			HubInstance.networkreceivemetric <- connection
		case consts.NetworkTransmitMetric:
			// network transmitMetric:
			HubInstance.networktransmitmetric <- connection
		case consts.CombinMetric:
			// combin metric request.
			HubInstance.combinmetric <- connection
		case consts.NodeBootMetric:
			// node boot time metric request.
			HubInstance.nodebootmetric <- connection
		case consts.ProcessRunningMetric:
			// node process running metric request.
			HubInstance.processrunningmetric <- connection
		case consts.ProcessBlockedMetric:
			// node process blocked metric request.
			HubInstance.processblockedmetric <- connection
		case consts.ContainerLastSeenMetric:
			// container last seen metric request.
			HubInstance.containerlastseenmetric <- connection
		case consts.ContainerMemoryUsageBytesMetric:
			// container memory usage bytes metric request.
			HubInstance.containermemoryusagebytesmetric <- connection
		case consts.TotalRancherServicesMetric:
			// total rancher services metric request.
			HubInstance.totalrancherservicesmetric <- connection
		default:
			// request type error.
			HubInstance.typeerr <- connection
		}
	}
}