websocket frame parsing works + controller dispatch but have to be improved

cmd/iface/main.go

@@ -16,13 +16,19 @@ func main(){
 
 	/* (2) Bind default controller */
 	err := serv.BindDefault(func(c *ws.Client, f *ws.Frame){
-		fmt.Printf("Default controller")
+		fmt.Printf("Default controller\n")
+		if f != nil {
+			fmt.Printf("Received: '%s'\n", f.Buf)
+		}
 	})
 	if err != nil { panic(err) }
 
 	/* (3) Bind to URI */
 	err = serv.Bind("/channel/./room/./", func(c *ws.Client, f *ws.Frame){
-		fmt.Printf("URI controller")
+		fmt.Printf("URI controller\n")
+		if f != nil {
+			fmt.Printf("Received: '%s'\n", f.Buf)
+		}
 	})
 	if err != nil { panic(err) }
 

internal/ws/reader/reader.go

@@ -1,52 +1,37 @@
 package reader
 
 import (
+	"fmt"
 	"io"
-	"bufio"
 )
 
-// Chunk reader
+// Maximum chunk size
-type chunkReader struct {
+const MaxChunkSize = 4096
-	reader  *bufio.Reader // the reader
-}
-
-
-// New creates a new reader
-func NewReader(r io.Reader) (reader *chunkReader) {
-
-	br, ok := r.(*bufio.Reader)
-	if !ok {
-		br = bufio.NewReader(r)
-	}
-
-	return &chunkReader{reader: br}
-
-}
-
 
 
 // Read reads a chunk of n bytes
 // err is io.EOF when done
-func (r *chunkReader) Read(n uint) ([]byte, error){
+func ReadBytes(r io.Reader, n uint) ([]byte, error){
 
-	/* (1) Read line */
+	res := make([]byte, 0, MaxChunkSize)
-	chunk := make([]byte, n)
-	read, err := r.reader.Read(chunk)
 
-	/* (2) manage errors */
+	totalRead := uint(0)
-	if err == io.EOF {
-		err = io.EOF
-	}
 
-	if err != nil {
+	// socket -> tmp( buffer)
-		return chunk, err
+	for totalRead < n {
-	}
 
-	/* (1) Manage ending chunk */
+		tmp := make([]byte, 1)
-	if uint(read) == 0 || len(chunk) == 0 {
-		return chunk, io.EOF
-	}
 
-	return chunk, nil
+		read, err := r.Read(tmp)
+		if err != nil { return nil, err }
+		if read == 0 { return nil, fmt.Errorf("Cannot read") }
+
+		totalRead += uint(1)
+
+		res = append(res, tmp[0])
+
+	}
+
+	return res, nil
 
 }

ws/frame/opcode/types.go

@@ -1,7 +1,7 @@
 package opcode
 
 import (
-	"git.xdrm.io/gws/internal/ws/frame"
+	"git.xdrm.io/gws/ws/frame"
 )
 
 // Represents OpCode Values

ws/frame/types.go

@@ -1,13 +1,19 @@
 package frame
 
+import (
+	"fmt"
+)
+
+// Represenst Frame errors
+var ErrMalFormed = fmt.Errorf("Malformed Frame")
+
+
 // Represents an OpCode
 type OpCode byte
 
-
-
 // Represents a frame metadata
 type Header struct {
-	fin bool
+	Fin bool
-	opc OpCode
+	Opc OpCode
-	msk []byte // len: 4 if set, else empty
+	Msk []byte // len: 4 if set, else empty
 }

ws/private.go

@@ -1,11 +1,16 @@
 package ws
 
 import (
+	"bytes"
+	"time"
+	"encoding/binary"
+	"git.xdrm.io/gws/ws/frame"
+	"fmt"
+	"git.xdrm.io/gws/internal/ws/reader"
 	"git.xdrm.io/gws/upgrader"
 	"net"
 )
 
-
 // dispatch finds a controller for a client
 func (s *Server) dispatch(sock net.Conn, u *upgrader.T){
 
@@ -70,9 +75,150 @@ func (s *Server) dispatch(sock net.Conn, u *upgrader.T){
 	/* (2) Add controller to client */
 	client.ctl = controller
 
-	/* (3) Launch controller routine */
+	/* (3) Launch controller routine (for new connection) */
 	go controller.fun(client, nil)
 
+	/* (4) Launch continuous frame reader */
+	go client.awaitFrame()
+
 	return
 
 }
+
+
+// awaitFrame reads a websocket frame chunk by chunk
+// for a given client
+func (c *Client) awaitFrame() {
+
+	// Get buffer
+	buf := new(bytes.Buffer)
+
+	// Get buffer frame
+	frame := new(Frame)
+
+
+	for {
+
+		var startTime int64 = time.Now().UnixNano()
+
+		// Try to read frame header
+		err := frame.ReadHeader(buf, c.soc)
+
+		if err != nil {
+			fmt.Printf(" - Header reading error: %s\n", err)
+			break
+		}
+
+
+		// Try to read frame payload
+		err = frame.ReadPayload(buf, c.soc)
+		if err != nil {
+			fmt.Printf(" - Payload reading error: %s\n", err)
+			break
+		}
+
+		fmt.Printf("+ elapsed: %.3f us\n", float32(time.Now().UnixNano()-startTime)/1e3)
+
+		// Call controller
+		c.ctl.fun(c, frame)
+
+	}
+
+	fmt.Printf(" - error\n")
+
+}
+
+
+// ReadHeader reads the frame header
+func (f *Frame) ReadHeader(buf *bytes.Buffer, s net.Conn) error{
+
+	/* (2) Byte 1: FIN and OpCode */
+	b, err := reader.ReadBytes(s, 1)
+	if err != nil { return fmt.Errorf("Cannot read byte Fin nor OpCode (%s)", err) }
+
+	f.H.Fin = b[0] & 0x80 == 0x80
+	f.H.Opc = frame.OpCode( b[0] & 0x0f )
+
+	/* (3) Byte 2: Mask and Length[0] */
+	b, err = reader.ReadBytes(s, 1)
+	if err != nil { return fmt.Errorf("Cannot read byte if has Mask nor Length (%s)", err) }
+
+	// if mask, byte array not nil
+	if b[0] & 0x80 == 0x80 {
+		f.H.Msk = []byte{}
+	}
+
+	// payload length
+	f.Len = uint64( b[0] & 0x7f )
+
+	/* (4) Extended payload */
+	if f.Len == 127 {
+
+		bx, err := reader.ReadBytes(s, 8)
+		if err != nil { return fmt.Errorf("Cannot read payload extended length of 64 bytes (%s)", err) }
+
+		f.Len = binary.BigEndian.Uint64(bx)
+
+	} else if f.Len == 126 {
+
+		bx, err := reader.ReadBytes(s, 2)
+		if err != nil { return fmt.Errorf("Cannot read payload extended length of 16 bytes (%s)", err) }
+
+		f.Len = uint64( binary.BigEndian.Uint16(bx) )
+
+	}
+
+	/* (5) Masking key */
+	if f.H.Msk != nil {
+
+		bx, err := reader.ReadBytes(s, 4)
+		if err != nil { return fmt.Errorf("Cannot read mask or 32 bytes (%s)", err) }
+
+		f.H.Msk = make([]byte, 4)
+		copy(f.H.Msk, bx)
+
+	}
+
+	// fmt.Printf("    + Header\n")
+	// fmt.Printf("     + FIN:   %t\n", f.H.Fin)
+	// fmt.Printf("     + MASK?: %t\n", f.H.Msk != nil)
+	// if f.H.Msk != nil {
+	// 	fmt.Printf("     + MASK:  %x\n", f.H.Msk)
+	// }
+	// fmt.Printf("     + LEN:   %d\n", f.Len)
+
+	return nil
+
+}
+
+
+// ReadPayload reads the frame payload
+func (f *Frame) ReadPayload(b *bytes.Buffer, s net.Conn) error{
+
+	/* (1) Read payload */
+	buf, err := reader.ReadBytes(s, uint(f.Len) )
+	if err != nil { return fmt.Errorf("Cannot read payload (%s)", err) }
+
+	f.Buf = make([]byte, 0, f.Len)
+
+	/* (2) Unmask payload */
+	if len(f.H.Msk) == 4 {
+
+
+		for i, b := range buf{
+
+			mi := i % 4 // mask index
+
+			f.Buf = append(f.Buf, b ^ f.H.Msk[mi])
+
+		}
+
+	}
+
+	// fmt.Printf("    + Payload\n")
+	// fmt.Printf("     + Read:  %d\n", len(f.Buf))
+
+	return nil
+
+}
+

ws/types.go

@@ -6,6 +6,8 @@ import (
 	"git.xdrm.io/gws/ws/frame"
 )
 
+const maxBufferLength = 4096
+
 // Represents a websocket controller callback function
 type ControllerFunc func(*Client, *Frame)
 
@@ -41,7 +43,7 @@ type Server struct {
 
 // Represents a websocket frame
 type Frame struct {
-	met frame.Header
+	H   frame.Header
-	buf []byte
+	Buf []byte
-	len uint64
+	Len uint64
 }