Fix unidentified client issue for TCP

Signed-off-by: Chaitanya Munukutla <[email protected]>

conn/tcp.go

@@ -11,6 +11,7 @@ func NewTCPConnection(conn *net.TCPConn) *TCPConnection {
 }
 
 func (tc *TCPConnection) WriteMessage(data []byte) error {
+	data = append(data, '\n')
 	_, err := tc.conn.Write(data)
 	return err
 }

handlers/conn.go

@@ -15,10 +15,10 @@ func handleConn(message []byte, client *broker.ConnectedClient, b *broker.Broker
 
 	client.SetId(event.ClientId)
 	b.Connect(event.ClientId, client)
-	subackEvent := &events.ConnAckEvent{
+	connackEvent := &events.ConnAckEvent{
 		Kind:     events.ConnAck,
 		Success:  true,
 		ClientId: event.ClientId,
 	}
-	return client.WriteInterface(subackEvent)
+	return client.WriteInterface(connackEvent)
 }

interfaces/tcp.go

@@ -0,0 +1,38 @@
+package interfaces
+
+import (
+	"bufio"
+	"github.com/c16a/microq/broker"
+	"github.com/c16a/microq/conn"
+	"github.com/c16a/microq/handlers"
+	"github.com/c16a/microq/storage"
+	"net"
+)
+
+func RunTcp(b *broker.Broker, storageProvider storage.Provider) error {
+	listener, err := net.ListenTCP("tcp", &net.TCPAddr{Port: 8081})
+	if err != nil {
+		return err
+	}
+
+	for {
+		c, err := listener.AcceptTCP()
+		if err != nil {
+			continue
+		}
+
+		scanner := bufio.NewScanner(c)
+
+		tcpConn := conn.NewTCPConnection(c)
+		client := broker.NewUnidentifiedClient(tcpConn)
+
+		for scanner.Scan() {
+			line := scanner.Text()
+
+			err = handlers.HandleMessage(client, b, storageProvider, []byte(line))
+			if err != nil {
+				continue
+			}
+		}
+	}
+}

interfaces/ws.go

@@ -0,0 +1,43 @@
+package interfaces
+
+import (
+	"github.com/c16a/microq/broker"
+	"github.com/c16a/microq/conn"
+	"github.com/c16a/microq/handlers"
+	"github.com/c16a/microq/storage"
+	"github.com/gorilla/websocket"
+	"log"
+	"net/http"
+)
+
+func RunWs(b *broker.Broker, storageProvider storage.Provider) {
+	var upgrader = websocket.Upgrader{}
+	http.HandleFunc("/echo", func(w http.ResponseWriter, r *http.Request) {
+		c, err := upgrader.Upgrade(w, r, nil)
+		if err != nil {
+			return
+		}
+		defer c.Close()
+
+		wsConn := conn.NewWebsocketConnection(c)
+		client := broker.NewUnidentifiedClient(wsConn)
+
+		for {
+			_, message, err := c.ReadMessage()
+			if err != nil {
+				if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseAbnormalClosure) {
+					b.Disconnect(client)
+					break
+				} else {
+					continue
+				}
+			}
+			err = handlers.HandleMessage(client, b, storageProvider, message)
+			if err != nil {
+				continue
+			}
+		}
+	})
+
+	log.Fatal(http.ListenAndServe(":8080", nil))
+}

main.go

@@ -1,87 +1,18 @@
 package main
 
 import (
-	"bufio"
 	"github.com/c16a/microq/broker"
-	"github.com/c16a/microq/conn"
-	"github.com/c16a/microq/handlers"
+	"github.com/c16a/microq/interfaces"
 	"github.com/c16a/microq/storage"
-	"github.com/gorilla/websocket"
 	"log"
-	"net"
-	"net/http"
 )
 
 func main() {
-	var storageProvider = storage.NewBadgerProvider()
-	defer storageProvider.Close()
-
 	b := broker.NewBroker()
-	go runWebSocketInterface(b, storageProvider)
-	log.Fatal(runTcpInterface(b, storageProvider))
-}
-
-func runTcpInterface(b *broker.Broker, storageProvider storage.Provider) error {
-	listener, err := net.ListenTCP("tcp", &net.TCPAddr{Port: 8081})
-	if err != nil {
-		return err
-	}
-
-	for {
-		// Handles one TCP client
-		c, err := listener.AcceptTCP()
-		if err != nil {
-			continue
-		}
-
-		scanner := bufio.NewScanner(c)
-
-		for scanner.Scan() {
-			line := scanner.Text()
-
-			tcpConn := conn.NewTCPConnection(c)
-			client := broker.NewUnidentifiedClient(tcpConn)
 
-			err = handlers.HandleMessage(client, b, storageProvider, []byte(line))
-			if err != nil {
-				continue
-			}
-		}
-	}
-}
-
-func runWebSocketInterface(b *broker.Broker, storageProvider storage.Provider) {
-	var upgrader = websocket.Upgrader{}
-	http.HandleFunc("/echo", echo(upgrader, b, storageProvider))
-
-	log.Fatal(http.ListenAndServe(":8080", nil))
-}
-
-func echo(upgrader websocket.Upgrader, b *broker.Broker, sp storage.Provider) func(http.ResponseWriter, *http.Request) {
-	return func(writer http.ResponseWriter, request *http.Request) {
-		c, err := upgrader.Upgrade(writer, request, nil)
-		if err != nil {
-			return
-		}
-		defer c.Close()
-
-		wsConn := conn.NewWebsocketConnection(c)
-		client := broker.NewUnidentifiedClient(wsConn)
+	var storageProvider = storage.NewBadgerProvider()
+	defer storageProvider.Close()
 
-		for {
-			_, message, err := c.ReadMessage()
-			if err != nil {
-				if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseAbnormalClosure) {
-					b.Disconnect(client)
-					break
-				} else {
-					continue
-				}
-			}
-			err = handlers.HandleMessage(client, b, sp, message)
-			if err != nil {
-				continue
-			}
-		}
-	}
+	go interfaces.RunWs(b, storageProvider)
+	log.Fatal(interfaces.RunTcp(b, storageProvider))
 }