adding raw bytes parser (WIP)

Author: swagftw

.idea/vcs.xml

@@ -3,23 +3,22 @@ package main
 import (
 	"bufio"
 	"bytes"
-	"encoding/json"
 	"errors"
+	"flag"
 	"io"
 	"log/slog"
 	"net"
 
-	"github.com/swagftw/rex/types"
+	"github.com/swagftw/rex"
 )
 
 var daemonAddr = "0.0.0.0:8080"
 
 func main() {
-	defer func() {
-		if r := recover(); r != nil {
-			slog.Error("panic: while running client", "err", r)
-		}
-	}()
+	verbose := flag.Bool("verbose", false, "verbose logging")
+
+	flag.Parse()
+	rex.InitLogger(*verbose)
 
 	conn, err := net.Dial("tcp", daemonAddr)
 	if err != nil {
@@ -30,83 +29,70 @@ func main() {
 
 	slog.Info("connected to the daemon")
 
-	for {
-		// check if the connection is still alive
-		_, err = conn.Write([]byte{'\n'})
-		if err != nil {
-			slog.Error("connection is not alive", "err", err)
-
-			return
-		}
+	reader := bufio.NewReader(conn)
 
-		reader := bufio.NewReader(conn)
+	err = registerClient(conn)
+	if err != nil {
+		return
+	}
 
-		err = readConnection(conn, reader)
+	for {
+		err = readConnection(reader)
 		if errors.Is(err, io.EOF) {
 			return
 		}
 	}
 }
 
-func readConnection(conn net.Conn, reader *bufio.Reader) error {
-	byteData, err := reader.ReadBytes('\n')
+func registerClient(conn net.Conn) error {
+	err := writeToConn(conn, []byte("REX REGISTER\r\n"))
 	if err != nil {
-		if errors.Is(err, io.EOF) {
-			slog.Error("connection closed", "err", err)
-
-			return err
-		}
-
-		slog.Error("failed to read data", "err", err)
-
 		return err
 	}
 
-	data := new(types.Data)
-	err = json.Unmarshal(byteData, data)
-	if err != nil {
-		slog.Error("failed to unmarshal data", "err", err)
+	slog.Info("registered client")
 
-		return nil
-	}
+	return nil
+}
 
-	if data.Type != types.HeartbeatPing {
-		return nil
-	}
+func readConnection(reader *bufio.Reader) error {
+	for {
+		byteData, err := reader.ReadBytes('\n')
+		if err != nil {
+			if errors.Is(err, io.EOF) {
+				slog.Debug("connection closed", "err", err)
 
-	slog.Info("ping received")
+				return err
+			}
 
-	data = &types.Data{
+			slog.Error("failed to read data", "err", err)
 
-		Type: types.HeartbeatPong,
-	}
+			return err
+		}
 
-	jsonData, err := json.Marshal(data)
-	if err != nil {
-		slog.Error("failed to marshal data", "err", err)
+		// end of the message
+		if bytes.Equal(byteData, []byte{'\r', '\n', '\r', '\n'}) {
+			break
+		}
 
-		return nil
+		// check if the message is "ping"
+		if bytes.HasPrefix(byteData, []byte("REX PING")) {
+			continue
+		}
 	}
 
-	buf := new(bytes.Buffer)
-
-	err = json.Compact(buf, jsonData)
-	if err != nil {
-		slog.Error("failed to compact data", "err", err)
-
-		return nil
-	}
+	return nil
+}
 
-	buf.WriteByte('\n')
+func writeToConn(conn net.Conn, data []byte) error {
+	data = append(data, []byte{'\r', '\n'}...)
 
-	_, err = conn.Write(buf.Bytes())
+	_, err := conn.Write(data)
 	if err != nil {
 		slog.Error("failed to write data", "err", err)
 
-		return nil
+		return err
 	}
 
-	slog.Info("pong sent")
-
 	return nil
 }

client/main.go

@@ -6,20 +6,30 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
 	"log/slog"
 	"net"
+	"sync"
 	"time"
 
+	gonanoid "github.com/matoous/go-nanoid"
+
 	"github.com/swagftw/rex/types"
 )
 
 var port = 8080
 
+var ActiveClients = sync.Map{}
+var clientLock = sync.RWMutex{}
+
 func StartListener() (net.Listener, error) {
-	slog.Info("starting listener...", "port", port)
-	listener, err := net.Listen("tcp", "0.0.0.0:"+fmt.Sprintf("%d", port))
+	slog.Info("starting heartbeat...", "port", port)
+
+	hostPort := net.JoinHostPort("0.0.0.0", fmt.Sprintf("%d", port))
+
+	listener, err := net.Listen("tcp", hostPort)
 	if err != nil {
-		slog.Error("failed to start listener", "err", err)
+		slog.Error("failed to start heartbeat", "err", err)
 
 		return nil, err
 	}
@@ -43,10 +53,12 @@ func StartHeartbeat() error {
 			continue
 		}
 
-		// write ping messages to the client
-		ticker := time.NewTicker(time.Second * 2)
+		go func(conn net.Conn) {
+			// write ping messages to the client
+			ticker := time.NewTicker(time.Second * 2)
+
+			defer ticker.Stop()
 
-		go func(conn net.Conn, ticker *time.Ticker) {
 			defer func() {
 				if r := recover(); r != nil {
 					slog.Error("panic: while sending ping", "err", r)
@@ -62,42 +74,15 @@ func StartHeartbeat() error {
 
 			for {
 				if <-ticker.C; true {
-					data := types.Data{
-						Type: types.HeartbeatPing,
-					}
-
-					buf := &bytes.Buffer{}
-
-					jsonData, err := json.Marshal(data)
+					err = PingClient(conn)
 					if err != nil {
-						slog.Error("failed to marshal data", "err", err)
-
-						panic(err)
-					}
-
-					err = json.Compact(buf, jsonData)
-					if err != nil {
-						slog.Error("failed to compact data", "err", err)
-
-						panic(err)
-					}
-
-					buf.WriteByte('\n')
-
-					_, err = conn.Write(buf.Bytes())
-					if err != nil {
-						ticker.Stop()
-
-						slog.Error("failed to send ping", "err", err)
-
 						return
 					}
-
-					slog.Info("ping sent")
 				}
 			}
-		}(conn, ticker)
+		}(conn)
 
+		// read from connection
 		go func(conn net.Conn) {
 			defer func() {
 				if r := recover(); r != nil {
@@ -119,18 +104,16 @@ func StartHeartbeat() error {
 
 				buf, err = reader.ReadBytes('\n')
 				if err != nil {
-					slog.Error("failed to read the bytes", "err", err)
-
-					if errors.Is(err, net.ErrClosed) {
+					if errors.Is(err, io.EOF) {
 						return
 					}
 
+					slog.Error("failed to read the bytes", "err", err)
+
 					return
 				}
 
-				if len(buf) < 1 || buf[0] == '\n' {
-					continue
-				}
+				bytes.TrimSuffix(buf, []byte{'\r'})
 
 				cd := new(types.Data)
 
@@ -142,10 +125,11 @@ func StartHeartbeat() error {
 				}
 
 				switch cd.Type {
-				case types.HeartbeatPong:
-					slog.Info("pong received")
 				case types.RegisterClient:
-					registerClient(cd)
+					err = registerClient(conn)
+					if err != nil {
+						return
+					}
 				case types.Close:
 					return
 				default:
@@ -156,6 +140,79 @@ func StartHeartbeat() error {
 	}
 }
 
-func registerClient(cd *types.Data) {
-	slog.Info("client registered", "data", cd)
+func PingClient(conn net.Conn) error {
+	data := types.Data{
+		Type: types.HeartbeatPing,
+	}
+
+	buf := &bytes.Buffer{}
+
+	jsonData, err := json.Marshal(data)
+	if err != nil {
+		slog.Error("failed to marshal data", "err", err)
+
+		return err
+	}
+
+	err = json.Compact(buf, jsonData)
+	if err != nil {
+		slog.Error("failed to compact data", "err", err)
+
+		return err
+	}
+
+	buf.WriteByte('\n')
+
+	_, err = conn.Write(buf.Bytes())
+	if err != nil {
+		if errors.Is(err, net.ErrClosed) {
+			return err
+		}
+
+		slog.Error("failed to send ping", "err", err)
+
+		return err
+	}
+
+	slog.Debug("ping sent")
+
+	return nil
+}
+
+func registerClient(conn net.Conn) error {
+	addr := conn.RemoteAddr().String()
+
+	slog.Info("client connected", "addr", addr)
+
+	id := generateID()
+
+	ActiveClients.Store(id, conn)
+
+	data := &types.Data{
+		Type: types.RegisterClient,
+	}
+
+	jsonBytes, err := json.Marshal(data)
+	if err != nil {
+		slog.Error("failed to marshal data", "err", err)
+
+		return err
+	}
+
+	jsonBytes = append(jsonBytes, []byte{'\r', '\n'}...)
+
+	_, err = conn.Write(jsonBytes)
+	if err != nil {
+		slog.Error("failed to write data", "err", err)
+
+		return err
+	}
+
+	return nil
+}
+
+var chars = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz"
+
+func generateID() string {
+	return gonanoid.MustGenerate(chars, 6)
 }