| import uvicorn |
| from fastapi import FastAPI, WebSocket, WebSocketDisconnect |
| from fastapi.responses import HTMLResponse |
| from typing import List |
|
|
| app = FastAPI() |
|
|
| |
| class ConnectionManager: |
| def __init__(self): |
| |
| self.macos_connections: List[WebSocket] = [] |
| self.linux_connections: List[WebSocket] = [] |
|
|
| async def connect_macos(self, websocket: WebSocket): |
| await websocket.accept() |
| self.macos_connections.append(websocket) |
| print(f"Client connected to macOS. Total: {len(self.macos_connections)}") |
|
|
| async def connect_linux(self, websocket: WebSocket): |
| await websocket.accept() |
| self.linux_connections.append(websocket) |
| print(f"Client connected to Linux. Total: {len(self.linux_connections)}") |
|
|
| def disconnect_macos(self, websocket: WebSocket): |
| if websocket in self.macos_connections: |
| self.macos_connections.remove(websocket) |
| print("Client disconnected from macOS") |
|
|
| def disconnect_linux(self, websocket: WebSocket): |
| if websocket in self.linux_connections: |
| self.linux_connections.remove(websocket) |
| print("Client disconnected from Linux") |
|
|
| async def forward_to_linux_clients(self, message: str): |
| |
| |
| for connection in self.linux_connections[:]: |
| try: |
| await connection.send_text(message) |
| except Exception as e: |
| print(f"Error sending to linux client: {e}") |
|
|
| async def forward_to_macos_clients(self, message: str): |
| |
| for connection in self.macos_connections[:]: |
| try: |
| await connection.send_text(message) |
| except Exception as e: |
| print(f"Error sending to macos client: {e}") |
|
|
| manager = ConnectionManager() |
|
|
| |
|
|
| @app.websocket("/macos") |
| async def websocket_macos(websocket: WebSocket): |
| await manager.connect_macos(websocket) |
| try: |
| while True: |
| |
| data = await websocket.receive_text() |
| print(f"Received from macOS: {data}") |
| |
| await manager.forward_to_linux_clients(data) |
| except WebSocketDisconnect: |
| manager.disconnect_macos(websocket) |
| except Exception as e: |
| print(f"Error in macOS socket: {e}") |
| manager.disconnect_macos(websocket) |
|
|
| @app.websocket("/linux") |
| async def websocket_linux(websocket: WebSocket): |
| await manager.connect_linux(websocket) |
| try: |
| while True: |
| |
| data = await websocket.receive_text() |
| print(f"Received from Linux: {data}") |
| |
| await manager.forward_to_macos_clients(data) |
| except WebSocketDisconnect: |
| manager.disconnect_linux(websocket) |
| except Exception as e: |
| print(f"Error in Linux socket: {e}") |
| manager.disconnect_linux(websocket) |
|
|
| |
| @app.get("/") |
| async def get(): |
| return HTMLResponse(""" |
| <!DOCTYPE html> |
| <html> |
| <head> |
| <title>Bridge Tester</title> |
| </head> |
| <body> |
| <h1>OS Bridge Tester</h1> |
| <div style="display:flex; gap: 20px;"> |
| <div style="border:1px solid #ccc; padding:10px;"> |
| <h2>Pretend to be macOS</h2> |
| <button onclick="connect('macos')">Connect to /macos</button> |
| <input type="text" id="msgMac" placeholder="Message to Linux" /> |
| <button onclick="send('macos')">Send</button> |
| <ul id="logMac"></ul> |
| </div> |
| |
| <div style="border:1px solid #ccc; padding:10px;"> |
| <h2>Pretend to be Linux</h2> |
| <button onclick="connect('linux')">Connect to /linux</button> |
| <input type="text" id="msgLin" placeholder="Message to macOS" /> |
| <button onclick="send('linux')">Send</button> |
| <ul id="logLin"></ul> |
| </div> |
| </div> |
| |
| <script> |
| var wsMac = null; |
| var wsLin = null; |
| |
| function connect(os) { |
| var ws = new WebSocket(((window.location.protocol === "https:") ? "wss://" : "ws://") + window.location.host + "/" + os); |
| |
| ws.onopen = function() { appendLog(os, "Connected!"); }; |
| ws.onmessage = function(event) { appendLog(os, "Received: " + event.data); }; |
| |
| if(os === 'macos') wsMac = ws; |
| else wsLin = ws; |
| } |
| |
| function send(os) { |
| var input = document.getElementById(os === 'macos' ? 'msgMac' : 'msgLin'); |
| var ws = os === 'macos' ? wsMac : wsLin; |
| if(ws) ws.send(input.value); |
| input.value = ''; |
| } |
| |
| function appendLog(os, msg) { |
| var ul = document.getElementById(os === 'macos' ? 'logMac' : 'logLin'); |
| var li = document.createElement("li"); |
| li.appendChild(document.createTextNode(msg)); |
| ul.appendChild(li); |
| } |
| </script> |
| </body> |
| </html> |
| """) |
|
|
| |
| if __name__ == "__main__": |
| uvicorn.run(app, host="0.0.0.0", port=7860) |
