Eth-ws: relaying

server/ethws_server.go

@@ -3,36 +3,179 @@ package server
 import (
 	"context"
 	"fmt"
+	"github.com/gorilla/websocket"
 	"github.com/notional-labs/subnode/config"
 	"github.com/notional-labs/subnode/state"
-	"github.com/notional-labs/subnode/utils"
 	"log"
 	"net/http"
+	"net/url"
+	"sync"
 )
 
 var ethWsServer *http.Server
+var upgrader = websocket.Upgrader{} // use default options
 
-func ethWsHandle(w http.ResponseWriter, r *http.Request) {
+func createWSClient() (*websocket.Conn, error) {
 	prunedNode := state.SelectPrunedNode(config.ProtocolTypeEthWs)
-	selectedHost := prunedNode.Backend.Eth // default to pruned node
+	selectedHost := prunedNode.Backend.EthWs // default to pruned node
+	targetEthWs, err := url.Parse(selectedHost)
+	if err != nil {
+		return nil, err
+	}
+	log.Printf("connecting to %s", targetEthWs.String())
+
+	c, _, err := websocket.DefaultDialer.Dial(targetEthWs.String(), nil)
+	if err != nil {
+		log.Println("dial err:", err)
+		return nil, err
+	}
+
+	return c, nil
+}
+
+func isClosed(ch <-chan []byte) bool {
+	select {
+	case <-ch:
+		return true
+	default:
+	}
+
+	return false
+}
+
+func closeAll(wsConServer *websocket.Conn, wsConClient *websocket.Conn, clientChannel chan []byte, serverChannel chan []byte) {
+	if !isClosed(clientChannel) {
+		close(clientChannel)
+	}
+	if !isClosed(serverChannel) {
+		close(serverChannel)
+	}
+	if wsConServer != nil {
+		wsConServer.Close()
+	}
+	if wsConClient != nil {
+		wsConClient.Close()
+	}
+}
+
+func wsClientConRelay(wsConServer *websocket.Conn, wsConClient *websocket.Conn, clientChannel chan []byte, serverChannel chan []byte, wg *sync.WaitGroup) {
+	defer wg.Done()
+	//defer close(clientChannel)
+
+	for {
+		msg := <-clientChannel // receive msg from clientChannel
+
+		// relay to server
+		log.Println("relay to server")
+		err := wsConServer.WriteMessage(websocket.TextMessage, msg)
+		if err != nil {
+			log.Println("relay to server err:", err)
+			closeAll(wsConServer, wsConClient, clientChannel, serverChannel)
+			break
+		}
+	}
+
+	log.Println("exit processing clientChannel")
+}
+
+func wsServerConRelay(wsConServer *websocket.Conn, wsConClient *websocket.Conn, clientChannel chan []byte, serverChannel chan []byte, wg *sync.WaitGroup) {
+	defer wg.Done()
+	//defer close(serverChannel)
+
+	for {
+		msg := <-serverChannel // receive msg from serverChannel
+
+		// relay to client
+		log.Println("relay to client")
+		err := wsConClient.WriteMessage(websocket.TextMessage, msg)
+		if err != nil {
+			log.Println("relay to client err:", err)
+			closeAll(wsConServer, wsConClient, clientChannel, serverChannel)
+			break
+		}
+	}
+
+	log.Println("exit processing serverChannel")
+}
+
+func wsClientHandle(wsConServer *websocket.Conn, wsConClient *websocket.Conn, clientChannel chan []byte, serverChannel chan []byte, wg *sync.WaitGroup) {
+	defer wg.Done()
+
+	for {
+		_, msg, err := wsConClient.ReadMessage()
+		if err != nil {
+			log.Println("ws-client read err:", err)
+			closeAll(wsConServer, wsConClient, clientChannel, serverChannel)
+			break
+		}
+		log.Printf("ws-client recv: %s", msg)
+		clientChannel <- msg // send msg to clientChannel
+	}
+
+	log.Println("exit ws-client")
+}
+
+func wsServerHandle(wsConServer *websocket.Conn, wsConClient *websocket.Conn, clientChannel chan []byte, serverChannel chan []byte, wg *sync.WaitGroup) {
+	defer wg.Done()
+
+	for {
+		_, msg, err := wsConServer.ReadMessage()
+		if err != nil {
+			log.Println("ws-server read err:", err)
+			closeAll(wsConServer, wsConClient, clientChannel, serverChannel)
+			break
+		}
+		log.Printf("ws-server recv: %s", msg)
+		serverChannel <- msg // send msg to serverChannel
+	}
 
-	r.Host = r.URL.Host
-	state.ProxyMapEthWs[selectedHost].ServeHTTP(w, r)
+	log.Println("exit ws-server")
+}
+
+func ethWsHandle(w http.ResponseWriter, r *http.Request) {
+	var wg sync.WaitGroup
+	var wsConServer *websocket.Conn
+	var wsConClient *websocket.Conn
+	clientChannel := make(chan []byte)
+	serverChannel := make(chan []byte)
+
+	var err error
+	wsConServer, err = upgrader.Upgrade(w, r, nil)
+	if err != nil {
+		log.Print("upgrade err:", err)
+		return
+	}
+
+	defer closeAll(wsConServer, wsConClient, clientChannel, serverChannel)
+
+	//---------------------------------
+	// ws-client
+	wsConClient, err = createWSClient()
+	if err != nil {
+		log.Print("error with createWSClient:", err)
+		return
+	}
+
+	wg.Add(1)
+	go wsClientConRelay(wsConServer, wsConClient, clientChannel, serverChannel, &wg)
+	wg.Add(1)
+	go wsServerConRelay(wsConServer, wsConClient, clientChannel, serverChannel, &wg)
+	wg.Add(1)
+	go wsClientHandle(wsConServer, wsConClient, clientChannel, serverChannel, &wg)
+	wg.Add(1)
+	go wsServerHandle(wsConServer, wsConClient, clientChannel, serverChannel, &wg)
+
+	wg.Wait()
+	log.Printf("WaitGroup counter is zero")
 }
 
 func StartEthWsServer() {
 	fmt.Println("StartEthWsServer...")
+
 	handler := func(w http.ResponseWriter, r *http.Request) {
-		if r.Method == "GET" { // URI over HTTP
-			ethWsHandle(w, r)
-		} else {
-			_ = utils.SendError(w)
-			return
-		}
+		ethWsHandle(w, r)
 	}
 
-	// handle all requests to your server using the proxy
-	//http.HandleFunc("/", handler)
 	serverMux := http.NewServeMux()
 	serverMux.HandleFunc("/", handler)
 	go func() {

state/state.go

@@ -22,15 +22,14 @@ type BackendState struct {
 }
 
 var (
-	PoolRpc       []*BackendState
-	PoolApi       []*BackendState
-	PoolGrpc      []*BackendState
-	PoolEth       []*BackendState
-	PoolEthWs     []*BackendState
-	ProxyMapRpc   = make(map[string]*httputil.ReverseProxy)
-	ProxyMapApi   = make(map[string]*httputil.ReverseProxy)
-	ProxyMapEth   = make(map[string]*httputil.ReverseProxy)
-	ProxyMapEthWs = make(map[string]*httputil.ReverseProxy)
+	PoolRpc     []*BackendState
+	PoolApi     []*BackendState
+	PoolGrpc    []*BackendState
+	PoolEth     []*BackendState
+	PoolEthWs   []*BackendState
+	ProxyMapRpc = make(map[string]*httputil.ReverseProxy)
+	ProxyMapApi = make(map[string]*httputil.ReverseProxy)
+	ProxyMapEth = make(map[string]*httputil.ReverseProxy)
 )
 
 func Init() {
@@ -40,7 +39,6 @@ func Init() {
 		rpcItem := PoolRpc[i]
 		apiItem := PoolApi[i]
 		ethItem := PoolEth[i]
-		ethWsItem := PoolEthWs[i]
 
 		// rpc
 		targetRpc, err := url.Parse(rpcItem.Name)
@@ -62,13 +60,6 @@ func Init() {
 			panic(err)
 		}
 		ProxyMapEth[ethItem.Name] = httputil.NewSingleHostReverseProxy(targetEth)
-
-		// eth-ws
-		targetEthWs, err := url.Parse(ethWsItem.Name)
-		if err != nil {
-			panic(err)
-		}
-		ProxyMapEthWs[ethWsItem.Name] = httputil.NewSingleHostReverseProxy(targetEthWs)
 	}
 }
 

test/test.config.evmos.yaml

@@ -3,7 +3,7 @@ upstream:
     api: "http://api-evmos-ia.cosmosia.notional.ventures:80"
     grpc: "grpc-evmos-ia.cosmosia.notional.ventures:9090"
     eth: "http://jsonrpc-evmos-ia.cosmosia.notional.ventures:80"
-    ethws: "http://jsonrpc-evmos-ia.cosmosia.notional.ventures:80"
+    ethws: "ws://jsonrpc-evmos-ia.cosmosia.notional.ventures:80/websocket/"
     blocks: [362880]
   - rpc: "http://rpc-evmos-archive-sub-ia.cosmosia.notional.ventures:80"
     api: "http://api-evmos-archive-sub-ia.cosmosia.notional.ventures:80"