Add peer and reflector servers

Author: ben221199

main.go

@@ -2,17 +2,19 @@ package main
 
 import "fmt"
 import "lbry/daemon/dht"
+import "lbry/daemon/peer"
 import "lbry/daemon/stream"
+import "lbry/daemon/reflector"
 import "lbry/daemon/rpc"
 import "net"
-import "net/http"
 import "strconv"
 import "sync"
 
 var wg sync.WaitGroup
 
 func main() {
 	node, _ := dht.NewNode(4444)
+
 	rpcServer := rpc.CreateServer()
 	contentServer := stream.CreateServer(stream.NewManager(node))
 
@@ -24,27 +26,49 @@ func main() {
 
 	wg.Go(func() {
 		fmt.Println("Starting RPC server on port 5279.")
-		startHTTPServer(rpcServer, 5279)
+		listener, err := getTCPListener("", 5279)
+		if err != nil {
+			fmt.Println("Error when getting TCP listener.")
+		}
+		defer listener.Close()
+		rpc.StartServer(rpcServer, listener)
 	})
 
 	wg.Go(func() {
 		fmt.Println("Starting content server on port 5280.")
-		startHTTPServer(contentServer, 5280)
+		listener, err := getTCPListener("", 5280)
+		if err != nil {
+			fmt.Println("Error when getting TCP listener.")
+		}
+		defer listener.Close()
+		stream.StartServer(contentServer, listener)
+	})
+
+	wg.Go(func() {
+		fmt.Println("Starting reflector server on port 5566.")
+		listener, err := getTCPListener("", 5566)
+		if err != nil {
+			fmt.Println("Error when getting TCP listener.")
+		}
+		defer listener.Close()
+		reflector.StartServer(listener)
+	})
+
+	wg.Go(func() {
+		fmt.Println("Starting peer server on port 5567.")
+		listener, err := getTCPListener("", 5567)
+		if err != nil {
+			fmt.Println("Error when getting TCP listener.")
+		}
+		defer listener.Close()
+		peer.StartServer(listener)
 	})
 
 	wg.Wait()
 
 	fmt.Println("All servers have stopped.")
 }
 
-func startHTTPServer(rpcServer *http.Server, port int) {
-	listener, err := net.Listen("tcp", net.JoinHostPort("", strconv.Itoa(port)))
-	if err != nil && err != http.ErrServerClosed {
-		fmt.Println("Error when starting listening.")
-	}
-
-	err = rpcServer.Serve(listener)
-	if err != nil && err != http.ErrServerClosed {
-		fmt.Println("Error when starting HTTP server.")
-	}
+func getTCPListener(hostname string, port int) (net.Listener, error) {
+	return net.Listen("tcp", net.JoinHostPort(hostname, strconv.Itoa(port)))
 }

peer/server.go

@@ -0,0 +1,42 @@
+package peer
+
+import "encoding/json"
+import "fmt"
+import "io"
+import "net"
+import "time"
+
+func StartServer(listener net.Listener) {
+	for {
+		conn, err := listener.Accept()
+		if err != nil {
+			fmt.Println("Error accepting:", err)
+			continue
+		}
+		go handleConnection(conn)
+	}
+}
+
+func handleConnection(conn net.Conn) {
+	defer conn.Close()
+	conn.SetReadDeadline(time.Now().Add(10 * time.Second)) // Prevent hanging
+
+	jsonDecoder := json.NewDecoder(conn)
+	jsonEncoder := json.NewEncoder(conn)
+
+	for {
+		var data map[string]any
+
+		err := jsonDecoder.Decode(&data)
+		if err != nil {
+			if err == io.EOF {
+				fmt.Println("Client disconnected")
+				return
+			}
+			// TODO Handle error (return OR continue)
+		}
+
+		fmt.Printf("%+v\n", data)
+		jsonEncoder.Encode(map[string]any{})
+	}
+}

reflector/server.go

@@ -0,0 +1,42 @@
+package reflector
+
+import "encoding/json"
+import "fmt"
+import "io"
+import "net"
+import "time"
+
+func StartServer(listener net.Listener) {
+	for {
+		conn, err := listener.Accept()
+		if err != nil {
+			fmt.Println("Error accepting:", err)
+			continue
+		}
+		go handleConnection(conn)
+	}
+}
+
+func handleConnection(conn net.Conn) {
+	defer conn.Close()
+	conn.SetReadDeadline(time.Now().Add(10 * time.Second)) // Prevent hanging
+
+	jsonDecoder := json.NewDecoder(conn)
+	jsonEncoder := json.NewEncoder(conn)
+
+	for {
+		var data map[string]any
+
+		err := jsonDecoder.Decode(&data)
+		if err != nil {
+			if err == io.EOF {
+				fmt.Println("Client disconnected")
+				return
+			}
+			// TODO Handle error (return OR continue)
+		}
+
+		fmt.Printf("%+v\n", data)
+		jsonEncoder.Encode(map[string]any{})
+	}
+}

rpc/server.go

@@ -20,6 +20,13 @@ func CreateServer() *http.Server {
 	return &http.Server{Handler: rpcServeMux}
 }
 
+func StartServer(rpcServer *http.Server, listener net.Listener) {
+	err := rpcServer.Serve(listener)
+	if err != nil && err != http.ErrServerClosed {
+		fmt.Println("Error when starting RPC server.")
+	}
+}
+
 func sendResultResponse(w http.ResponseWriter, result any) {
 	json.NewEncoder(w).Encode(map[string]any{
 		"jsonrpc": "2.0",

stream/server.go

@@ -4,6 +4,7 @@ import "fmt"
 import "log"
 import "lbry/daemon/blob"
 import "lbry/daemon/dht"
+import "net"
 import "net/http"
 import "runtime/debug"
 import "strconv"
@@ -27,6 +28,13 @@ func CreateServer(m *Manager) *http.Server {
 	return &http.Server{Handler: contentServeMux}
 }
 
+func StartServer(contentServer *http.Server, listener net.Listener) {
+	err := contentServer.Serve(listener)
+	if err != nil && err != http.ErrServerClosed {
+		fmt.Println("Error when starting Stream server.")
+	}
+}
+
 func NewManager(dhtNode *dht.Node) *Manager {
 	return &Manager{
 		dhtNode: dhtNode,