ws/ws/client.go

package ws

import (
	"time"
	"bufio"
	"encoding/binary"
	"git.xdrm.io/gws/internal/http/upgrade/request"
	"net"
	"fmt"
)

// Represents a client socket utility (reader, writer, ..)
type clientIO struct {
	sock    net.Conn
	reader  *bufio.Reader
	kill    chan<- *client // unregisters client
}

// Represents all channels that need a client
type clientChannelSet struct{
	receive chan Message
	send    chan *Message
}

// Represents a websocket client
type client struct {
	io         clientIO
	iface      *Client
	ch         clientChannelSet
	status     MessageError // close status ; 0 = nothing ; else -> must close
}




// Create creates a new client
func buildClient(s net.Conn, ctl ControllerSet, serverCh serverChannelSet) (*client, error){

	/* (1) Manage UPGRADE request
	---------------------------------------------------------*/
	/* (1) Parse request */
	req, _ := request.Parse(s)

	/* (3) Build response */
	res := req.BuildResponse()

	/* (4) Write into socket */
	_, err := res.Send(s)
	if err != nil {
		return nil, fmt.Errorf("Upgrade write error: %s", err)
	}

	if res.GetStatusCode() != 101 {
		s.Close()
		return nil, fmt.Errorf("Upgrade error (HTTP %d)\n", res.GetStatusCode())
	}


	/* (2) Initialise client
	---------------------------------------------------------*/
	/* (1) Get upgrade data */
	clientURI := req.GetURI()
	clientProtocol := res.GetProtocol()

	/* (2) Initialise client */
	instance := &client{
		io: clientIO{
			sock: s,
			reader: bufio.NewReader(s),
			kill:    make(chan<- *client, 1),
		},

		iface: &Client{
			Protocol:  string(clientProtocol),
			Arguments: [][]string{ []string{ clientURI } },
		},

		ch: clientChannelSet{
			receive: make(chan Message,   1),
			send:    make(chan *Message,  1),
		},
	}



	/* (3) Find controller by URI
	---------------------------------------------------------*/
	/* (1) Try to find one */
	controller, arguments := ctl.Match(clientURI);

	/* (2) If nothing found -> error */
	if controller == nil {
		return nil, fmt.Errorf("No controller found, no default controller set\n")
	}

	/* (3) Copy arguments */
	instance.iface.Arguments  = arguments



	/* (4) Launch client routines
	---------------------------------------------------------*/
	/* (1) Launch client controller */
	go controller.Fun(
		instance.iface,      // pass the client
		instance.ch.receive, // the receiver
		instance.ch.send,    // the sender
		serverCh.broadcast,  // broadcast sender
	)

	/* (2) Launch message reader */
	go instance.reader()

	/* (3) Launc writer */
	go instance.writer()

	return instance, nil

}


// reader reads and parses messages from the buffer
func (c *client) reader(){

	for {

		/* (1) If error code -> close */
		if c.status != NONE {
			c.close(c.status)
			break
		}

		/* (2) Wait for available data */
		// c.io.sock.SetReadDeadline(time.Now().Add(10*time.Microsecond))
		// _, err := c.io.reader.Peek(1)

		// // timeout -> continune checking
		// if neterr, ok := err.(net.Error); ok && neterr.Timeout() {
		// 	time.Sleep(10*time.Microsecond)
		// 	continue
		// }

		// // another error -> stop reading
		// if err != nil {
		// 	break
		// }

		// c.io.sock.SetReadDeadline(time.Time{}) // remove timeout

		/* (3) Parse message */
		msg, err := readMessage(c.io.reader)
		if err != nil {
			// fmt.Printf(" [reader] %s\n", err)
			c.close(NORMAL)
			return
		}

		/* (4) CLOSE */
		if msg.Type == CLOSE {
			c.close(NORMAL)
			return

		}

		/* (5) PING size error */
		if msg.Type == PING && msg.Size > 125 {
			c.close(PROTOCOL_ERR)
			return
		}

		/* (6) Send PONG */
		if msg.Type == PING {
			msg.Final = true
			msg.Type  = PONG
			c.ch.send <- msg
			continue
		}

		/* (7) Unknown opcode */
		if msg.Type != TEXT && msg.Type != BINARY {
			c.close(PROTOCOL_ERR)
			return
		}

		/* (7) Dispatch to receiver */
		c.ch.receive <- *msg

	}

	/* (8) close channel */
	c.close(NORMAL)

}



// writer writes into websocket
// and is triggered by client.ch.send channel
func (c *client) writer(){

	for msg := range c.ch.send {

		/* (1) If empty message -> close properly */
		if msg == nil {
			break
		}

		/* (2) Send message */
		err := msg.Send(c.io.sock)

		/* (3) Fail on error */
		if err != nil {
			fmt.Printf(" [writer] %s\n", err)
			break
		}

	}

	/* (4) proper close */
	c.close(NORMAL)

}




// close writes the error message (if needed)
// and it closes the socket
func (c *client) close(status MessageError){

	/* (1) If error status -> send close frame */
	if status != NONE {

		/* Create message */
		msg := &Message{
			Final: true,
			Type: CLOSE,
			Data: make([]byte, 8),
		}
		binary.BigEndian.PutUint16(msg.Data, uint16(status))
		msg.Data = append(msg.Data, []byte(" close")...)
		msg.Size = uint( len(msg.Data) )

		/* Send message */
		msg.Send(c.io.sock)

	}

	/* (2) Close socket */
	c.io.sock.SetReadDeadline(time.Now())
	time.Sleep(time.Second * 3)
	c.io.sock.Close()


	/* (4) Unregister */
	c.io.kill <- c

}