Merge ccfd63b84f into 69168ff062

Add ICE servers and local IP address returned by the API to fix connectivity issues behind NAT

cloud.go

@@ -7,13 +7,14 @@ import (
 	"fmt"
 	"net/http"
 	"net/url"
-	"github.com/coder/websocket/wsjson"
 	"time"
 
+	"github.com/coder/websocket/wsjson"
+
 	"github.com/coreos/go-oidc/v3/oidc"
 
-	"github.com/gin-gonic/gin"
 	"github.com/coder/websocket"
+	"github.com/gin-gonic/gin"
 )
 
 type CloudRegisterRequest struct {
@@ -192,7 +193,11 @@ func handleSessionRequest(ctx context.Context, c *websocket.Conn, req WebRTCSess
 		return fmt.Errorf("google identity mismatch")
 	}
 
-	session, err := newSession()
+	session, err := newSession(SessionConfig{
+		ICEServers: req.ICEServers,
+		LocalIP:    req.IP,
+		IsCloud:    true,
+	})
 	if err != nil {
 		_ = wsjson.Write(context.Background(), c, gin.H{"error": err})
 		return err

internal/jsonrpc/router.go

@@ -0,0 +1,179 @@
+package jsonrpc
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"reflect"
+)
+
+type JSONRPCRouter struct {
+	writer io.Writer
+
+	handlers map[string]*RPCHandler
+}
+
+func NewJSONRPCRouter(writer io.Writer, handlers map[string]*RPCHandler) *JSONRPCRouter {
+	return &JSONRPCRouter{
+		writer:   writer,
+		handlers: handlers,
+	}
+}
+
+func (s *JSONRPCRouter) HandleMessage(data []byte) error {
+	var request JSONRPCRequest
+	err := json.Unmarshal(data, &request)
+	if err != nil {
+		errorResponse := JSONRPCResponse{
+			JSONRPC: "2.0",
+			Error: map[string]interface{}{
+				"code":    -32700,
+				"message": "Parse error",
+			},
+			ID: 0,
+		}
+		return s.writeResponse(errorResponse)
+	}
+
+	//log.Printf("Received RPC request: Method=%s, Params=%v, ID=%d", request.Method, request.Params, request.ID)
+	handler, ok := s.handlers[request.Method]
+	if !ok {
+		errorResponse := JSONRPCResponse{
+			JSONRPC: "2.0",
+			Error: map[string]interface{}{
+				"code":    -32601,
+				"message": "Method not found",
+			},
+			ID: request.ID,
+		}
+		return s.writeResponse(errorResponse)
+	}
+
+	result, err := callRPCHandler(handler, request.Params)
+	if err != nil {
+		errorResponse := JSONRPCResponse{
+			JSONRPC: "2.0",
+			Error: map[string]interface{}{
+				"code":    -32603,
+				"message": "Internal error",
+				"data":    err.Error(),
+			},
+			ID: request.ID,
+		}
+		return s.writeResponse(errorResponse)
+	}
+
+	response := JSONRPCResponse{
+		JSONRPC: "2.0",
+		Result:  result,
+		ID:      request.ID,
+	}
+	return s.writeResponse(response)
+}
+
+func (s *JSONRPCRouter) writeResponse(response JSONRPCResponse) error {
+	responseBytes, err := json.Marshal(response)
+	if err != nil {
+		return err
+	}
+	_, err = s.writer.Write(responseBytes)
+	return err
+}
+
+func callRPCHandler(handler *RPCHandler, params map[string]interface{}) (interface{}, error) {
+	handlerValue := reflect.ValueOf(handler.Func)
+	handlerType := handlerValue.Type()
+
+	if handlerType.Kind() != reflect.Func {
+		return nil, errors.New("handler is not a function")
+	}
+
+	numParams := handlerType.NumIn()
+	args := make([]reflect.Value, numParams)
+	// Get the parameter names from the RPCHandler
+	paramNames := handler.Params
+
+	if len(paramNames) != numParams {
+		return nil, errors.New("mismatch between handler parameters and defined parameter names")
+	}
+
+	for i := 0; i < numParams; i++ {
+		paramType := handlerType.In(i)
+		paramName := paramNames[i]
+		paramValue, ok := params[paramName]
+		if !ok {
+			return nil, errors.New("missing parameter: " + paramName)
+		}
+
+		convertedValue := reflect.ValueOf(paramValue)
+		if !convertedValue.Type().ConvertibleTo(paramType) {
+			if paramType.Kind() == reflect.Slice && (convertedValue.Kind() == reflect.Slice || convertedValue.Kind() == reflect.Array) {
+				newSlice := reflect.MakeSlice(paramType, convertedValue.Len(), convertedValue.Len())
+				for j := 0; j < convertedValue.Len(); j++ {
+					elemValue := convertedValue.Index(j)
+					if elemValue.Kind() == reflect.Interface {
+						elemValue = elemValue.Elem()
+					}
+					if !elemValue.Type().ConvertibleTo(paramType.Elem()) {
+						// Handle float64 to uint8 conversion
+						if elemValue.Kind() == reflect.Float64 && paramType.Elem().Kind() == reflect.Uint8 {
+							intValue := int(elemValue.Float())
+							if intValue < 0 || intValue > 255 {
+								return nil, fmt.Errorf("value out of range for uint8: %v", intValue)
+							}
+							newSlice.Index(j).SetUint(uint64(intValue))
+						} else {
+							fromType := elemValue.Type()
+							toType := paramType.Elem()
+							return nil, fmt.Errorf("invalid element type in slice for parameter %s: from %v to %v", paramName, fromType, toType)
+						}
+					} else {
+						newSlice.Index(j).Set(elemValue.Convert(paramType.Elem()))
+					}
+				}
+				args[i] = newSlice
+			} else if paramType.Kind() == reflect.Struct && convertedValue.Kind() == reflect.Map {
+				jsonData, err := json.Marshal(convertedValue.Interface())
+				if err != nil {
+					return nil, fmt.Errorf("failed to marshal map to JSON: %v", err)
+				}
+
+				newStruct := reflect.New(paramType).Interface()
+				if err := json.Unmarshal(jsonData, newStruct); err != nil {
+					return nil, fmt.Errorf("failed to unmarshal JSON into struct: %v", err)
+				}
+				args[i] = reflect.ValueOf(newStruct).Elem()
+			} else {
+				return nil, fmt.Errorf("invalid parameter type for: %s", paramName)
+			}
+		} else {
+			args[i] = convertedValue.Convert(paramType)
+		}
+	}
+
+	results := handlerValue.Call(args)
+
+	if len(results) == 0 {
+		return nil, nil
+	}
+
+	if len(results) == 1 {
+		if results[0].Type().Implements(reflect.TypeOf((*error)(nil)).Elem()) {
+			if !results[0].IsNil() {
+				return nil, results[0].Interface().(error)
+			}
+			return nil, nil
+		}
+		return results[0].Interface(), nil
+	}
+
+	if len(results) == 2 && results[1].Type().Implements(reflect.TypeOf((*error)(nil)).Elem()) {
+		if !results[1].IsNil() {
+			return nil, results[1].Interface().(error)
+		}
+		return results[0].Interface(), nil
+	}
+
+	return nil, errors.New("unexpected return values from handler")
+}

internal/jsonrpc/types.go

@@ -0,0 +1,26 @@
+package jsonrpc
+
+type JSONRPCRequest struct {
+	JSONRPC string                 `json:"jsonrpc"`
+	Method  string                 `json:"method"`
+	Params  map[string]interface{} `json:"params,omitempty"`
+	ID      interface{}            `json:"id,omitempty"`
+}
+
+type JSONRPCResponse struct {
+	JSONRPC string      `json:"jsonrpc"`
+	Result  interface{} `json:"result,omitempty"`
+	Error   interface{} `json:"error,omitempty"`
+	ID      interface{} `json:"id"`
+}
+
+type JSONRPCEvent struct {
+	JSONRPC string      `json:"jsonrpc"`
+	Method  string      `json:"method"`
+	Params  interface{} `json:"params,omitempty"`
+}
+
+type RPCHandler struct {
+	Func   interface{}
+	Params []string
+}

jsonrpc.go

@@ -5,50 +5,44 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"kvm/internal/jsonrpc"
 	"log"
 	"os"
 	"os/exec"
 	"path/filepath"
-	"reflect"
 
 	"github.com/pion/webrtc/v4"
 )
 
-type JSONRPCRequest struct {
-	JSONRPC string                 `json:"jsonrpc"`
-	Method  string                 `json:"method"`
-	Params  map[string]interface{} `json:"params,omitempty"`
-	ID      interface{}            `json:"id,omitempty"`
+type DataChannelWriter struct {
+	dataChannel *webrtc.DataChannel
 }
 
-type JSONRPCResponse struct {
-	JSONRPC string      `json:"jsonrpc"`
-	Result  interface{} `json:"result,omitempty"`
-	Error   interface{} `json:"error,omitempty"`
-	ID      interface{} `json:"id"`
-}
-
-type JSONRPCEvent struct {
-	JSONRPC string      `json:"jsonrpc"`
-	Method  string      `json:"method"`
-	Params  interface{} `json:"params,omitempty"`
-}
-
-func writeJSONRPCResponse(response JSONRPCResponse, session *Session) {
-	responseBytes, err := json.Marshal(response)
-	if err != nil {
-		log.Println("Error marshalling JSONRPC response:", err)
-		return
+func NewDataChannelWriter(dataChannel *webrtc.DataChannel) *DataChannelWriter {
+	return &DataChannelWriter{
+		dataChannel: dataChannel,
 	}
-	err = session.RPCChannel.SendText(string(responseBytes))
+}
+
+func (w *DataChannelWriter) Write(data []byte) (int, error) {
+	err := w.dataChannel.SendText(string(data))
 	if err != nil {
 		log.Println("Error sending JSONRPC response:", err)
-		return
+		return 0, err
 	}
+	return len(data), nil
 }
 
+func NewDataChannelJsonRpcRouter(dataChannel *webrtc.DataChannel) *jsonrpc.JSONRPCRouter {
+	return jsonrpc.NewJSONRPCRouter(
+		NewDataChannelWriter(dataChannel),
+		rpcHandlers,
+	)
+}
+
+// TODO: embed this into the session's rpc server
 func writeJSONRPCEvent(event string, params interface{}, session *Session) {
-	request := JSONRPCEvent{
+	request := jsonrpc.JSONRPCEvent{
 		JSONRPC: "2.0",
 		Method:  event,
 		Params:  params,
@@ -69,60 +63,6 @@ func writeJSONRPCEvent(event string, params interface{}, session *Session) {
 	}
 }
 
-func onRPCMessage(message webrtc.DataChannelMessage, session *Session) {
-	var request JSONRPCRequest
-	err := json.Unmarshal(message.Data, &request)
-	if err != nil {
-		errorResponse := JSONRPCResponse{
-			JSONRPC: "2.0",
-			Error: map[string]interface{}{
-				"code":    -32700,
-				"message": "Parse error",
-			},
-			ID: 0,
-		}
-		writeJSONRPCResponse(errorResponse, session)
-		return
-	}
-
-	//log.Printf("Received RPC request: Method=%s, Params=%v, ID=%d", request.Method, request.Params, request.ID)
-	handler, ok := rpcHandlers[request.Method]
-	if !ok {
-		errorResponse := JSONRPCResponse{
-			JSONRPC: "2.0",
-			Error: map[string]interface{}{
-				"code":    -32601,
-				"message": "Method not found",
-			},
-			ID: request.ID,
-		}
-		writeJSONRPCResponse(errorResponse, session)
-		return
-	}
-
-	result, err := callRPCHandler(handler, request.Params)
-	if err != nil {
-		errorResponse := JSONRPCResponse{
-			JSONRPC: "2.0",
-			Error: map[string]interface{}{
-				"code":    -32603,
-				"message": "Internal error",
-				"data":    err.Error(),
-			},
-			ID: request.ID,
-		}
-		writeJSONRPCResponse(errorResponse, session)
-		return
-	}
-
-	response := JSONRPCResponse{
-		JSONRPC: "2.0",
-		Result:  result,
-		ID:      request.ID,
-	}
-	writeJSONRPCResponse(response, session)
-}
-
 func rpcPing() (string, error) {
 	return "pong", nil
 }
@@ -315,108 +255,6 @@ func rpcSetSSHKeyState(sshKey string) error {
 	return nil
 }
 
-func callRPCHandler(handler RPCHandler, params map[string]interface{}) (interface{}, error) {
-	handlerValue := reflect.ValueOf(handler.Func)
-	handlerType := handlerValue.Type()
-
-	if handlerType.Kind() != reflect.Func {
-		return nil, errors.New("handler is not a function")
-	}
-
-	numParams := handlerType.NumIn()
-	args := make([]reflect.Value, numParams)
-	// Get the parameter names from the RPCHandler
-	paramNames := handler.Params
-
-	if len(paramNames) != numParams {
-		return nil, errors.New("mismatch between handler parameters and defined parameter names")
-	}
-
-	for i := 0; i < numParams; i++ {
-		paramType := handlerType.In(i)
-		paramName := paramNames[i]
-		paramValue, ok := params[paramName]
-		if !ok {
-			return nil, errors.New("missing parameter: " + paramName)
-		}
-
-		convertedValue := reflect.ValueOf(paramValue)
-		if !convertedValue.Type().ConvertibleTo(paramType) {
-			if paramType.Kind() == reflect.Slice && (convertedValue.Kind() == reflect.Slice || convertedValue.Kind() == reflect.Array) {
-				newSlice := reflect.MakeSlice(paramType, convertedValue.Len(), convertedValue.Len())
-				for j := 0; j < convertedValue.Len(); j++ {
-					elemValue := convertedValue.Index(j)
-					if elemValue.Kind() == reflect.Interface {
-						elemValue = elemValue.Elem()
-					}
-					if !elemValue.Type().ConvertibleTo(paramType.Elem()) {
-						// Handle float64 to uint8 conversion
-						if elemValue.Kind() == reflect.Float64 && paramType.Elem().Kind() == reflect.Uint8 {
-							intValue := int(elemValue.Float())
-							if intValue < 0 || intValue > 255 {
-								return nil, fmt.Errorf("value out of range for uint8: %v", intValue)
-							}
-							newSlice.Index(j).SetUint(uint64(intValue))
-						} else {
-							fromType := elemValue.Type()
-							toType := paramType.Elem()
-							return nil, fmt.Errorf("invalid element type in slice for parameter %s: from %v to %v", paramName, fromType, toType)
-						}
-					} else {
-						newSlice.Index(j).Set(elemValue.Convert(paramType.Elem()))
-					}
-				}
-				args[i] = newSlice
-			} else if paramType.Kind() == reflect.Struct && convertedValue.Kind() == reflect.Map {
-				jsonData, err := json.Marshal(convertedValue.Interface())
-				if err != nil {
-					return nil, fmt.Errorf("failed to marshal map to JSON: %v", err)
-				}
-
-				newStruct := reflect.New(paramType).Interface()
-				if err := json.Unmarshal(jsonData, newStruct); err != nil {
-					return nil, fmt.Errorf("failed to unmarshal JSON into struct: %v", err)
-				}
-				args[i] = reflect.ValueOf(newStruct).Elem()
-			} else {
-				return nil, fmt.Errorf("invalid parameter type for: %s", paramName)
-			}
-		} else {
-			args[i] = convertedValue.Convert(paramType)
-		}
-	}
-
-	results := handlerValue.Call(args)
-
-	if len(results) == 0 {
-		return nil, nil
-	}
-
-	if len(results) == 1 {
-		if results[0].Type().Implements(reflect.TypeOf((*error)(nil)).Elem()) {
-			if !results[0].IsNil() {
-				return nil, results[0].Interface().(error)
-			}
-			return nil, nil
-		}
-		return results[0].Interface(), nil
-	}
-
-	if len(results) == 2 && results[1].Type().Implements(reflect.TypeOf((*error)(nil)).Elem()) {
-		if !results[1].IsNil() {
-			return nil, results[1].Interface().(error)
-		}
-		return results[0].Interface(), nil
-	}
-
-	return nil, errors.New("unexpected return values from handler")
-}
-
-type RPCHandler struct {
-	Func   interface{}
-	Params []string
-}
-
 func rpcSetMassStorageMode(mode string) (string, error) {
 	log.Printf("[jsonrpc.go:rpcSetMassStorageMode] Setting mass storage mode to: %s", mode)
 	var cdrom bool
@@ -508,7 +346,7 @@ func rpcResetConfig() error {
 }
 
 // TODO: replace this crap with code generator
-var rpcHandlers = map[string]RPCHandler{
+var rpcHandlers = map[string]*jsonrpc.RPCHandler{
 	"ping":                   {Func: rpcPing},
 	"getDeviceID":            {Func: rpcGetDeviceID},
 	"deregisterDevice":       {Func: rpcDeregisterDevice},

ui/package.json

@@ -10,6 +10,7 @@
     "dev": "vite dev --mode=development",
     "build": "npm run build:prod",
     "build:device": "tsc && vite build --mode=device --emptyOutDir",
+    "dev:device": "vite dev --mode=device",
     "build:prod": "tsc && vite build --mode=production",
     "lint": "eslint . --ext ts,tsx --report-unused-disable-directives --max-warnings 0"
   },

ui/src/components/ActionBar.tsx

@@ -4,6 +4,7 @@ import {
   useMountMediaStore,
   useUiStore,
   useSettingsStore,
+  useVideoStore,
 } from "@/hooks/stores";
 import { MdOutlineContentPasteGo } from "react-icons/md";
 import Container from "@components/Container";
@@ -33,6 +34,7 @@ export default function Actionbar({
     state => state.remoteVirtualMediaState,
   );
   const developerMode = useSettingsStore(state => state.developerMode);
+  const hdmiState = useVideoStore(state => state.hdmiState);
 
   // This is the only way to get a reliable state change for the popover
   // at time of writing this there is no mount, or unmount event for the popover
@@ -247,6 +249,7 @@ export default function Actionbar({
               size="XS"
               theme="light"
               text="Fullscreen"
+              disabled={hdmiState !== 'ready'}
               LeadingIcon={LuMaximize}
               onClick={() => requestFullscreen()}
             />

ui/src/components/WebRTCVideo.tsx

@@ -30,6 +30,8 @@ export default function WebRTCVideo() {
   const {
     setClientSize: setVideoClientSize,
     setSize: setVideoSize,
+    width: videoWidth,
+    height: videoHeight,
     clientWidth: videoClientWidth,
     clientHeight: videoClientHeight,
   } = useVideoStore();
@@ -102,20 +104,43 @@ export default function WebRTCVideo() {
   const mouseMoveHandler = useCallback(
     (e: MouseEvent) => {
       if (!videoClientWidth || !videoClientHeight) return;
-      const { buttons } = e;
+      // Get the aspect ratios of the video element and the video stream
+      const videoElementAspectRatio = videoClientWidth / videoClientHeight;
+      const videoStreamAspectRatio = videoWidth / videoHeight;
 
-      // Clamp mouse position within the video boundaries
-      const currMouseX = Math.min(Math.max(1, e.offsetX), videoClientWidth);
-      const currMouseY = Math.min(Math.max(1, e.offsetY), videoClientHeight);
+      // Calculate the effective video display area
+      let effectiveWidth = videoClientWidth;
+      let effectiveHeight = videoClientHeight;
+      let offsetX = 0;
+      let offsetY = 0;
 
-      // Normalize mouse position to 0-32767 range (HID absolute coordinate system)
-      const x = Math.round((currMouseX / videoClientWidth) * 32767);
-      const y = Math.round((currMouseY / videoClientHeight) * 32767);
+      if (videoElementAspectRatio > videoStreamAspectRatio) {
+        // Pillarboxing: black bars on the left and right
+        effectiveWidth = videoClientHeight * videoStreamAspectRatio;
+        offsetX = (videoClientWidth - effectiveWidth) / 2;
+      } else if (videoElementAspectRatio < videoStreamAspectRatio) {
+        // Letterboxing: black bars on the top and bottom
+        effectiveHeight = videoClientWidth / videoStreamAspectRatio;
+        offsetY = (videoClientHeight - effectiveHeight) / 2;
+      }
+
+      // Clamp mouse position within the effective video boundaries
+      const clampedX = Math.min(Math.max(offsetX, e.offsetX), offsetX + effectiveWidth);
+      const clampedY = Math.min(Math.max(offsetY, e.offsetY), offsetY + effectiveHeight);
+
+      // Map clamped mouse position to the video stream's coordinate system
+      const relativeX = (clampedX - offsetX) / effectiveWidth;
+      const relativeY = (clampedY - offsetY) / effectiveHeight;
+
+      // Convert to HID absolute coordinate system (0-32767 range)
+      const x = Math.round(relativeX * 32767);
+      const y = Math.round(relativeY * 32767);
 
       // Send mouse movement
+      const { buttons } = e;
       sendMouseMovement(x, y, buttons);
     },
-    [sendMouseMovement, videoClientHeight, videoClientWidth],
+    [sendMouseMovement, videoClientHeight, videoClientWidth, videoWidth, videoHeight],
   );
 
   const mouseWheelHandler = useCallback(

ui/vite.config.ts

@@ -2,13 +2,31 @@ import { defineConfig } from "vite";
 import react from "@vitejs/plugin-react-swc";
 import tsconfigPaths from "vite-tsconfig-paths";
 
-export default defineConfig(({ mode }) => {
+declare const process: {
+  env: {
+    JETKVM_PROXY_URL: string;
+  };
+};
+
+export default defineConfig(({ mode, command }) => {
   const isCloud = mode === "production";
   const onDevice = mode === "device";
+  const { JETKVM_PROXY_URL } = process.env;
+
   return {
     plugins: [tsconfigPaths(), react()],
     build: { outDir: isCloud ? "dist" : "../static" },
-    server: { host: "0.0.0.0" },
-    base: onDevice ? "/static" : "/",
+    server: {
+      host: "0.0.0.0",
+      proxy: JETKVM_PROXY_URL ? {
+        '/me': JETKVM_PROXY_URL,
+        '/device': JETKVM_PROXY_URL,
+        '/webrtc': JETKVM_PROXY_URL,
+        '/auth': JETKVM_PROXY_URL,
+        '/storage': JETKVM_PROXY_URL,
+        '/cloud': JETKVM_PROXY_URL,
+      } : undefined
+    },
+    base: onDevice && command === 'build' ? "/static" : "/",
   };
 });

web.go

@@ -17,8 +17,10 @@ import (
 var staticFiles embed.FS
 
 type WebRTCSessionRequest struct {
-	Sd         string `json:"sd"`
-	OidcGoogle string `json:"OidcGoogle,omitempty"`
+	Sd         string   `json:"sd"`
+	OidcGoogle string   `json:"OidcGoogle,omitempty"`
+	IP         string   `json:"ip,omitempty"`
+	ICEServers []string `json:"iceServers,omitempty"`
 }
 
 type SetPasswordRequest struct {
@@ -116,7 +118,7 @@ func handleWebRTCSession(c *gin.Context) {
 		return
 	}
 
-	session, err := newSession()
+	session, err := newSession(SessionConfig{})
 	if err != nil {
 		c.JSON(http.StatusInternalServerError, gin.H{"error": err})
 		return

webrtc.go

@@ -4,6 +4,7 @@ import (
 	"encoding/base64"
 	"encoding/json"
 	"fmt"
+	"net"
 	"strings"
 
 	"github.com/pion/webrtc/v4"
@@ -19,6 +20,12 @@ type Session struct {
 	shouldUmountVirtualMedia bool
 }
 
+type SessionConfig struct {
+	ICEServers []string
+	LocalIP    string
+	IsCloud    bool
+}
+
 func (s *Session) ExchangeOffer(offerStr string) (string, error) {
 	b, err := base64.StdEncoding.DecodeString(offerStr)
 	if err != nil {
@@ -61,9 +68,29 @@ func (s *Session) ExchangeOffer(offerStr string) (string, error) {
 	return base64.StdEncoding.EncodeToString(localDescription), nil
 }
 
-func newSession() (*Session, error) {
-	peerConnection, err := webrtc.NewPeerConnection(webrtc.Configuration{
-		ICEServers: []webrtc.ICEServer{{}},
+func newSession(config SessionConfig) (*Session, error) {
+	webrtcSettingEngine := webrtc.SettingEngine{}
+	iceServer := webrtc.ICEServer{}
+
+	if config.IsCloud {
+		if config.ICEServers == nil {
+			fmt.Printf("ICE Servers not provided by cloud")
+		} else {
+			iceServer.URLs = config.ICEServers
+			fmt.Printf("Using ICE Servers provided by cloud: %v\n", iceServer.URLs)
+		}
+
+		if config.LocalIP == "" || net.ParseIP(config.LocalIP) == nil {
+			fmt.Printf("Local IP address %v not provided or invalid, won't set NAT1To1IPs\n", config.LocalIP)
+		} else {
+			webrtcSettingEngine.SetNAT1To1IPs([]string{config.LocalIP}, webrtc.ICECandidateTypeSrflx)
+			fmt.Printf("Setting NAT1To1IPs to %s\n", config.LocalIP)
+		}
+	}
+
+	api := webrtc.NewAPI(webrtc.WithSettingEngine(webrtcSettingEngine))
+	peerConnection, err := api.NewPeerConnection(webrtc.Configuration{
+		ICEServers: []webrtc.ICEServer{iceServer},
 	})
 	if err != nil {
 		return nil, err
@@ -75,8 +102,9 @@ func newSession() (*Session, error) {
 		switch d.Label() {
 		case "rpc":
 			session.RPCChannel = d
+			rpcServer := NewDataChannelJsonRpcRouter(d)
 			d.OnMessage(func(msg webrtc.DataChannelMessage) {
-				go onRPCMessage(msg, session)
+				go rpcServer.HandleMessage(msg.Data)
 			})
 			triggerOTAStateUpdate()
 			triggerVideoStateUpdate()